Repository: kumahq/kuma
Pull request #1007 (2020-09-04 11:32:54)

fix(kuma-cp) signing token in multizone
Dataplane Token is an identity that we give to Dataplanes and verify it on the CP on Universal deployments. On K8S we use service account tokens. Dataplane Token is a JWT token signed by Signing Token that is generated when CP starts.

With multizone setup when Global is on K8S and you connect Remote Universal, KDS syncs all the secrets from Global to Remote. Since Global is a centralized single source of truth for everything other than Dataplanes, if Secret is not available on Global, it will be removed from the Remote.  What happened is that we were generating token on Remote but not on Global so it was removed from Remote. This PR fixes this behavior.

This time we generate Signing Key with a multizone setup only on Global but in all environments. Even if Global is on K8S, we don't know when the user will connect Universal Remote and there will be a need to generate DP tokens. We don't generate token on the Remote anymore. From now, to deploy app on Remote Universal, you need to first connect to Global to sync Signing Key, otherwise you won't be able to verify Token that DP presents.

Additionally, you should be able to generate Tokens on both Global and Remote. The Global-Remote design assumes that you can spin up and down Dataplanes without Global. There might be a case that CI that deploys the app generates token beforehand therefore there should be a way to generate it also on the Remote.

I added a simple E2E test to check that if I have Global on K8S and Remote Universal I can deploy 2 apps.

### Documentation

- TODO with DP improvements.



Patch:
diff --git a/pkg/admin-server/server.go b/pkg/admin-server/server.go
index 2299d4a62..4d6944d8d 100644
--- a/pkg/admin-server/server.go
+++ b/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
 }
diff --git a/pkg/core/bootstrap/bootstrap.go b/pkg/core/bootstrap/bootstrap.go
index 8da3e4d92..77c79b483 100644
--- a/pkg/core/bootstrap/bootstrap.go
+++ b/pkg/core/bootstrap/bootstrap.go
@@ -131,15 +131,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 {
diff --git a/pkg/tokens/builtin/components.go b/pkg/tokens/builtin/components.go
index e8e433c67..5286b60e1 100644
--- a/pkg/tokens/builtin/components.go
+++ b/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
 }
diff --git a/pkg/tokens/builtin/issuer/issuer.go b/pkg/tokens/builtin/issuer/issuer.go
index 91d347ad3..5bc6d3f72 100644
--- a/pkg/tokens/builtin/issuer/issuer.go
+++ b/pkg/tokens/builtin/issuer/issuer.go
@@ -29,17 +29,35 @@ 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(identity DataplaneIdentity) (auth.Credential, error) {
+	signingKey, err := i.signingKey()
+	if err != nil {
+		return "", err
+	}
+
 	tags := map[string][]string{}
 	for tagName := range identity.Tags {
 		tags[tagName] = identity.Tags.Values(tagName)
@@ -53,7 +71,7 @@ func (i *jwtTokenIssuer) Generate(identity DataplaneIdentity) (auth.Credential,
 	}
 
 	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")
 	}
@@ -61,10 +79,15 @@ func (i *jwtTokenIssuer) Generate(identity DataplaneIdentity) (auth.Credential,
 }
 
 func (i *jwtTokenIssuer) Validate(credential auth.Credential) (DataplaneIdentity, error) {
+	signingKey, err := i.signingKey()
+	if err != nil {
+		return DataplaneIdentity{}, 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 DataplaneIdentity{}, errors.Wrap(err, "could not parse token")
diff --git a/pkg/tokens/builtin/issuer/signing_key.go b/pkg/tokens/builtin/issuer/signing_key.go
index 09d4898bb..b4938a704 100644
--- a/pkg/tokens/builtin/issuer/signing_key.go
+++ b/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 patch:
diff --git a/pkg/sds/auth/universal/auth_test.go b/pkg/sds/auth/universal/auth_test.go
index 45ab91cd7..3a4153ad5 100644
--- a/pkg/sds/auth/universal/auth_test.go
+++ b/pkg/sds/auth/universal/auth_test.go
@@ -22,7 +22,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,7 +151,7 @@ var _ = Describe("Authentication flow", func() {
 					},
 				},
 			},
-			err: `dataplane contains tag "kuma.io/service" with value "web" which is not allowed with this token. Allowed values in token are ["backend"]`,
+			err: `which is not allowed with this token. Allowed values in token are ["backend"]`,
 		}),
 		Entry("on token with tag that is absent in dataplane", testCase{
 			id: builtin_issuer.DataplaneIdentity{
@@ -208,4 +210,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(builtin_issuer.DataplaneIdentity{})
+
+		// 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."))
+	})
 })
