Bootstrapper

cli/internal/cmd/verify.go

@@ -180,7 +180,7 @@ func verifyCompletion(cmd *cobra.Command, args []string, toComplete string) ([]s
 }
 
 type constellationVerifier struct {
-	dialer grpcDialer
+	dialer grpcInsecureDialer
 }
 
 // Verify retrieves an attestation statement from the Constellation and verifies it using the validator.
@@ -215,6 +215,6 @@ type verifyClient interface {
 	Verify(ctx context.Context, endpoint string, req *verifyproto.GetAttestationRequest, validator atls.Validator) error
 }
 
-type grpcDialer interface {
+type grpcInsecureDialer interface {
 	DialInsecure(ctx context.Context, endpoint string) (conn *grpc.ClientConn, err error)
 }

cli/internal/cmd/vpnconfig.go

@@ -1,9 +0,0 @@
-package cmd
-
-import wgquick "github.com/nmiculinic/wg-quick-go"
-
-type vpnHandler interface {
-	Create(coordinatorPubKey string, coordinatorPubIP string, clientPrivKey string, clientVPNIP string, mtu int) (*wgquick.Config, error)
-	Apply(*wgquick.Config) error
-	Marshal(*wgquick.Config) ([]byte, error)
-}

cli/internal/cmd/vpnconfig_test.go

@@ -1,25 +0,0 @@
-package cmd
-
-import wgquick "github.com/nmiculinic/wg-quick-go"
-
-type stubVPNHandler struct {
-	configured bool
-	marshalRes string
-
-	createErr  error
-	applyErr   error
-	marshalErr error
-}
-
-func (c *stubVPNHandler) Create(coordinatorPubKey string, coordinatorPubIP string, clientPrivKey string, clientVPNIP string, mtu int) (*wgquick.Config, error) {
-	return &wgquick.Config{}, c.createErr
-}
-
-func (c *stubVPNHandler) Apply(*wgquick.Config) error {
-	c.configured = true
-	return c.applyErr
-}
-
-func (c *stubVPNHandler) Marshal(*wgquick.Config) ([]byte, error) {
-	return []byte(c.marshalRes), c.marshalErr
-}

coordinator/cloudprovider/azure/metadata.go

@@ -242,11 +242,6 @@ func (m *Metadata) GetLoadBalancerIP(ctx context.Context) (string, error) {
 	return *resp.Properties.IPAddress, nil
 }
 
-// SetVPNIP stores the internally used VPN IP in cloud provider metadata (not required on azure).
-func (m *Metadata) SetVPNIP(ctx context.Context, vpnIP string) error {
-	return nil
-}
-
 // Supported is used to determine if metadata API is implemented for this cloud provider.
 func (m *Metadata) Supported() bool {
 	return true

coordinator/cloudprovider/azure/metadata_test.go

@@ -488,12 +488,6 @@ func TestGetLoadBalancerIP(t *testing.T) {
 	}
 }
 
-func TestSetVPNIP(t *testing.T) {
-	assert := assert.New(t)
-	metadata := Metadata{}
-	assert.NoError(metadata.SetVPNIP(context.Background(), "192.0.2.0"))
-}
-
 func TestMetadataSupported(t *testing.T) {
 	assert := assert.New(t)
 	metadata := Metadata{}

coordinator/cloudprovider/gcp/autoscaler.go

@@ -2,7 +2,6 @@ package gcp
 
 import (
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes/k8sapi/resources"
-	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	k8s "k8s.io/api/core/v1"
 )
 
@@ -15,7 +14,7 @@ func (a *Autoscaler) Name() string {
 }
 
 // Secrets returns a list of secrets to deploy together with the k8s cluster-autoscaler.
-func (a *Autoscaler) Secrets(instance metadata.InstanceMetadata, cloudServiceAccountURI string) (resources.Secrets, error) {
+func (a *Autoscaler) Secrets(instance, cloudServiceAccountURI string) (resources.Secrets, error) {
 	return resources.Secrets{}, nil
 }
 

coordinator/cloudprovider/gcp/autoscaler_test.go

@@ -3,7 +3,6 @@ package gcp
 import (
 	"testing"
 
-	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -12,7 +11,7 @@ func TestTrivialAutoscalerFunctions(t *testing.T) {
 	autoscaler := Autoscaler{}
 
 	assert.NotEmpty(autoscaler.Name())
-	assert.Empty(autoscaler.Secrets(metadata.InstanceMetadata{}, ""))
+	assert.Empty(autoscaler.Secrets("", ""))
 	assert.NotEmpty(autoscaler.Volumes())
 	assert.NotEmpty(autoscaler.VolumeMounts())
 	assert.NotEmpty(autoscaler.Env())

coordinator/cloudprovider/gcp/metadata.go

@@ -104,6 +104,11 @@ func (m *Metadata) GetSubnetworkCIDR(ctx context.Context) (string, error) {
 	return m.api.RetrieveSubnetworkAliasCIDR(ctx, project, zone, instanceName)
 }
 
+// SupportsLoadBalancer returns true if the cloud provider supports load balancers.
+func (m *Metadata) SupportsLoadBalancer() bool {
+	return true
+}
+
 // GetLoadBalancerIP returns the IP of the load balancer.
 func (m *Metadata) GetLoadBalancerIP(ctx context.Context) (string, error) {
 	project, err := m.api.RetrieveProjectID()
@@ -116,3 +121,8 @@ func (m *Metadata) GetLoadBalancerIP(ctx context.Context) (string, error) {
 	}
 	return m.api.RetrieveLoadBalancerIP(ctx, project, zone)
 }
+
+// Supported is used to determine if metadata API is implemented for this cloud provider.
+func (m *Metadata) Supported() bool {
+	return true
+}

coordinator/cloudprovider/qemu/autoscaler.go

@@ -2,7 +2,6 @@ package qemu
 
 import (
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes/k8sapi/resources"
-	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	k8s "k8s.io/api/core/v1"
 )
 
@@ -15,7 +14,7 @@ func (a Autoscaler) Name() string {
 }
 
 // Secrets returns a list of secrets to deploy together with the k8s cluster-autoscaler.
-func (a Autoscaler) Secrets(instance metadata.InstanceMetadata, cloudServiceAccountURI string) (resources.Secrets, error) {
+func (a Autoscaler) Secrets(providerID, cloudServiceAccountURI string) (resources.Secrets, error) {
 	return resources.Secrets{}, nil
 }
 

coordinator/cloudprovider/qemu/ccm.go

@@ -39,7 +39,7 @@ func (c CloudControllerManager) ConfigMaps(instance metadata.InstanceMetadata) (
 
 // Secrets returns a list of secrets to deploy together with the k8s cloud-controller-manager.
 // Reference: https://kubernetes.io/docs/concepts/configuration/secret/ .
-func (c CloudControllerManager) Secrets(ctx context.Context, instance metadata.InstanceMetadata, cloudServiceAccountURI string) (resources.Secrets, error) {
+func (c CloudControllerManager) Secrets(ctx context.Context, providerID, cloudServiceAccountURI string) (resources.Secrets, error) {
 	return resources.Secrets{}, nil
 }
 

coordinator/cloudprovider/qemu/metadata.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"net/url"
 
-	"github.com/edgelesssys/constellation/coordinator/role"
 	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 )
 
@@ -61,16 +60,6 @@ func (m Metadata) GetInstance(ctx context.Context, providerID string) (metadata.
 	return metadata.InstanceMetadata{}, errors.New("instance not found")
 }
 
-// SignalRole signals the constellation role via cloud provider metadata (if supported by the CSP and deployment type, otherwise does nothing).
-func (m Metadata) SignalRole(ctx context.Context, role role.Role) error {
-	return nil
-}
-
-// SetVPNIP stores the internally used VPN IP in cloud provider metadata (if supported and required for autoscaling by the CSP, otherwise does nothing).
-func (m Metadata) SetVPNIP(ctx context.Context, vpnIP string) error {
-	return nil
-}
-
 // SupportsLoadBalancer returns true if the cloud provider supports load balancers.
 func (m Metadata) SupportsLoadBalancer() bool {
 	return false

coordinator/cmd/coordinator/main.go

@@ -12,8 +12,7 @@ import (
 	azurecloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/azure"
 	gcpcloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/gcp"
 	qemucloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/qemu"
-	"github.com/edgelesssys/constellation/coordinator/config"
+	"github.com/edgelesssys/constellation/coordinator/internal/joinclient"
-	"github.com/edgelesssys/constellation/coordinator/core"
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes"
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes/k8sapi"
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes/k8sapi/kubectl"
@@ -34,12 +33,18 @@ import (
 const (
 	defaultIP   = "0.0.0.0"
 	defaultPort = "9000"
+	// ConstellationCSP is the Cloud Service Provider Constellation is running on.
+	constellationCSP = "CONSTEL_CSP"
 )
 
 func main() {
+	ctx := context.Background()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
 	var bindIP, bindPort string
-	var clusterInitJoiner ClusterInitJoiner
+	var clusterInitJoiner clusterInitJoiner
-	var coreMetadata core.ProviderMetadata
+	var metadataAPI joinclient.MetadataAPI
 	var cloudLogger logging.CloudLogger
 	cfg := zap.NewDevelopmentConfig()
 
@@ -62,7 +67,7 @@ func main() {
 	var openTPM vtpm.TPMOpenFunc
 	var fs afero.Fs
 
-	switch strings.ToLower(os.Getenv(config.ConstellationCSP)) {
+	switch strings.ToLower(os.Getenv(constellationCSP)) {
 	case "aws":
 		panic("AWS cloud provider currently unsupported")
 	case "gcp":
@@ -74,20 +79,20 @@ func main() {
 
 		issuer = gcp.NewIssuer()
 
-		gcpClient, err := gcpcloud.NewClient(context.Background())
+		gcpClient, err := gcpcloud.NewClient(ctx)
 		if err != nil {
 			log.Fatalf("failed to create GCP client: %v\n", err)
 		}
 		metadata := gcpcloud.New(gcpClient)
-		descr, err := metadata.Self(context.Background())
+		descr, err := metadata.Self(ctx)
 		if err != nil {
 			log.Fatal(err)
 		}
-		cloudLogger, err = gcpcloud.NewLogger(context.Background(), descr.ProviderID, "constellation-boot-log")
+		cloudLogger, err = gcpcloud.NewLogger(ctx, descr.ProviderID, "constellation-boot-log")
 		if err != nil {
 			log.Fatal(err)
 		}
-		coreMetadata = metadata
+		metadataAPI = metadata
 		pcrsJSON, err := json.Marshal(pcrs)
 		if err != nil {
 			log.Fatal(err)
@@ -108,15 +113,15 @@ func main() {
 
 		issuer = azure.NewIssuer()
 
-		metadata, err := azurecloud.NewMetadata(context.Background())
+		metadata, err := azurecloud.NewMetadata(ctx)
 		if err != nil {
 			log.Fatal(err)
 		}
-		cloudLogger, err = azurecloud.NewLogger(context.Background(), metadata)
+		cloudLogger, err = azurecloud.NewLogger(ctx, metadata)
 		if err != nil {
 			log.Fatal(err)
 		}
-		coreMetadata = metadata
+		metadataAPI = metadata
 		pcrsJSON, err := json.Marshal(pcrs)
 		if err != nil {
 			log.Fatal(err)
@@ -148,7 +153,7 @@ func main() {
 			"qemu", k8sapi.NewKubernetesUtil(), &k8sapi.CoreOSConfiguration{}, kubectl.New(), &qemucloud.CloudControllerManager{},
 			&qemucloud.CloudNodeManager{}, &qemucloud.Autoscaler{}, metadata, pcrsJSON,
 		)
-		coreMetadata = metadata
+		metadataAPI = metadata
 
 		bindIP = defaultIP
 		bindPort = defaultPort
@@ -156,8 +161,8 @@ func main() {
 		fs = afero.NewOsFs()
 	default:
 		issuer = atls.NewFakeIssuer(oid.Dummy{})
-		clusterInitJoiner = &core.ClusterFake{}
+		clusterInitJoiner = &clusterFake{}
-		coreMetadata = &core.ProviderMetadataFake{}
+		metadataAPI = &providerMetadataFake{}
 		cloudLogger = &logging.NopLogger{}
 		bindIP = defaultIP
 		bindPort = defaultPort
@@ -170,6 +175,6 @@ func main() {
 	fileHandler := file.NewHandler(fs)
 
 	run(issuer, openTPM, fileHandler, clusterInitJoiner,
-		coreMetadata, bindIP,
+		metadataAPI, bindIP,
 		bindPort, zapLoggerCore, cloudLogger, fs)
 }

coordinator/cmd/coordinator/run.go

@@ -2,23 +2,23 @@ package main
 
 import (
 	"net"
-	"sync"
 
-	"github.com/edgelesssys/constellation/coordinator/core"
 	"github.com/edgelesssys/constellation/coordinator/internal/initserver"
 	"github.com/edgelesssys/constellation/coordinator/internal/joinclient"
 	"github.com/edgelesssys/constellation/coordinator/internal/logging"
+	"github.com/edgelesssys/constellation/coordinator/internal/nodelock"
 	"github.com/edgelesssys/constellation/internal/attestation/vtpm"
 	"github.com/edgelesssys/constellation/internal/file"
 	"github.com/edgelesssys/constellation/internal/grpc/dialer"
+	"github.com/edgelesssys/constellation/internal/oid"
 	"github.com/spf13/afero"
 	"go.uber.org/zap"
 )
 
 var version = "0.0.0"
 
-func run(issuer core.QuoteIssuer, tpm vtpm.TPMOpenFunc, fileHandler file.Handler,
+func run(issuer quoteIssuer, tpm vtpm.TPMOpenFunc, fileHandler file.Handler,
-	kube ClusterInitJoiner, metadata core.ProviderMetadata,
+	kube clusterInitJoiner, metadata joinclient.MetadataAPI,
 	bindIP, bindPort string, logger *zap.Logger,
 	cloudLogger logging.CloudLogger, fs afero.Fs,
 ) {
@@ -40,7 +40,7 @@ func run(issuer core.QuoteIssuer, tpm vtpm.TPMOpenFunc, fileHandler file.Handler
 		return
 	}
 
-	nodeLock := &sync.Mutex{}
+	nodeLock := nodelock.New()
 	initServer := initserver.New(nodeLock, kube, logger)
 
 	dialer := dialer.New(issuer, nil, &net.Dialer{})
@@ -54,8 +54,14 @@ func run(issuer core.QuoteIssuer, tpm vtpm.TPMOpenFunc, fileHandler file.Handler
 	}
 }
 
-type ClusterInitJoiner interface {
+type clusterInitJoiner interface {
 	joinclient.ClusterJoiner
 	initserver.ClusterInitializer
 	StartKubelet() error
 }
+
+type quoteIssuer interface {
+	oid.Getter
+	// Issue issues a quote for remote attestation for a given message
+	Issue(userData []byte, nonce []byte) (quote []byte, err error)
+}

coordinator/cmd/coordinator/test.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+
+	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes"
+	"github.com/edgelesssys/constellation/coordinator/role"
+	attestationtypes "github.com/edgelesssys/constellation/internal/attestation/types"
+	"github.com/edgelesssys/constellation/internal/cloud/metadata"
+	kubeadm "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1beta3"
+)
+
+// ClusterFake behaves like a real cluster, but does not actually initialize or join Kubernetes.
+type clusterFake struct{}
+
+// InitCluster fakes bootstrapping a new cluster with the current node being the master, returning the arguments required to join the cluster.
+func (c *clusterFake) InitCluster(context.Context, []string, string, string, attestationtypes.ID, kubernetes.KMSConfig, map[string]string,
+) ([]byte, error) {
+	return []byte{}, nil
+}
+
+// JoinCluster will fake joining the current node to an existing cluster.
+func (c *clusterFake) JoinCluster(context.Context, *kubeadm.BootstrapTokenDiscovery, string, role.Role) error {
+	return nil
+}
+
+// StartKubelet starts the kubelet service.
+func (c *clusterFake) StartKubelet() error {
+	return nil
+}
+
+type providerMetadataFake struct{}
+
+func (f *providerMetadataFake) List(ctx context.Context) ([]metadata.InstanceMetadata, error) {
+	self, err := f.Self(ctx)
+	return []metadata.InstanceMetadata{self}, err
+}
+
+func (f *providerMetadataFake) Self(ctx context.Context) (metadata.InstanceMetadata, error) {
+	return metadata.InstanceMetadata{
+		Name:       "instanceName",
+		ProviderID: "fake://instance-id",
+		Role:       role.Unknown,
+		PrivateIPs: []string{"192.0.2.1"},
+	}, nil
+}
+
+func (f *providerMetadataFake) SignalRole(ctx context.Context, role role.Role) error {
+	return nil
+}
+
+func (f *providerMetadataFake) SetVPNIP(ctx context.Context, vpnIP string) error {
+	return nil
+}
+
+func (f *providerMetadataFake) Supported() bool {
+	return true
+}

coordinator/config/config.go

@@ -1,7 +0,0 @@
-package config
-
-// ConstellationCSP is the Cloud Service Provider Constellation is running on.
-const ConstellationCSP = "CONSTEL_CSP"
-
-// RNGLengthDefault is the number of bytes used for generating nonces.
-const RNGLengthDefault = 32

coordinator/internal/initserver/initserver.go

@@ -5,16 +5,16 @@ import (
 	"fmt"
 	"net"
 	"strings"
-	"sync"
 
-	"github.com/edgelesssys/constellation/coordinator/config"
 	"github.com/edgelesssys/constellation/coordinator/initproto"
 	"github.com/edgelesssys/constellation/coordinator/internal/diskencryption"
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes"
+	"github.com/edgelesssys/constellation/coordinator/internal/nodelock"
 	"github.com/edgelesssys/constellation/coordinator/nodestate"
 	"github.com/edgelesssys/constellation/coordinator/role"
 	"github.com/edgelesssys/constellation/coordinator/util"
 	attestationtypes "github.com/edgelesssys/constellation/internal/attestation/types"
+	"github.com/edgelesssys/constellation/internal/constants"
 	"github.com/edgelesssys/constellation/internal/file"
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
@@ -25,26 +25,29 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+// Server is the initialization server, which is started on each node.
+// The server handles initialization calls from the CLI and initializes the
+// Kubernetes cluster.
 type Server struct {
-	nodeLock *sync.Mutex
+	nodeLock    *nodelock.Lock
-
+	initializer ClusterInitializer
-	kube        ClusterInitializer
+	disk        encryptedDisk
-	disk        EncryptedDisk
 	fileHandler file.Handler
-	grpcServer  *grpc.Server
+	grpcServer  serveStopper
 
 	logger *zap.Logger
 
 	initproto.UnimplementedAPIServer
 }
 
-func New(nodeLock *sync.Mutex, kube ClusterInitializer, logger *zap.Logger) *Server {
+// New creates a new initialization server.
+func New(lock *nodelock.Lock, kube ClusterInitializer, logger *zap.Logger) *Server {
 	logger = logger.Named("initServer")
 	server := &Server{
-		nodeLock: nodeLock,
+		nodeLock:    lock,
-		disk:     diskencryption.New(),
+		disk:        diskencryption.New(),
-		kube:     kube,
+		initializer: kube,
-		logger:   logger,
+		logger:      logger,
 	}
 
 	grpcLogger := logger.Named("gRPC")
@@ -74,8 +77,14 @@ func (s *Server) Serve(ip, port string) error {
 	return s.grpcServer.Serve(lis)
 }
 
+// Init initializes the cluster.
 func (s *Server) Init(ctx context.Context, req *initproto.InitRequest) (*initproto.InitResponse, error) {
-	if ok := s.nodeLock.TryLock(); !ok {
+	if ok := s.nodeLock.TryLockOnce(); !ok {
+		// The join client seems to already have a connection to an
+		// existing join service. At this point, any further call to
+		// init does not make sense, so we just stop.
+		//
+		// The server stops itself after the current call is done.
 		go s.grpcServer.GracefulStop()
 		return nil, status.Error(codes.FailedPrecondition, "node is already being activated")
 	}
@@ -98,7 +107,7 @@ func (s *Server) Init(ctx context.Context, req *initproto.InitRequest) (*initpro
 		return nil, status.Errorf(codes.Internal, "persisting node state: %s", err)
 	}
 
-	kubeconfig, err := s.kube.InitCluster(ctx,
+	kubeconfig, err := s.initializer.InitCluster(ctx,
 		req.AutoscalingNodeGroups,
 		req.CloudServiceAccountUri,
 		req.KubernetesVersion,
@@ -145,13 +154,13 @@ func (s *Server) setupDisk(masterSecret []byte) error {
 }
 
 func (s *Server) deriveAttestationID(masterSecret []byte) (attestationtypes.ID, error) {
-	clusterID, err := util.GenerateRandomBytes(config.RNGLengthDefault)
+	clusterID, err := util.GenerateRandomBytes(constants.RNGLengthDefault)
 	if err != nil {
 		return attestationtypes.ID{}, err
 	}
 
 	// TODO: Choose a way to salt the key derivation
-	ownerID, err := util.DeriveKey(masterSecret, []byte("Constellation"), []byte("id"), config.RNGLengthDefault)
+	ownerID, err := util.DeriveKey(masterSecret, []byte("Constellation"), []byte("id"), constants.RNGLengthDefault)
 	if err != nil {
 		return attestationtypes.ID{}, err
 	}
@@ -167,20 +176,21 @@ func sshProtoKeysToMap(keys []*initproto.SSHUserKey) map[string]string {
 	return keyMap
 }
 
+// ClusterInitializer has the ability to initialize a cluster.
 type ClusterInitializer interface {
+	// InitCluster initializes a new Kubernetes cluster.
 	InitCluster(
 		ctx context.Context,
 		autoscalingNodeGroups []string,
 		cloudServiceAccountURI string,
-		kubernetesVersion string,
+		k8sVersion string,
 		id attestationtypes.ID,
-		config kubernetes.KMSConfig,
+		kmsConfig kubernetes.KMSConfig,
 		sshUserKeys map[string]string,
 	) ([]byte, error)
 }
 
-// EncryptedDisk manages the encrypted state disk.
+type encryptedDisk interface {
-type EncryptedDisk interface {
 	// Open prepares the underlying device for disk operations.
 	Open() error
 	// Close closes the underlying device.
@@ -190,3 +200,10 @@ type EncryptedDisk interface {
 	// UpdatePassphrase switches the initial random passphrase of the encrypted disk to a permanent passphrase.
 	UpdatePassphrase(passphrase string) error
 }
+
+type serveStopper interface {
+	// Serve starts the server.
+	Serve(lis net.Listener) error
+	// GracefulStop stops the server and blocks until all requests are done.
+	GracefulStop()
+}

coordinator/internal/initserver/initserver_test.go

@@ -0,0 +1,238 @@
+package initserver
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/edgelesssys/constellation/coordinator/initproto"
+	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes"
+	"github.com/edgelesssys/constellation/coordinator/internal/nodelock"
+	attestationtypes "github.com/edgelesssys/constellation/internal/attestation/types"
+	"github.com/edgelesssys/constellation/internal/file"
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest"
+)
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}
+
+func TestNew(t *testing.T) {
+	assert := assert.New(t)
+
+	server := New(nodelock.New(), &stubClusterInitializer{}, zap.NewNop())
+	assert.NotNil(server)
+	assert.NotNil(server.logger)
+	assert.NotNil(server.nodeLock)
+	assert.NotNil(server.initializer)
+	assert.NotNil(server.grpcServer)
+	assert.NotNil(server.fileHandler)
+	assert.NotNil(server.disk)
+}
+
+func TestInit(t *testing.T) {
+	someErr := errors.New("failed")
+	lockedNodeLock := nodelock.New()
+	require.True(t, lockedNodeLock.TryLockOnce())
+
+	testCases := map[string]struct {
+		nodeLock     *nodelock.Lock
+		initializer  ClusterInitializer
+		disk         encryptedDisk
+		fileHandler  file.Handler
+		req          *initproto.InitRequest
+		wantErr      bool
+		wantShutdown bool
+	}{
+		"successful init": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{},
+			disk:        &stubDisk{},
+			fileHandler: file.NewHandler(afero.NewMemMapFs()),
+			req:         &initproto.InitRequest{},
+		},
+		"node locked": {
+			nodeLock:     lockedNodeLock,
+			initializer:  &stubClusterInitializer{},
+			disk:         &stubDisk{},
+			fileHandler:  file.NewHandler(afero.NewMemMapFs()),
+			req:          &initproto.InitRequest{},
+			wantErr:      true,
+			wantShutdown: true,
+		},
+		"disk open error": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{},
+			disk:        &stubDisk{openErr: someErr},
+			fileHandler: file.NewHandler(afero.NewMemMapFs()),
+			req:         &initproto.InitRequest{},
+			wantErr:     true,
+		},
+		"disk uuid error": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{},
+			disk:        &stubDisk{uuidErr: someErr},
+			fileHandler: file.NewHandler(afero.NewMemMapFs()),
+			req:         &initproto.InitRequest{},
+			wantErr:     true,
+		},
+		"disk update passphrase error": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{},
+			disk:        &stubDisk{updatePassphraseErr: someErr},
+			fileHandler: file.NewHandler(afero.NewMemMapFs()),
+			req:         &initproto.InitRequest{},
+			wantErr:     true,
+		},
+		"write state file error": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{},
+			disk:        &stubDisk{},
+			fileHandler: file.NewHandler(afero.NewReadOnlyFs(afero.NewMemMapFs())),
+			req:         &initproto.InitRequest{},
+			wantErr:     true,
+		},
+		"initialize cluster error": {
+			nodeLock:    nodelock.New(),
+			initializer: &stubClusterInitializer{initClusterErr: someErr},
+			disk:        &stubDisk{},
+			fileHandler: file.NewHandler(afero.NewMemMapFs()),
+			req:         &initproto.InitRequest{},
+			wantErr:     true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			assert := assert.New(t)
+
+			serveStopper := newStubServeStopper()
+			server := &Server{
+				nodeLock:    tc.nodeLock,
+				initializer: tc.initializer,
+				disk:        tc.disk,
+				fileHandler: tc.fileHandler,
+				logger:      zaptest.NewLogger(t),
+				grpcServer:  serveStopper,
+			}
+
+			kubeconfig, err := server.Init(context.Background(), tc.req)
+
+			if tc.wantErr {
+				assert.Error(err)
+
+				if tc.wantShutdown {
+					select {
+					case <-serveStopper.shutdownCalled:
+					case <-time.After(time.Second):
+						t.Fatal("grpc server did not shut down")
+					}
+				}
+
+				return
+			}
+
+			assert.NoError(err)
+			assert.NotNil(kubeconfig)
+			assert.False(server.nodeLock.TryLockOnce()) // lock should be locked
+		})
+	}
+}
+
+func TestSSHProtoKeysToMap(t *testing.T) {
+	testCases := map[string]struct {
+		keys []*initproto.SSHUserKey
+		want map[string]string
+	}{
+		"empty": {
+			keys: []*initproto.SSHUserKey{},
+			want: map[string]string{},
+		},
+		"one key": {
+			keys: []*initproto.SSHUserKey{
+				{Username: "key1", PublicKey: "key1-key"},
+			},
+			want: map[string]string{
+				"key1": "key1-key",
+			},
+		},
+		"two keys": {
+			keys: []*initproto.SSHUserKey{
+				{Username: "key1", PublicKey: "key1-key"},
+				{Username: "key2", PublicKey: "key2-key"},
+			},
+			want: map[string]string{
+				"key1": "key1-key",
+				"key2": "key2-key",
+			},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			assert := assert.New(t)
+
+			got := sshProtoKeysToMap(tc.keys)
+			assert.Equal(tc.want, got)
+		})
+	}
+}
+
+type stubDisk struct {
+	openErr                error
+	closeErr               error
+	uuid                   string
+	uuidErr                error
+	updatePassphraseErr    error
+	updatePassphraseCalled bool
+}
+
+func (d *stubDisk) Open() error {
+	return d.openErr
+}
+
+func (d *stubDisk) Close() error {
+	return d.closeErr
+}
+
+func (d *stubDisk) UUID() (string, error) {
+	return d.uuid, d.uuidErr
+}
+
+func (d *stubDisk) UpdatePassphrase(string) error {
+	d.updatePassphraseCalled = true
+	return d.updatePassphraseErr
+}
+
+type stubClusterInitializer struct {
+	initClusterKubeconfig []byte
+	initClusterErr        error
+}
+
+func (i *stubClusterInitializer) InitCluster(context.Context, []string, string, string, attestationtypes.ID, kubernetes.KMSConfig, map[string]string,
+) ([]byte, error) {
+	return i.initClusterKubeconfig, i.initClusterErr
+}
+
+type stubServeStopper struct {
+	shutdownCalled chan struct{}
+}
+
+func newStubServeStopper() *stubServeStopper {
+	return &stubServeStopper{shutdownCalled: make(chan struct{}, 1)}
+}
+
+func (s *stubServeStopper) Serve(net.Listener) error {
+	panic("should not be called in a test")
+}
+
+func (s *stubServeStopper) GracefulStop() {
+	s.shutdownCalled <- struct{}{}
+}

coordinator/internal/joinclient/client.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/edgelesssys/constellation/activation/activationproto"
 	"github.com/edgelesssys/constellation/coordinator/internal/diskencryption"
+	"github.com/edgelesssys/constellation/coordinator/internal/nodelock"
 	"github.com/edgelesssys/constellation/coordinator/nodestate"
 	"github.com/edgelesssys/constellation/coordinator/role"
 	"github.com/edgelesssys/constellation/internal/cloud/metadata"
@@ -19,7 +20,7 @@ import (
 	"github.com/spf13/afero"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
-	"k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
+	kubeadm "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1beta3"
 	"k8s.io/utils/clock"
 )
 
@@ -30,11 +31,11 @@ const (
 
 // JoinClient is a client for self-activation of node.
 type JoinClient struct {
-	nodeLock    *sync.Mutex
+	nodeLock    *nodelock.Lock
 	diskUUID    string
 	nodeName    string
 	role        role.Role
-	disk        EncryptedDisk
+	disk        encryptedDisk
 	fileHandler file.Handler
 
 	timeout  time.Duration
@@ -43,7 +44,7 @@ type JoinClient struct {
 
 	dialer      grpcDialer
 	joiner      ClusterJoiner
-	metadataAPI metadataAPI
+	metadataAPI MetadataAPI
 
 	log *zap.Logger
 
@@ -53,7 +54,7 @@ type JoinClient struct {
 }
 
 // New creates a new SelfActivationClient.
-func New(nodeLock *sync.Mutex, dial grpcDialer, joiner ClusterJoiner, meta metadataAPI, log *zap.Logger) *JoinClient {
+func New(lock *nodelock.Lock, dial grpcDialer, joiner ClusterJoiner, meta MetadataAPI, log *zap.Logger) *JoinClient {
 	return &JoinClient{
 		disk:        diskencryption.New(),
 		fileHandler: file.NewHandler(afero.NewOsFs()),
@@ -87,11 +88,11 @@ func (c *JoinClient) Start() {
 	go func() {
 		defer ticker.Stop()
 		defer func() { c.stopDone <- struct{}{} }()
+		defer c.log.Info("Client stopped")
 
-		diskUUID, err := c.GetDiskUUID()
+		diskUUID, err := c.getDiskUUID()
 		if err != nil {
 			c.log.Error("Failed to get disk UUID", zap.Error(err))
-			c.log.Error("Stopping self-activation client")
 			return
 		}
 		c.diskUUID = diskUUID
@@ -117,6 +118,9 @@ func (c *JoinClient) Start() {
 			if err == nil {
 				c.log.Info("Activated successfully. SelfActivationClient shut down.")
 				return
+			} else if isUnrecoverable(err) {
+				c.log.Error("Unrecoverable error occurred", zap.Error(err))
+				return
 			}
 			c.log.Info("Activation failed for all available endpoints", zap.Error(err))
 
@@ -247,8 +251,18 @@ func (c *JoinClient) joinAsControlPlaneNode(ctx context.Context, client activati
 
 func (c *JoinClient) startNodeAndJoin(ctx context.Context, diskKey, ownerID, clusterID, kubeletKey, kubeletCert []byte, endpoint, token,
 	discoveryCACertHash, certKey string,
-) error {
+) (retErr error) {
-	if ok := c.nodeLock.TryLock(); !ok {
+	// If an error occurs in this func, the client cannot continue.
+	defer func() {
+		if retErr != nil {
+			retErr = unrecoverableError{retErr}
+		}
+	}()
+
+	if ok := c.nodeLock.TryLockOnce(); !ok {
+		// There is already a cluster initialization in progress on
+		// this node, so there is no need to also join the cluster,
+		// as the initializing node is automatically part of the cluster.
 		return errors.New("node is already being initialized")
 	}
 
@@ -310,7 +324,7 @@ func (c *JoinClient) updateDiskPassphrase(passphrase string) error {
 	return c.disk.UpdatePassphrase(passphrase)
 }
 
-func (c *JoinClient) GetDiskUUID() (string, error) {
+func (c *JoinClient) getDiskUUID() (string, error) {
 	if err := c.disk.Open(); err != nil {
 		return "", fmt.Errorf("opening disk: %w", err)
 	}
@@ -343,11 +357,21 @@ func (c *JoinClient) timeoutCtx() (context.Context, context.CancelFunc) {
 	return context.WithTimeout(context.Background(), c.timeout)
 }
 
+type unrecoverableError struct{ error }
+
+func isUnrecoverable(err error) bool {
+	var ue *unrecoverableError
+	ok := errors.As(err, &ue)
+	return ok
+}
+
 type grpcDialer interface {
 	Dial(ctx context.Context, target string) (*grpc.ClientConn, error)
 }
 
+// ClusterJoiner has the ability to join a new node to an existing cluster.
 type ClusterJoiner interface {
+	// JoinCluster joins a new node to an existing cluster.
 	JoinCluster(
 		ctx context.Context,
 		args *kubeadm.BootstrapTokenDiscovery,
@@ -356,15 +380,15 @@ type ClusterJoiner interface {
 	) error
 }
 
-type metadataAPI interface {
+// MetadataAPI provides information about the instances.
+type MetadataAPI interface {
 	// List retrieves all instances belonging to the current constellation.
 	List(ctx context.Context) ([]metadata.InstanceMetadata, error)
 	// Self retrieves the current instance.
 	Self(ctx context.Context) (metadata.InstanceMetadata, error)
 }
 
-// EncryptedDisk manages the encrypted state disk.
+type encryptedDisk interface {
-type EncryptedDisk interface {
 	// Open prepares the underlying device for disk operations.
 	Open() error
 	// Close closes the underlying device.

coordinator/internal/joinclient/client_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/edgelesssys/constellation/activation/activationproto"
+	"github.com/edgelesssys/constellation/coordinator/internal/nodelock"
 	"github.com/edgelesssys/constellation/coordinator/role"
 	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	"github.com/edgelesssys/constellation/internal/constants"
@@ -18,11 +19,12 @@ import (
 	"github.com/edgelesssys/constellation/internal/grpc/dialer"
 	"github.com/edgelesssys/constellation/internal/grpc/testdialer"
 	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
 	"go.uber.org/goleak"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 	"google.golang.org/grpc"
-	"k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm"
+	kubeadm "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1beta3"
 	testclock "k8s.io/utils/clock/testing"
 )
 
@@ -41,9 +43,9 @@ func TestClient(t *testing.T) {
 
 	testCases := map[string]struct {
 		role          role.Role
-		clusterJoiner ClusterJoiner
+		clusterJoiner *stubClusterJoiner
-		disk          EncryptedDisk
+		disk          encryptedDisk
-		nodeLock      *sync.Mutex
+		nodeLock      *nodelock.Lock
 		apiAnswers    []any
 	}{
 		"on node: metadata self: errors occur": {
@@ -57,7 +59,7 @@ func TestClient(t *testing.T) {
 				activateWorkerNodeAnswer{},
 			},
 			clusterJoiner: &stubClusterJoiner{},
-			nodeLock:      &sync.Mutex{},
+			nodeLock:      nodelock.New(),
 			disk:          &stubDisk{},
 		},
 		"on node: metadata self: invalid answer": {
@@ -71,7 +73,7 @@ func TestClient(t *testing.T) {
 				activateWorkerNodeAnswer{},
 			},
 			clusterJoiner: &stubClusterJoiner{},
-			nodeLock:      &sync.Mutex{},
+			nodeLock:      nodelock.New(),
 			disk:          &stubDisk{},
 		},
 		"on node: metadata list: errors occur": {
@@ -85,7 +87,7 @@ func TestClient(t *testing.T) {
 				activateWorkerNodeAnswer{},
 			},
 			clusterJoiner: &stubClusterJoiner{},
-			nodeLock:      &sync.Mutex{},
+			nodeLock:      nodelock.New(),
 			disk:          &stubDisk{},
 		},
 		"on node: metadata list: no coordinators in answer": {
@@ -99,7 +101,7 @@ func TestClient(t *testing.T) {
 				activateWorkerNodeAnswer{},
 			},
 			clusterJoiner: &stubClusterJoiner{},
-			nodeLock:      &sync.Mutex{},
+			nodeLock:      nodelock.New(),
 			disk:          &stubDisk{},
 		},
 		"on node: aaas ActivateNode: errors": {
@@ -114,13 +116,15 @@ func TestClient(t *testing.T) {
 				activateWorkerNodeAnswer{},
 			},
 			clusterJoiner: &stubClusterJoiner{},
-			nodeLock:      &sync.Mutex{},
+			nodeLock:      nodelock.New(),
 			disk:          &stubDisk{},
 		},
 	}
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
+			assert := assert.New(t)
+
 			clock := testclock.NewFakeClock(time.Now())
 			metadataAPI := newStubMetadataAPI()
 			fileHandler := file.NewHandler(afero.NewMemMapFs())
@@ -165,12 +169,24 @@ func TestClient(t *testing.T) {
 			}
 
 			client.Stop()
+
+			assert.True(tc.clusterJoiner.joinClusterCalled)
+			assert.False(client.nodeLock.TryLockOnce()) // lock should be locked
 		})
 	}
 }
 
 func TestClientConcurrentStartStop(t *testing.T) {
+	netDialer := testdialer.NewBufconnDialer()
+	dialer := dialer.New(nil, nil, netDialer)
 	client := &JoinClient{
+		nodeLock:    nodelock.New(),
+		timeout:     30 * time.Second,
+		interval:    30 * time.Second,
+		dialer:      dialer,
+		disk:        &stubDisk{},
+		joiner:      &stubClusterJoiner{},
+		fileHandler: file.NewHandler(afero.NewMemMapFs()),
 		metadataAPI: &stubRepeaterMetadataAPI{},
 		clock:       testclock.NewFakeClock(time.Now()),
 		log:         zap.NewNop(),
@@ -293,10 +309,12 @@ type activateControlPlaneNodeAnswer struct {
 }
 
 type stubClusterJoiner struct {
-	joinClusterErr error
+	joinClusterCalled bool
+	joinClusterErr    error
 }
 
 func (j *stubClusterJoiner) JoinCluster(context.Context, *kubeadm.BootstrapTokenDiscovery, string, role.Role) error {
+	j.joinClusterCalled = true
 	return j.joinClusterErr
 }
 

coordinator/internal/kubernetes/cloud_provider.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/edgelesssys/constellation/coordinator/internal/kubernetes/k8sapi/resources"
-	"github.com/edgelesssys/constellation/coordinator/role"
 	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	k8s "k8s.io/api/core/v1"
 )
@@ -23,10 +22,6 @@ type ProviderMetadata interface {
 	GetLoadBalancerIP(ctx context.Context) (string, error)
 	// GetInstance retrieves an instance using its providerID.
 	GetInstance(ctx context.Context, providerID string) (metadata.InstanceMetadata, error)
-	// SignalRole signals the constellation role via cloud provider metadata (if supported by the CSP and deployment type, otherwise does nothing).
-	SignalRole(ctx context.Context, role role.Role) error
-	// SetVPNIP stores the internally used VPN IP in cloud provider metadata (if supported and required for autoscaling by the CSP, otherwise does nothing).
-	SetVPNIP(ctx context.Context, vpnIP string) error
 	// Supported is used to determine if metadata API is implemented for this cloud provider.
 	Supported() bool
 }
@@ -96,9 +91,6 @@ type stubProviderMetadata struct {
 	ListErr  error
 	ListResp []metadata.InstanceMetadata
 
-	SignalRoleErr error
-	SetVPNIPErr   error
-
 	SelfErr  error
 	SelfResp metadata.InstanceMetadata
 
@@ -129,14 +121,6 @@ func (m *stubProviderMetadata) GetInstance(ctx context.Context, providerID strin
 	return m.GetInstanceResp, m.GetInstanceErr
 }
 
-func (m *stubProviderMetadata) SignalRole(ctx context.Context, role role.Role) error {
-	return m.SignalRoleErr
-}
-
-func (m *stubProviderMetadata) SetVPNIP(ctx context.Context, vpnIP string) error {
-	return m.SetVPNIPErr
-}
-
 func (m *stubProviderMetadata) Supported() bool {
 	return m.SupportedResp
 }

coordinator/internal/kubernetes/k8sapi/resources/images.go

@@ -0,0 +1,12 @@
+package resources
+
+const (
+	// Constellation images.
+	activationImage    = "ghcr.io/edgelesssys/constellation/activation-service:v1.2"
+	accessManagerImage = "ghcr.io/edgelesssys/constellation/access-manager:v1.2"
+	kmsImage           = "ghcr.io/edgelesssys/constellation/kmsserver:v1.2"
+	verificationImage  = "ghcr.io/edgelesssys/constellation/verification-service:v1.2"
+
+	// external images.
+	clusterAutoscalerImage = "k8s.gcr.io/autoscaling/cluster-autoscaler:v1.23.0"
+)

coordinator/internal/kubernetes/k8sapi/resources/verification.go

@@ -0,0 +1,153 @@
+package resources
+
+import (
+	"fmt"
+
+	"github.com/edgelesssys/constellation/internal/constants"
+	"github.com/edgelesssys/constellation/internal/secrets"
+	apps "k8s.io/api/apps/v1"
+	k8s "k8s.io/api/core/v1"
+	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+type verificationDaemonset struct {
+	DaemonSet apps.DaemonSet
+	Service   k8s.Service
+}
+
+func NewVerificationDaemonSet(csp string) *verificationDaemonset {
+	return &verificationDaemonset{
+		DaemonSet: apps.DaemonSet{
+			TypeMeta: meta.TypeMeta{
+				APIVersion: "apps/v1",
+				Kind:       "DaemonSet",
+			},
+			ObjectMeta: meta.ObjectMeta{
+				Name:      "verification-service",
+				Namespace: "kube-system",
+				Labels: map[string]string{
+					"k8s-app":   "verification-service",
+					"component": "verification-service",
+				},
+			},
+			Spec: apps.DaemonSetSpec{
+				Selector: &meta.LabelSelector{
+					MatchLabels: map[string]string{
+						"k8s-app": "verification-service",
+					},
+				},
+				Template: k8s.PodTemplateSpec{
+					ObjectMeta: meta.ObjectMeta{
+						Labels: map[string]string{
+							"k8s-app": "verification-service",
+						},
+					},
+					Spec: k8s.PodSpec{
+						Tolerations: []k8s.Toleration{
+							{
+								Key:      "node-role.kubernetes.io/master",
+								Operator: k8s.TolerationOpEqual,
+								Value:    "true",
+								Effect:   k8s.TaintEffectNoSchedule,
+							},
+							{
+								Key:      "node-role.kubernetes.io/control-plane",
+								Operator: k8s.TolerationOpExists,
+								Effect:   k8s.TaintEffectNoSchedule,
+							},
+							{
+								Operator: k8s.TolerationOpExists,
+								Effect:   k8s.TaintEffectNoExecute,
+							},
+							{
+								Operator: k8s.TolerationOpExists,
+								Effect:   k8s.TaintEffectNoSchedule,
+							},
+						},
+						ImagePullSecrets: []k8s.LocalObjectReference{
+							{
+								Name: secrets.PullSecretName,
+							},
+						},
+						Containers: []k8s.Container{
+							{
+								Name:  "verification-service",
+								Image: verificationImage,
+								Ports: []k8s.ContainerPort{
+									{
+										Name:          "http",
+										ContainerPort: constants.VerifyServicePortHTTP,
+									},
+									{
+										Name:          "grpc",
+										ContainerPort: constants.VerifyServicePortGRPC,
+									},
+								},
+								SecurityContext: &k8s.SecurityContext{
+									Privileged: func(b bool) *bool { return &b }(true),
+								},
+								Args: []string{
+									fmt.Sprintf("--cloud-provider=%s", csp),
+								},
+								VolumeMounts: []k8s.VolumeMount{
+									{
+										Name:      "event-log",
+										ReadOnly:  true,
+										MountPath: "/sys/kernel/security/",
+									},
+								},
+							},
+						},
+						Volumes: []k8s.Volume{
+							{
+								Name: "event-log",
+								VolumeSource: k8s.VolumeSource{
+									HostPath: &k8s.HostPathVolumeSource{
+										Path: "/sys/kernel/security/",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		Service: k8s.Service{
+			TypeMeta: meta.TypeMeta{
+				APIVersion: "v1",
+				Kind:       "Service",
+			},
+			ObjectMeta: meta.ObjectMeta{
+				Name:      "activation-service",
+				Namespace: "kube-system",
+			},
+			Spec: k8s.ServiceSpec{
+				Type: k8s.ServiceTypeNodePort,
+				Ports: []k8s.ServicePort{
+					{
+						Name:       "http",
+						Protocol:   k8s.ProtocolTCP,
+						Port:       constants.VerifyServicePortHTTP,
+						TargetPort: intstr.FromInt(constants.VerifyServicePortHTTP),
+						NodePort:   constants.VerifyServiceNodePortHTTP,
+					},
+					{
+						Name:       "grpc",
+						Protocol:   k8s.ProtocolTCP,
+						Port:       constants.VerifyServicePortGRPC,
+						TargetPort: intstr.FromInt(constants.VerifyServicePortGRPC),
+						NodePort:   constants.VerifyServiceNodePortGRPC,
+					},
+				},
+				Selector: map[string]string{
+					"k8s-app": "verification-service",
+				},
+			},
+		},
+	}
+}
+
+func (v *verificationDaemonset) Marshal() ([]byte, error) {
+	return MarshalK8SResources(v)
+}

coordinator/internal/kubernetes/k8sapi/resources/verification_test.go

@@ -0,0 +1,18 @@
+package resources
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewVerificationDaemonset(t *testing.T) {
+	deployment := NewVerificationDaemonSet("csp")
+	deploymentYAML, err := deployment.Marshal()
+	require.NoError(t, err)
+
+	var recreated verificationDaemonset
+	require.NoError(t, UnmarshalK8SResources(deploymentYAML, &recreated))
+	assert.Equal(t, deployment, &recreated)
+}

coordinator/internal/kubernetes/kubernetes.go

@@ -74,15 +74,15 @@ type KMSConfig struct {
 func (k *KubeWrapper) InitCluster(
 	ctx context.Context, autoscalingNodeGroups []string, cloudServiceAccountURI, k8sVersion string,
 	id attestationtypes.ID, kmsConfig KMSConfig, sshUsers map[string]string,
-) error {
+) ([]byte, error) {
 	// TODO: k8s version should be user input
 	if err := k.clusterUtil.InstallComponents(ctx, k8sVersion); err != nil {
-		return err
+		return nil, err
 	}
 
 	ip, err := k.getIPAddr()
 	if err != nil {
-		return err
+		return nil, err
 	}
 	nodeName := ip
 	var providerID string
@@ -98,7 +98,7 @@ func (k *KubeWrapper) InitCluster(
 	if k.providerMetadata.Supported() {
 		instance, err = k.providerMetadata.Self(ctx)
 		if err != nil {
-			return fmt.Errorf("retrieving own instance metadata failed: %w", err)
+			return nil, fmt.Errorf("retrieving own instance metadata failed: %w", err)
 		}
 		nodeName = k8sCompliantHostname(instance.Name)
 		providerID = instance.ProviderID
@@ -113,13 +113,13 @@ func (k *KubeWrapper) InitCluster(
 		}
 		subnetworkPodCIDR, err = k.providerMetadata.GetSubnetworkCIDR(ctx)
 		if err != nil {
-			return fmt.Errorf("retrieving subnetwork CIDR failed: %w", err)
+			return nil, fmt.Errorf("retrieving subnetwork CIDR failed: %w", err)
 		}
 		controlPlaneEndpointIP = publicIP
 		if k.providerMetadata.SupportsLoadBalancer() {
 			controlPlaneEndpointIP, err = k.providerMetadata.GetLoadBalancerIP(ctx)
 			if err != nil {
-				return fmt.Errorf("retrieving load balancer IP failed: %w", err)
+				return nil, fmt.Errorf("retrieving load balancer IP failed: %w", err)
 			}
 		}
 	}
@@ -133,14 +133,14 @@ func (k *KubeWrapper) InitCluster(
 	initConfig.SetControlPlaneEndpoint(controlPlaneEndpointIP)
 	initConfigYAML, err := initConfig.Marshal()
 	if err != nil {
-		return fmt.Errorf("encoding kubeadm init configuration as YAML: %w", err)
+		return nil, fmt.Errorf("encoding kubeadm init configuration as YAML: %w", err)
 	}
 	if err := k.clusterUtil.InitCluster(ctx, initConfigYAML); err != nil {
-		return fmt.Errorf("kubeadm init: %w", err)
+		return nil, fmt.Errorf("kubeadm init: %w", err)
 	}
 	kubeConfig, err := k.GetKubeconfig()
 	if err != nil {
-		return fmt.Errorf("reading kubeconfig after cluster initialization: %w", err)
+		return nil, fmt.Errorf("reading kubeconfig after cluster initialization: %w", err)
 	}
 	k.client.SetKubeconfig(kubeConfig)
 
@@ -154,43 +154,43 @@ func (k *KubeWrapper) InitCluster(
 		ProviderID:        providerID,
 	}
 	if err = k.clusterUtil.SetupPodNetwork(ctx, setupPodNetworkInput); err != nil {
-		return fmt.Errorf("setting up pod network: %w", err)
+		return nil, fmt.Errorf("setting up pod network: %w", err)
 	}
 
 	kms := resources.NewKMSDeployment(k.cloudProvider, kmsConfig.MasterSecret)
 	if err = k.clusterUtil.SetupKMS(k.client, kms); err != nil {
-		return fmt.Errorf("setting up kms: %w", err)
+		return nil, fmt.Errorf("setting up kms: %w", err)
 	}
 
 	if err := k.setupActivationService(k.cloudProvider, k.initialMeasurementsJSON, id); err != nil {
-		return fmt.Errorf("setting up activation service failed: %w", err)
+		return nil, fmt.Errorf("setting up activation service failed: %w", err)
 	}
 
 	if err := k.setupCCM(ctx, subnetworkPodCIDR, cloudServiceAccountURI, instance); err != nil {
-		return fmt.Errorf("setting up cloud controller manager: %w", err)
+		return nil, fmt.Errorf("setting up cloud controller manager: %w", err)
 	}
 	if err := k.setupCloudNodeManager(); err != nil {
-		return fmt.Errorf("setting up cloud node manager: %w", err)
+		return nil, fmt.Errorf("setting up cloud node manager: %w", err)
 	}
 
 	if err := k.setupClusterAutoscaler(instance, cloudServiceAccountURI, autoscalingNodeGroups); err != nil {
-		return fmt.Errorf("setting up cluster autoscaler: %w", err)
+		return nil, fmt.Errorf("setting up cluster autoscaler: %w", err)
 	}
 
 	accessManager := resources.NewAccessManagerDeployment(sshUsers)
 	if err := k.clusterUtil.SetupAccessManager(k.client, accessManager); err != nil {
-		return fmt.Errorf("failed to setup access-manager: %w", err)
+		return nil, fmt.Errorf("failed to setup access-manager: %w", err)
 	}
 
 	if err := k.clusterUtil.SetupVerificationService(
 		k.client, resources.NewVerificationDaemonSet(k.cloudProvider),
 	); err != nil {
-		return fmt.Errorf("failed to setup verification service: %w", err)
+		return nil, fmt.Errorf("failed to setup verification service: %w", err)
 	}
 
 	go k.clusterUtil.FixCilium(nodeName)
 
-	return nil
+	return k.GetKubeconfig()
 }
 
 // JoinCluster joins existing Kubernetes cluster.

coordinator/internal/kubernetes/kubernetes_test.go

@@ -268,7 +268,7 @@ func TestInitCluster(t *testing.T) {
 				kubeconfigReader:       tc.kubeconfigReader,
 				getIPAddr:              func() (string, error) { return privateIP, nil },
 			}
-			err := kube.InitCluster(context.Background(), autoscalingNodeGroups, serviceAccountUri, k8sVersion, attestationtypes.ID{}, KMSConfig{MasterSecret: masterSecret}, nil)
+			_, err := kube.InitCluster(context.Background(), autoscalingNodeGroups, serviceAccountUri, k8sVersion, attestationtypes.ID{}, KMSConfig{MasterSecret: masterSecret}, nil)
 
 			if tc.wantErr {
 				assert.Error(err)

coordinator/internal/nodelock/nodelock.go

@@ -0,0 +1,25 @@
+package nodelock
+
+import "sync"
+
+// Lock locks the node once there the join or the init is at a point
+// where there is no turning back and the other operation does not need
+// to continue.
+//
+// This can be viewed as a state machine with two states: unlocked and locked.
+// There is no way to unlock, so the state changes only once from unlock to
+// locked.
+type Lock struct {
+	mux *sync.Mutex
+}
+
+// New creates a new NodeLock, which is unlocked.
+func New() *Lock {
+	return &Lock{mux: &sync.Mutex{}}
+}
+
+// TryLockOnce tries to lock the node. If the node is already locked, it
+// returns false. If the node is unlocked, it locks it and returns true.
+func (n *Lock) TryLockOnce() bool {
+	return n.mux.TryLock()
+}

debugd/debugd/metadata/cloudprovider/cloudprovider.go

@@ -5,17 +5,17 @@ import (
 	"fmt"
 
 	azurecloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/azure"
-	"github.com/edgelesssys/constellation/coordinator/cloudprovider/cloudtypes"
 	gcpcloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/gcp"
 	qemucloud "github.com/edgelesssys/constellation/coordinator/cloudprovider/qemu"
+	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	"github.com/edgelesssys/constellation/internal/deploy/ssh"
 )
 
 type providerMetadata interface {
 	// List retrieves all instances belonging to the current constellation.
-	List(ctx context.Context) ([]cloudtypes.Instance, error)
+	List(ctx context.Context) ([]metadata.InstanceMetadata, error)
 	// Self retrieves the current instance.
-	Self(ctx context.Context) (cloudtypes.Instance, error)
+	Self(ctx context.Context) (metadata.InstanceMetadata, error)
 }
 
 // Fetcher checks the metadata service to search for instances that were set up for debugging and cloud provider specific SSH keys.

debugd/debugd/metadata/cloudprovider/cloudprovider_test.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/edgelesssys/constellation/coordinator/cloudprovider/cloudtypes"
+	"github.com/edgelesssys/constellation/internal/cloud/metadata"
 	"github.com/edgelesssys/constellation/internal/deploy/ssh"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -29,7 +29,7 @@ func TestDiscoverDebugIPs(t *testing.T) {
 	}{
 		"disovery works": {
 			meta: stubMetadata{
-				listRes: []cloudtypes.Instance{
+				listRes: []metadata.InstanceMetadata{
 					{
 						PrivateIPs: []string{"192.0.2.0"},
 					},
@@ -83,7 +83,7 @@ func TestFetchSSHKeys(t *testing.T) {
 	}{
 		"fetch works": {
 			meta: stubMetadata{
-				selfRes: cloudtypes.Instance{
+				selfRes: metadata.InstanceMetadata{
 					Name:       "name",
 					ProviderID: "provider-id",
 					SSHKeys:    map[string][]string{"bob": {"ssh-rsa bobskey"}},
@@ -125,24 +125,24 @@ func TestFetchSSHKeys(t *testing.T) {
 }
 
 type stubMetadata struct {
-	listRes        []cloudtypes.Instance
+	listRes        []metadata.InstanceMetadata
 	listErr        error
-	selfRes        cloudtypes.Instance
+	selfRes        metadata.InstanceMetadata
 	selfErr        error
-	getInstanceRes cloudtypes.Instance
+	getInstanceRes metadata.InstanceMetadata
 	getInstanceErr error
 	supportedRes   bool
 }
 
-func (m *stubMetadata) List(ctx context.Context) ([]cloudtypes.Instance, error) {
+func (m *stubMetadata) List(ctx context.Context) ([]metadata.InstanceMetadata, error) {
 	return m.listRes, m.listErr
 }
 
-func (m *stubMetadata) Self(ctx context.Context) (cloudtypes.Instance, error) {
+func (m *stubMetadata) Self(ctx context.Context) (metadata.InstanceMetadata, error) {
 	return m.selfRes, m.selfErr
 }
 
-func (m *stubMetadata) GetInstance(ctx context.Context, providerID string) (cloudtypes.Instance, error) {
+func (m *stubMetadata) GetInstance(ctx context.Context, providerID string) (metadata.InstanceMetadata, error) {
 	return m.getInstanceRes, m.getInstanceErr
 }
 

internal/atls/atls.go

@@ -17,8 +17,8 @@ import (
 	"math/big"
 	"time"
 
-	"github.com/edgelesssys/constellation/coordinator/config"
 	"github.com/edgelesssys/constellation/coordinator/util"
+	"github.com/edgelesssys/constellation/internal/constants"
 	"github.com/edgelesssys/constellation/internal/oid"
 )
 
@@ -45,7 +45,7 @@ func CreateAttestationServerTLSConfig(issuer Issuer, validators []Validator) (*t
 // If no validators are set, the server's attestation document will not be verified.
 // If issuer is nil, the client will be unable to perform mutual aTLS.
 func CreateAttestationClientTLSConfig(issuer Issuer, validators []Validator) (*tls.Config, error) {
-	clientNonce, err := util.GenerateRandomBytes(config.RNGLengthDefault)
+	clientNonce, err := util.GenerateRandomBytes(constants.RNGLengthDefault)
 	if err != nil {
 		return nil, err
 	}
@@ -87,7 +87,7 @@ func getATLSConfigForClientFunc(issuer Issuer, validators []Validator) (func(*tl
 	// this function will be called once for every client
 	return func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
 		// generate nonce for this connection
-		serverNonce, err := util.GenerateRandomBytes(config.RNGLengthDefault)
+		serverNonce, err := util.GenerateRandomBytes(constants.RNGLengthDefault)
 		if err != nil {
 			return nil, err
 		}

internal/azureshared/metadata_test.go

@@ -81,56 +81,6 @@ func TestUIDFromProviderID(t *testing.T) {
 	}
 }
 
-func TestVMInformationFromProviderID(t *testing.T) {
-	testCases := map[string]struct {
-		providerID         string
-		wantSubscriptionID string
-		wantResourceGroup  string
-		wantInstanceName   string
-		wantErr            bool
-	}{
-		"simple id": {
-			providerID:         "azure:///subscriptions/subscription-id/resourceGroups/resource-group/providers/Microsoft.Compute/virtualMachines/instance-id",
-			wantSubscriptionID: "subscription-id",
-			wantResourceGroup:  "resource-group",
-			wantInstanceName:   "instance-id",
-		},
-		"missing instance": {
-			providerID: "azure:///subscriptions/subscription-id/resourceGroups/resource-group/providers/Microsoft.Compute/virtualMachines",
-			wantErr:    true,
-		},
-		"providerID for scale set instance must fail": {
-			providerID: "azure:///subscriptions/subscription-id/resourceGroups/resource-group/providers/Microsoft.Compute/virtualMachineScaleSets/scale-set-name/virtualMachines/instance-id",
-			wantErr:    true,
-		},
-		"wrong provider": {
-			providerID: "gcp:///subscriptions/subscription-id/resourceGroups/resource-group/providers/Microsoft.Compute/virtualMachines/instance-id",
-			wantErr:    true,
-		},
-		"providerID is malformed": {
-			providerID: "malformed-provider-id",
-			wantErr:    true,
-		},
-	}
-
-	for name, tc := range testCases {
-		t.Run(name, func(t *testing.T) {
-			assert := assert.New(t)
-
-			subscriptionID, resourceGroup, instanceName, err := VMInformationFromProviderID(tc.providerID)
-
-			if tc.wantErr {
-				assert.Error(err)
-				return
-			}
-			assert.NoError(err)
-			assert.Equal(tc.wantSubscriptionID, subscriptionID)
-			assert.Equal(tc.wantResourceGroup, resourceGroup)
-			assert.Equal(tc.wantInstanceName, instanceName)
-		})
-	}
-}
-
 func TestScaleSetInformationFromProviderID(t *testing.T) {
 	testCases := map[string]struct {
 		providerID         string

internal/cloud/metadata/metadata.go

@@ -27,10 +27,6 @@ type metadataAPI interface {
 	List(ctx context.Context) ([]InstanceMetadata, error)
 	// Self retrieves the current instance.
 	Self(ctx context.Context) (InstanceMetadata, error)
-	// SignalRole signals the constellation role via cloud provider metadata (if supported by the CSP and deployment type, otherwise does nothing).
-	SignalRole(ctx context.Context, role role.Role) error
-	// SetVPNIP stores the internally used VPN IP in cloud provider metadata (if supported and required for autoscaling by the CSP, otherwise does nothing).
-	SetVPNIP(ctx context.Context, vpnIP string) error
 	// Supported is used to determine if metadata API is implemented for this cloud provider.
 	Supported() bool
 }

internal/constants/constants.go

@@ -82,6 +82,8 @@ const (
 	MasterSecretLengthDefault = 32
 	// MasterSecretLengthMin is the minimal length in bytes for user provided master secrets.
 	MasterSecretLengthMin = 16
+	// RNGLengthDefault is the number of bytes used for generating nonces.
+	RNGLengthDefault = 32
 
 	//
 	// CLI.

state/keyservice/keyservice.go

@@ -7,9 +7,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/edgelesssys/constellation/coordinator/config"
 	"github.com/edgelesssys/constellation/coordinator/core"
 	"github.com/edgelesssys/constellation/coordinator/pubapi/pubproto"
+	"github.com/edgelesssys/constellation/internal/constants"
 	"github.com/edgelesssys/constellation/internal/grpc/atlscredentials"
 	"github.com/edgelesssys/constellation/internal/logger"
 	"github.com/edgelesssys/constellation/internal/oid"
@@ -51,8 +51,8 @@ func (a *KeyAPI) PushStateDiskKey(ctx context.Context, in *keyproto.PushStateDis
 	if len(a.key) != 0 {
 		return nil, status.Error(codes.FailedPrecondition, "node already received a passphrase")
 	}
-	if len(in.StateDiskKey) != config.RNGLengthDefault {
+	if len(in.StateDiskKey) != constants.RNGLengthDefault {
-		return nil, status.Errorf(codes.InvalidArgument, "received invalid passphrase: expected length: %d, but got: %d", config.RNGLengthDefault, len(in.StateDiskKey))
+		return nil, status.Errorf(codes.InvalidArgument, "received invalid passphrase: expected length: %d, but got: %d", constants.RNGLengthDefault, len(in.StateDiskKey))
 	}
 
 	a.key = in.StateDiskKey

state/keyservice/keyservice_test.go

@@ -214,14 +214,6 @@ func (s stubMetadata) Self(ctx context.Context) (cloudtypes.Instance, error) {
 	return cloudtypes.Instance{}, nil
 }
 
-func (s stubMetadata) SignalRole(ctx context.Context, role role.Role) error {
-	return nil
-}
-
-func (s stubMetadata) SetVPNIP(ctx context.Context, vpnIP string) error {
-	return nil
-}
-
 func (s stubMetadata) Supported() bool {
 	return true
 }

state/setup/setup.go

@@ -8,9 +8,9 @@ import (
 	"path/filepath"
 	"syscall"
 
-	"github.com/edgelesssys/constellation/coordinator/config"
 	"github.com/edgelesssys/constellation/coordinator/nodestate"
 	"github.com/edgelesssys/constellation/internal/attestation/vtpm"
+	"github.com/edgelesssys/constellation/internal/constants"
 	"github.com/edgelesssys/constellation/internal/file"
 	"github.com/edgelesssys/constellation/internal/logger"
 	"github.com/spf13/afero"
@@ -99,7 +99,7 @@ func (s *SetupManager) PrepareNewDisk() error {
 		return err
 	}
 
-	passphrase := make([]byte, config.RNGLengthDefault)
+	passphrase := make([]byte, constants.RNGLengthDefault)
 	if _, err := rand.Read(passphrase); err != nil {
 		return err
 	}

state/setup/setup_test.go

@@ -7,9 +7,9 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/edgelesssys/constellation/coordinator/config"
 	"github.com/edgelesssys/constellation/coordinator/nodestate"
 	"github.com/edgelesssys/constellation/internal/attestation/vtpm"
+	"github.com/edgelesssys/constellation/internal/constants"
 	"github.com/edgelesssys/constellation/internal/file"
 	"github.com/edgelesssys/constellation/internal/logger"
 	"github.com/spf13/afero"
@@ -193,7 +193,7 @@ func TestPrepareNewDisk(t *testing.T) {
 
 				data, err := tc.fs.ReadFile(filepath.Join(keyPath, keyFile))
 				require.NoError(t, err)
-				assert.Len(data, config.RNGLengthDefault)
+				assert.Len(data, constants.RNGLengthDefault)
 			}
 		})
 	}