Setting Peer attributes from clusterlink cli (clusterlink-net#679)

* Attributes are passed to the controlplane thru its cli, and are added to the set of client attributes

* more logging

* Must delete privileged policy before continuing to next test

Signed-off-by: Ziv Nevo <[email protected]>

cmd/cl-controlplane/app/server.go

@@ -97,6 +97,8 @@ type Options struct {
 	LogFile string
 	// LogLevel is the log level.
 	LogLevel string
+	// PeerLabels hold the peer attributes (as "<key>:<value>" strings), to be used in access policies
+	PeerLabels map[string]string
 }
 
 // AddFlags adds flags to fs and binds them to options.
@@ -105,6 +107,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 		"Path to a file where logs will be written. If not specified, logs will be printed to stderr.")
 	fs.StringVar(&o.LogLevel, "log-level", logLevel,
 		"The log level. One of fatal, error, warn, info, debug.")
+	fs.StringToStringVar(&o.PeerLabels, "peer-label", nil,
+		`Peer attributes to be used in access policies. Values should have the form "<key>=<value>"`)
 }
 
 // Run the various controlplane servers.
@@ -189,7 +193,7 @@ func (o *Options) Run() error {
 	controlplaneServerListenAddress := fmt.Sprintf("0.0.0.0:%d", api.ListenPort)
 	grpcServer := grpc.NewServer("controlplane-grpc", controlplaneCertData.ServerConfig())
 
-	authzManager := authz.NewManager(mgr.GetClient(), namespace)
+	authzManager := authz.NewManager(mgr.GetClient(), namespace, o.PeerLabels)
 	peerCertsWatcher.AddConsumer(authzManager)
 
 	err = authz.CreateControllers(authzManager, mgr)

cmd/clusterlink/cmd/deploy/deploy_peer.go

@@ -79,6 +79,8 @@ type PeerOptions struct {
 	DataplaneReplicas uint16
 	// DataplaneType is the type of dataplane to create (envoy or go-based)
 	DataplaneType string
+	// Labels hold the peer attributes to be considered by access policies
+	Labels map[string]string
 	// LogLevel is the log level.
 	LogLevel string
 }
@@ -122,18 +124,21 @@ func (o *PeerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.StartInstance, "start", StartAll,
 		"Represents which component to deploy and start in the cluster: "+
 			"`all` (clusterlink components and operator), `operator`, or `none`.")
-	fs.StringVar(&o.Ingress, "ingress", string(apis.IngressTypeLoadBalancer), "Represents the type of service used"+
+	fs.StringVar(&o.Ingress, "ingress", string(apis.IngressTypeLoadBalancer), "Represents the type of service used "+
 		"to expose the ClusterLink deployment (LoadBalancer/NodePort/none).")
 	fs.Uint16Var(&o.IngressPort, "ingress-port", apis.DefaultExternalPort,
 		"Represents the ingress port. By default it is set to 443 for LoadBalancer"+
 			" and a random port in range (30000 to 32767) for NodePort.")
-	fs.StringToStringVar(&o.IngressAnnotations, "ingress-annotations", nil, "Represents the annotations that"+
+	fs.StringToStringVar(&o.IngressAnnotations, "ingress-annotations", nil, "Represents the annotations that "+
 		"will be added to ingress services.\nThe flag can be repeated to add several annotations.\n"+
 		"For example: --ingress-annotations <key1>=<value1> --ingress-annotations <key2>=<value2>.")
 	fs.StringVar(&o.DataplaneType, "dataplane", platform.DataplaneTypeEnvoy,
 		"Type of dataplane, Supported values: \"envoy\", \"go\"")
 	fs.Uint16Var(&o.ControlplaneReplicas, "controlplane-replicas", 1, "Number of controlplanes.")
 	fs.Uint16Var(&o.DataplaneReplicas, "dataplane-replicas", 1, "Number of dataplanes.")
+	fs.StringToStringVar(&o.Labels, "label", nil, "Key-value attributes to assign to the peer. "+
+		"These attributes can be used in access policies.\nThe flag can be repeated to add several attributes.\n"+
+		"For example: --label <key1>=<value1> --label <key2>=<value2>.")
 	fs.StringVar(&o.LogLevel, "log-level", "info",
 		"The log level. One of fatal, error, warn, info, debug.")
 }
@@ -193,6 +198,7 @@ func (o *PeerOptions) Run() error {
 		DataplaneCertificate:    dataplaneCert,
 		Dataplanes:              o.DataplaneReplicas,
 		DataplaneType:           o.DataplaneType,
+		PeerLabels:              o.Labels,
 		LogLevel:                o.LogLevel,
 		ContainerRegistry:       o.ContainerRegistry,
 		Namespace:               o.Namespace,

config/crds/clusterlink.net_instances.yaml

@@ -108,6 +108,12 @@ spec:
                 description: Namespace represents the namespace where the ClusterLink
                   project components are deployed.
                 type: string
+              peerLabels:
+                additionalProperties:
+                  type: string
+                description: PeerLabels holds peer attributes to be considered by
+                  access policies.
+                type: object
               tag:
                 default: latest
                 description: Tag represents the tag of the ClusterLink project images.

pkg/apis/clusterlink.net/v1alpha1/instance_types.go

@@ -125,6 +125,8 @@ type InstanceSpec struct {
 	// +kubebuilder:default="clusterlink-system"
 	// Namespace represents the namespace where the ClusterLink project components are deployed.
 	Namespace string `json:"namespace,omitempty"`
+	// PeerLabels holds peer attributes to be considered by access policies.
+	PeerLabels map[string]string `json:"peerLabels,omitempty"`
 }
 
 // +kubebuilder:object:root=true

pkg/bootstrap/platform/config.go

@@ -44,6 +44,8 @@ type Config struct {
 	// DataplaneType is the type of dataplane to create (envoy or go-based)
 	DataplaneType string
 
+	// PeerLabels are the peer attributes to be considered by access policies
+	PeerLabels map[string]string
 	// LogLevel is the log level.
 	LogLevel string
 	// ContainerRegistry is the container registry to pull the project images.

pkg/bootstrap/platform/k8s.go

@@ -102,7 +102,7 @@ spec:
       containers:
         - name: {{.controlplaneName}}
           image: {{.containerRegistry}}{{.controlplaneName}}:{{.tag}}
-          args: ["--log-level", "{{.logLevel}}"]
+          args: ["--log-level", "{{.logLevel}}" {{range $key, $val := .peerLabels }}, "--peer-label", "{{$key}}={{$val}}"{{end}}]
           imagePullPolicy: IfNotPresent
           readinessProbe:
             httpGet:
@@ -260,6 +260,7 @@ spec:
 {{ end }}
     annotations: {{.ingressAnnotations}}
   logLevel: {{.logLevel}}
+  peerLabels: {{.peerLabels}}
   containerRegistry: {{.containerRegistry}}
   namespace: {{.namespace}}
   tag: {{.tag}}
@@ -313,6 +314,7 @@ func K8SConfig(config *Config) ([]byte, error) {
 		"dataplanes":        dataplanes,
 		"dataplaneType":     dataplaneType,
 		"logLevel":          config.LogLevel,
+		"peerLabels":        config.PeerLabels,
 		"containerRegistry": containerRegistry,
 		"tag":               config.Tag,
 
@@ -400,25 +402,32 @@ func K8SCertificateConfig(config *Config) ([]byte, error) {
 	return certConfig.Bytes(), nil
 }
 
+// mapToStr dumps the key-value pairs in a map into a multi-line string.
+func mapToStr(m map[string]string, indentation string) string {
+	mapAsString := "\n"
+	for key, value := range m {
+		mapAsString += fmt.Sprintf("%s%s: %s\n", indentation, key, value)
+	}
+	return mapAsString
+}
+
 // K8SClusterLinkInstanceConfig returns a YAML file for the ClusterLink instance.
 func K8SClusterLinkInstanceConfig(config *Config, name string) ([]byte, error) {
 	containerRegistry := config.ContainerRegistry
 	if containerRegistry != "" {
 		containerRegistry = config.ContainerRegistry + "/"
 	}
 
-	// Convert ingress annotations map to string.
-	ingressAnnotationsStr := "\n"
-	for key, value := range config.IngressAnnotations {
-		ingressAnnotationsStr += fmt.Sprintf("      %s: %s\n", key, value)
-	}
+	ingressAnnotationsStr := mapToStr(config.IngressAnnotations, "      ")
+	peerLabelsStr := mapToStr(config.PeerLabels, "    ")
 
 	args := map[string]interface{}{
 		"name":               name,
 		"controlplanes":      config.Controlplanes,
 		"dataplanes":         config.Dataplanes,
 		"dataplaneType":      config.DataplaneType,
 		"logLevel":           config.LogLevel,
+		"peerLabels":         peerLabelsStr,
 		"containerRegistry":  containerRegistry,
 		"namespace":          config.Namespace,
 		"ingressType":        config.IngressType,

pkg/controlplane/authz/manager.go

@@ -47,6 +47,7 @@ const (
 	ServiceNamespaceLabel = "export.clusterlink.net/namespace"
 	ServiceLabelsPrefix   = "export.clusterlink.net/labels."
 	PeerNameLabel         = "peer.clusterlink.net/name"
+	PeerLabelsPrefix      = "peer.clusterlink.net/labels."
 )
 
 // egressAuthorizationRequest (from local dataplane)
@@ -104,6 +105,7 @@ type Manager struct {
 	selfPeerLock sync.RWMutex
 	peerTLS      *tls.ParsedCertData
 	peerName     string
+	peerLabels   map[string]string
 
 	peerClientLock sync.RWMutex
 	peerClient     map[string]*peer.Client
@@ -220,7 +222,7 @@ func (m *Manager) getPodInfoByIP(ip string) *podInfo {
 	return nil
 }
 
-func (m *Manager) getClientAttributes(req *egressAuthorizationRequest) connectivitypdp.WorkloadAttrs {
+func (m *Manager) getSrcAttributes(req *egressAuthorizationRequest) connectivitypdp.WorkloadAttrs {
 	podInfo := m.getPodInfoByIP(req.IP)
 	if podInfo == nil {
 		m.logger.Infof("Pod has no info: IP=%v.", req.IP)
@@ -237,7 +239,11 @@ func (m *Manager) getClientAttributes(req *egressAuthorizationRequest) connectiv
 		clientAttrs[ClientLabelsPrefix+k] = v
 	}
 
-	m.logger.Debugf("Client attributes: %v.", clientAttrs)
+	for k, v := range m.peerLabels {
+		clientAttrs[PeerLabelsPrefix+k] = v
+	}
+
+	m.logger.Infof("Client attributes: %v.", clientAttrs)
 
 	return clientAttrs
 }
@@ -246,7 +252,7 @@ func (m *Manager) getClientAttributes(req *egressAuthorizationRequest) connectiv
 func (m *Manager) authorizeEgress(ctx context.Context, req *egressAuthorizationRequest) (*egressAuthorizationResponse, error) {
 	m.logger.Infof("Received egress authorization request: %v.", req)
 
-	srcAttributes := m.getClientAttributes(req)
+	srcAttributes := m.getSrcAttributes(req)
 	if len(srcAttributes) == 0 && m.connectivityPDP.DependsOnClientAttrs() {
 		return nil, fmt.Errorf("failed to extract client attributes, however, access policies depend on such attributes")
 	}
@@ -295,6 +301,7 @@ func (m *Manager) authorizeEgress(ctx context.Context, req *egressAuthorizationR
 		}
 
 		if decision.Decision != connectivitypdp.DecisionAllow {
+			m.logger.Infof("PDP not allowing connection: src:%v, dst:%v, decision: %+v", srcAttributes, dstAttributes, decision)
 			continue
 		}
 
@@ -316,6 +323,7 @@ func (m *Manager) authorizeEgress(ctx context.Context, req *egressAuthorizationR
 			DstNamespace = req.ImportName.Namespace
 		}
 
+		m.logger.Infof("Egress authorized. Sending authorization request to %s", importSource.Peer)
 		accessToken, err := cl.Authorize(&cpapi.AuthorizationRequest{
 			ServiceName:      DstName,
 			ServiceNamespace: DstNamespace,
@@ -404,6 +412,7 @@ func (m *Manager) authorizeIngress(
 
 	// do not allow requests from clients with no attributes if the PDP has attribute-dependent policies
 	if len(req.SrcAttributes) == 0 && m.connectivityPDP.DependsOnClientAttrs() {
+		m.logger.Infof("PDP not allowing connection: No client attributes")
 		resp.Allowed = false
 		return resp, nil
 	}
@@ -414,6 +423,7 @@ func (m *Manager) authorizeIngress(
 	}
 
 	if decision.Decision != connectivitypdp.DecisionAllow {
+		m.logger.Infof("PDP not allowing connection: src:%v, dst:%v, decision: %+v", req.SrcAttributes, dstAttributes, decision)
 		resp.Allowed = false
 		return resp, nil
 	}
@@ -445,6 +455,7 @@ func (m *Manager) authorizeIngress(
 	}
 	resp.AccessToken = string(signed)
 
+	m.logger.Infof("Ingress authorized. Sending authorization response: %v", resp)
 	return resp, nil
 }
 
@@ -486,10 +497,11 @@ func (m *Manager) IsReady() bool {
 }
 
 // NewManager returns a new authorization manager.
-func NewManager(cl client.Client, namespace string) *Manager {
+func NewManager(cl client.Client, namespace string, peerLabels map[string]string) *Manager {
 	return &Manager{
 		client:          cl,
 		namespace:       namespace,
+		peerLabels:      peerLabels,
 		connectivityPDP: connectivitypdp.NewPDP(),
 		loadBalancer:    NewLoadBalancer(),
 		peerClient:      make(map[string]*peer.Client),

pkg/operator/controller/instance_controller.go

@@ -211,6 +211,10 @@ func (r *InstanceReconciler) applyClusterLink(ctx context.Context, instance *clu
 // applyControlplane sets up the controlplane deployment.
 func (r *InstanceReconciler) applyControlplane(ctx context.Context, instance *clusterlink.Instance) error {
 	cpDeployment := r.setDeployment(cpapi.Name, instance.Spec.Namespace, 1)
+	containerArgs := []string{"--log-level", instance.Spec.LogLevel}
+	for key, val := range instance.Spec.PeerLabels {
+		containerArgs = append(containerArgs, "--peer-label", fmt.Sprintf("%s=%s", key, val))
+	}
 	cpDeployment.Spec.Template.Spec = corev1.PodSpec{
 		ServiceAccountName: cpapi.Name,
 		Volumes: []corev1.Volume{
@@ -244,7 +248,7 @@ func (r *InstanceReconciler) applyControlplane(ctx context.Context, instance *cl
 				Name:            cpapi.Name,
 				Image:           instance.Spec.ContainerRegistry + cpapi.Name + ":" + instance.Spec.Tag,
 				ImagePullPolicy: corev1.PullIfNotPresent,
-				Args:            []string{"--log-level", instance.Spec.LogLevel},
+				Args:            containerArgs,
 				ReadinessProbe: &corev1.Probe{
 					ProbeHandler: corev1.ProbeHandler{
 						HTTPGet: &corev1.HTTPGetAction{

tests/e2e/k8s/test_policy.go

@@ -237,6 +237,33 @@ func (s *TestSuite) TestPrivilegedPolicies() {
 	require.ErrorIs(s.T(), err, &services.ConnectionResetError{})
 }
 
+func (s *TestSuite) TestPeerLabels() {
+	cl, importedService := s.createTwoClustersWithEchoSvc()
+	require.Nil(s.T(), cl[0].CreatePolicy(util.PolicyAllowAll))
+	require.Nil(s.T(), cl[1].CreatePolicy(util.PolicyAllowAll))
+
+	// 1. Denying clients whose peer sits in a cluster with a given name
+	srcLabels := map[string]string{
+		authz.PeerLabelsPrefix + util.ClusterNameLabel: cl[1].Cluster().Name(),
+	}
+	denyCl1 := util.NewPolicy("deny-cl1-cluster", v1alpha1.AccessPolicyActionDeny, srcLabels, nil)
+	require.Nil(s.T(), cl[0].CreatePolicy(denyCl1))
+	_, err := cl[1].AccessService(httpecho.RunClientInPod, importedService, false, nil)
+	require.NotNil(s.T(), err)
+
+	// 2. Creating a privileged policy to override the deny in 1. and allow requests from cl[1], based on its ip
+	srcLabels = map[string]string{
+		authz.PeerLabelsPrefix + util.PeerIPLabel: cl[1].Cluster().IP(),
+	}
+	allowCl1 := util.NewPrivilegedPolicy("allow-cl1-cluster", v1alpha1.AccessPolicyActionAllow, srcLabels, nil)
+	require.Nil(s.T(), cl[0].CreatePrivilegedPolicy(allowCl1))
+	_, err = cl[1].AccessService(httpecho.RunClientInPod, importedService, false, nil)
+	require.Nil(s.T(), err)
+
+	// privileged policies are not namespaced, so remain after the test's namespace is deleted
+	require.Nil(s.T(), cl[0].DeletePrivilegedPolicy(allowCl1.Name))
+}
+
 func (s *TestSuite) createTwoClustersWithEchoSvc() ([]*util.ClusterLink, *util.Service) {
 	cl, err := s.fabric.DeployClusterlinks(2, nil)
 	require.Nil(s.T(), err)

tests/e2e/k8s/util/k8s_yaml.go

@@ -22,6 +22,11 @@ import (
 	"github.com/clusterlink-net/clusterlink/pkg/controlplane/api"
 )
 
+const (
+	ClusterNameLabel = "cluster"
+	PeerIPLabel      = "ip"
+)
+
 // replaceOnce replaces <search> exactly once.
 func replaceOnce(s, search, replace string) (string, error) {
 	searchCount := strings.Count(s, search)
@@ -69,6 +74,10 @@ func (f *Fabric) generateK8SYAML(p *peer, cfg *PeerConfig) (string, error) {
 		ContainerRegistry:       "",
 		Namespace:               f.namespace,
 		Tag:                     "latest",
+		PeerLabels: map[string]string{
+			ClusterNameLabel: p.cluster.name,
+			PeerIPLabel:      p.cluster.ip,
+		},
 	})
 	if err != nil {
 		return "", err