diff --git a/test/e2e/kuma_deploy_hybrid_kube_global_test.go b/test/e2e/kuma_deploy_hybrid_kube_global_test.go
new file mode 100644
index 000000000..9e33ce937
--- /dev/null
+++ b/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")
+			})
+	})
+})
diff --git a/test/e2e/kuma_deploy_hybrid_test.go b/test/e2e/kuma_deploy_hybrid_test.go
index a38e46893..7da125d15 100644
--- a/test/e2e/kuma_deploy_hybrid_test.go
+++ b/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()
diff --git a/test/e2e/kuma_deploy_universal_test.go b/test/e2e/kuma_deploy_universal_test.go
index 6c0a4ee12..fbe3b4144 100644
--- a/test/e2e/kuma_deploy_universal_test.go
+++ b/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()
diff --git a/test/e2e/tracing_universal_test.go b/test/e2e/tracing_universal_test.go
index 98a02201b..c66717ffd 100644
--- a/test/e2e/tracing_universal_test.go
+++ b/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() {
diff --git a/test/framework/constants.go b/test/framework/constants.go
index c153def46..51032772c 100644
--- a/test/framework/constants.go
+++ b/test/framework/constants.go
@@ -63,5 +63,6 @@ const (
 	helmChartPath = "../../deployments/charts/kuma"
 
 	kumaCPAPIPort        = 5681
+	kumaCPAdminPort      = 5679
 	kumaCPAPIPortFwdBase = 32000 + kumaCPAPIPort
 )
diff --git a/test/framework/interface.go b/test/framework/interface.go
index a9acf258c..a8843c640 100644
--- a/test/framework/interface.go
+++ b/test/framework/interface.go
@@ -97,7 +97,7 @@ type Cluster interface {
 	GetKubectlOptions(namespace ...string) *k8s.KubectlOptions
 	CreateNamespace(namespace string) error
 	DeleteNamespace(namespace string) error
-	DeployApp(namespace, appname string) error
+	DeployApp(namespace, appname, token string) error
 	DeleteApp(namespace, appname string) error
 	Exec(namespace, podName, containerName string, cmd ...string) (string, string, error)
 	ExecWithRetries(namespace, podName, containerName string, cmd ...string) (string, string, error)
@@ -112,4 +112,5 @@ type ControlPlane interface {
 	GetKDSServerAddress() string
 	GetIngressAddress() string
 	GetGlobaStatusAPI() string
+	GenerateDpToken(appname string) (string, error)
 }
diff --git a/test/framework/k8s_cluster.go b/test/framework/k8s_cluster.go
index 1121696f0..285757e47 100644
--- a/test/framework/k8s_cluster.go
+++ b/test/framework/k8s_cluster.go
@@ -565,7 +565,7 @@ func (c *K8sCluster) DeleteNamespace(namespace string) error {
 	return nil
 }
 
-func (c *K8sCluster) DeployApp(namespace, appname string) error {
+func (c *K8sCluster) DeployApp(namespace, appname, token string) error {
 	retry.DoWithRetry(c.GetTesting(), "apply "+appname+" svc", DefaultRetries, DefaultTimeout,
 		func() (string, error) {
 			err := k8s.KubectlApplyE(c.GetTesting(),
diff --git a/test/framework/k8s_clusters.go b/test/framework/k8s_clusters.go
index fd1e40c31..31afd8975 100644
--- a/test/framework/k8s_clusters.go
+++ b/test/framework/k8s_clusters.go
@@ -158,9 +158,9 @@ func (c *K8sClusters) GetKumactlOptions() *KumactlOptions {
 	return nil
 }
 
-func (cs *K8sClusters) DeployApp(namespace, appname string) error {
+func (cs *K8sClusters) DeployApp(namespace, appname, token string) error {
 	for name, c := range cs.clusters {
-		if err := c.DeployApp(namespace, appname); err != nil {
+		if err := c.DeployApp(namespace, appname, token); err != nil {
 			return errors.Wrapf(err, "Labeling Namespace %s on %s failed: %v", namespace, name, err)
 		}
 	}
diff --git a/test/framework/k8s_controlplane.go b/test/framework/k8s_controlplane.go
index 2a6e13880..1fbc6e8b2 100644
--- a/test/framework/k8s_controlplane.go
+++ b/test/framework/k8s_controlplane.go
@@ -20,9 +20,10 @@ import (
 )
 
 type PortFwd struct {
-	lowFwdPort   uint32
-	hiFwdPort    uint32
-	localAPIPort uint32
+	lowFwdPort     uint32
+	hiFwdPort      uint32
+	localAPIPort   uint32
+	localAdminPort uint32
 }
 
 type K8sControlPlane struct {
@@ -73,9 +74,6 @@ func (c *K8sControlPlane) GetKubectlOptions(namespace ...string) *k8s.KubectlOpt
 }
 
 func (c *K8sControlPlane) PortForwardKumaCP() error {
-	var apiPort uint32
-	var err error
-
 	kumacpPods := c.GetKumaCPPods()
 	if len(kumacpPods) != 1 {
 		return errors.Errorf("Kuma CP pods: %d", len(kumacpPods))
@@ -84,7 +82,7 @@ func (c *K8sControlPlane) PortForwardKumaCP() error {
 	kumacpPodName := kumacpPods[0].Name
 
 	// API
-	apiPort, err = util_net.PickTCPPort("", c.portFwd.lowFwdPort+1, c.portFwd.hiFwdPort)
+	apiPort, err := util_net.PickTCPPort("", c.portFwd.lowFwdPort+1, c.portFwd.hiFwdPort)
 	if err != nil {
 		return errors.Errorf("No free port found in range:  %d - %d", c.portFwd.lowFwdPort, c.portFwd.hiFwdPort)
 	}
@@ -92,6 +90,15 @@ func (c *K8sControlPlane) PortForwardKumaCP() error {
 	c.cluster.PortForwardPod(kumaNamespace, kumacpPodName, apiPort, kumaCPAPIPort)
 	c.portFwd.localAPIPort = apiPort
 
+	// Admin
+	adminPort, err := util_net.PickTCPPort("", c.portFwd.lowFwdPort+2, c.portFwd.hiFwdPort)
+	if err != nil {
+		return errors.Errorf("No free port found in range:  %d - %d", c.portFwd.lowFwdPort, c.portFwd.hiFwdPort)
+	}
+
+	c.cluster.PortForwardPod(kumaNamespace, kumacpPodName, adminPort, kumaCPAdminPort)
+	c.portFwd.localAdminPort = adminPort
+
 	return nil
 }
 
@@ -237,3 +244,17 @@ func (c *K8sControlPlane) GetIngressAddress() string {
 func (c *K8sControlPlane) GetGlobaStatusAPI() string {
 	return "http://localhost:" + strconv.FormatUint(uint64(c.portFwd.localAPIPort), 10) + "/status/zones"
 }
+
+func (c *K8sControlPlane) GenerateDpToken(service string) (string, error) {
+	return http_helper.HTTPDoWithRetryE(
+		c.t,
+		"POST",
+		fmt.Sprintf("http://localhost:%d/tokens", c.portFwd.localAdminPort),
+		[]byte(fmt.Sprintf(`{"mesh": "default", "tags": {"kuma.io/service": ["%s"]}}`, service)),
+		map[string]string{"content-type": "application/json"},
+		200,
+		DefaultRetries,
+		DefaultTimeout,
+		&tls.Config{},
+	)
+}
diff --git a/test/framework/setup.go b/test/framework/setup.go
index c0729f9ea..8921be4de 100644
--- a/test/framework/setup.go
+++ b/test/framework/setup.go
@@ -103,9 +103,9 @@ func EchoServerK8s() InstallFunc {
 	)
 }
 
-func EchoServerUniversal() InstallFunc {
+func EchoServerUniversal(token string) InstallFunc {
 	return func(cluster Cluster) error {
-		return cluster.DeployApp("", AppModeEchoServer)
+		return cluster.DeployApp("", AppModeEchoServer, token)
 	}
 }
 
@@ -167,9 +167,9 @@ func DemoClientK8s() InstallFunc {
 	)
 }
 
-func DemoClientUniversal() InstallFunc {
+func DemoClientUniversal(token string) InstallFunc {
 	return func(cluster Cluster) error {
-		return cluster.DeployApp("", AppModeDemoClient)
+		return cluster.DeployApp("", AppModeDemoClient, token)
 	}
 }
 
diff --git a/test/framework/universal_cluster.go b/test/framework/universal_cluster.go
index 193336aa6..874327e0a 100644
--- a/test/framework/universal_cluster.go
+++ b/test/framework/universal_cluster.go
@@ -93,7 +93,7 @@ func (c *UniversalCluster) DeployKuma(mode string, fs ...DeployOptionsFunc) erro
 	switch mode {
 	case core.Remote:
 		dpyaml := fmt.Sprintf(IngressDataplane, kdsPort)
-		err = c.CreateDP(app, "ingress", app.ip, dpyaml)
+		err = c.CreateDP(app, "ingress", app.ip, dpyaml, "XYZ") // todo token is static now, Ingress does not access SDS so the token does not matter for now
 		if err != nil {
 			return err
 		}
@@ -142,25 +142,13 @@ func (c *UniversalCluster) DeleteNamespace(namespace string) error {
 	return nil
 }
 
-func (c *UniversalCluster) CreateDP(app *UniversalApp, appname, ip, dpyaml string) error {
-	// generate the token on the CP node
-	sshApp := NewSshApp(c.verbose, c.apps[AppModeCP].ports["22"], []string{}, []string{"curl",
-		"-H", "\"Content-Type: application/json\"",
-		"--data", "'{\"name\": \"dp-" + appname + "\", \"mesh\": \"default\"}'",
-		"http://localhost:5679/tokens"})
-	if err := sshApp.Run(); err != nil {
-		return err
-	}
-
-	token := sshApp.Out()
-
+func (c *UniversalCluster) CreateDP(app *UniversalApp, appname, ip, dpyaml, token string) error {
 	cpAddress := "http://" + c.apps[AppModeCP].ip + ":5681"
 	app.CreateDP(token, cpAddress, appname, ip, dpyaml)
-
 	return app.dpApp.Start()
 }
 
-func (c *UniversalCluster) DeployApp(namespace, appname string) error {
+func (c *UniversalCluster) DeployApp(namespace, appname, token string) error {
 	var args []string
 	switch appname {
 	case AppModeEchoServer:
@@ -190,7 +178,7 @@ func (c *UniversalCluster) DeployApp(namespace, appname string) error {
 		dpyaml = fmt.Sprintf(DemoClientDataplane, "13000", "3000", "80", "8080")
 	}
 
-	err = c.CreateDP(app, appname, ip, dpyaml)
+	err = c.CreateDP(app, appname, ip, dpyaml, token)
 	if err != nil {
 		return err
 	}
diff --git a/test/framework/universal_clusters.go b/test/framework/universal_clusters.go
index b71988244..2ba0ca0a1 100644
--- a/test/framework/universal_clusters.go
+++ b/test/framework/universal_clusters.go
@@ -133,9 +133,9 @@ func (c *UniversalClusters) GetKumactlOptions() *KumactlOptions {
 	return nil
 }
 
-func (cs *UniversalClusters) DeployApp(namespace, appname string) error {
+func (cs *UniversalClusters) DeployApp(namespace, appname, token string) error {
 	for name, c := range cs.clusters {
-		if err := c.DeployApp(namespace, appname); err != nil {
+		if err := c.DeployApp(namespace, appname, token); err != nil {
 			return errors.Wrapf(err, "Labeling Namespace %s on %s failed: %v", namespace, name, err)
 		}
 	}
diff --git a/test/framework/universal_controlplane.go b/test/framework/universal_controlplane.go
index 7a59be13b..6ec2d8930 100644
--- a/test/framework/universal_controlplane.go
+++ b/test/framework/universal_controlplane.go
@@ -1,8 +1,11 @@
 package framework
 
 import (
+	"fmt"
 	"strconv"
 
+	"github.com/pkg/errors"
+
 	"github.com/kumahq/kuma/pkg/config/core"
 
 	"github.com/gruntwork-io/terratest/modules/testing"
@@ -52,3 +55,18 @@ func (c *UniversalControlPlane) GetIngressAddress() string {
 func (c *UniversalControlPlane) GetGlobaStatusAPI() string {
 	panic("not implemented")
 }
+
+func (c *UniversalControlPlane) GenerateDpToken(service string) (string, error) {
+	sshApp := NewSshApp(c.verbose, c.cluster.apps[AppModeCP].ports["22"], []string{}, []string{"curl",
+		"--fail", "--show-error",
+		"-H", "\"Content-Type: application/json\"",
+		"--data", fmt.Sprintf(`'{"mesh": "default", "tags": {"kuma.io/service":["%s"]}}'`, service),
+		"http://localhost:5679/tokens"})
+	if err := sshApp.Run(); err != nil {
+		return "", err
+	}
+	if sshApp.Err() != "" {
+		return "", errors.New(sshApp.Err())
+	}
+	return sshApp.Out(), nil
+}