Fix enabling of PSS if Calico API server is not installed (#515)

Refactoring in PSS enrichment and applying

kubemarine/admission.py

@@ -27,6 +27,7 @@
 from kubemarine.core.cluster import KubernetesCluster
 from kubemarine.core.group import NodeGroup, RunnersGroupResult
 from kubemarine.core.yaml_merger import default_merger
+from kubemarine.plugins import builtin
 
 privileged_policy_filename = "privileged.yaml"
 policies_file_path = "./resources/psp/"
@@ -45,11 +46,6 @@
 valid_modes = ['enforce', 'audit', 'warn']
 valid_versions_templ = r"^v1\.\d{1,2}$"
 
-baseline_plugins = {"kubernetes-dashboard": "kubernetes-dashboard",
-                    "calico": "calico-apiserver"}
-privileged_plugins = {"nginx-ingress-controller": "ingress-nginx", 
-                      "local-path-provisioner": "local-path-storage"}
-
 loaded_oob_policies = {}
 
 # TODO: When KubeMarine is not support Kubernetes version lower than 1.25, the PSP implementation code should be deleted 
@@ -890,45 +886,37 @@ def copy_pss(group: NodeGroup) -> Optional[RunnersGroupResult]:
     return result
 
 
+def _get_default_labels(profile: str) -> Dict[str, str]:
+    return {f"pod-security.kubernetes.io/{k}": v
+            for mode in valid_modes
+            for k, v in ((mode, profile), (f'{mode}-version', 'latest'))}
+
+
+def get_labels_to_ensure_profile(inventory: dict, profile: str) -> Dict[str, str]:
+    enforce_profile: str = inventory['rbac']['pss']['defaults']['enforce']
+    if (enforce_profile == 'restricted' and profile != 'restricted'
+            or enforce_profile == 'baseline' and profile == 'privileged'):
+        return _get_default_labels(profile)
+
+    return {}
+
+
 def label_namespace_pss(cluster: KubernetesCluster, manage_type: str) -> None:
     first_control_plane = cluster.nodes["control-plane"].get_first_member()
     # set/delete labels on predifined plugins namsespaces
-    for plugin in cluster.inventory["plugins"]:
-        is_install = cluster.inventory["plugins"][plugin].get("install")
-        if manage_type in ["apply", "install"]:
-            if is_install and plugin in privileged_plugins.keys():
-                # set label 'pod-security.kubernetes.io/enforce: privileged' for local provisioner and ingress namespaces
-                cluster.log.debug(f"Set PSS labels on namespace {privileged_plugins[plugin]}")
-                for mode in valid_modes:
-                    first_control_plane.sudo(f"kubectl label ns {privileged_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}=privileged --overwrite")
-                    first_control_plane.sudo(f"kubectl label ns {privileged_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}-version=latest --overwrite")
-            elif is_install and plugin in baseline_plugins.keys():
-                # set label 'pod-security.kubernetes.io/enforce: baseline' for kubernetes dashboard
-                cluster.log.debug(f"Set PSS labels on namespace {baseline_plugins[plugin]}")
-                for mode in valid_modes:
-                    first_control_plane.sudo(f"kubectl label ns {baseline_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}=baseline --overwrite")
-                    first_control_plane.sudo(f"kubectl label ns {baseline_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}-version=latest --overwrite")
-        elif manage_type == "delete":
-            if is_install and plugin in privileged_plugins.keys():
-                # delete label 'pod-security.kubernetes.io/enforce: privileged' for local provisioner and ingress namespaces
-                cluster.log.debug(f"Delete PSS labels from namespace {privileged_plugins[plugin]}")
-                for mode in valid_modes:
-                    first_control_plane.sudo(f"kubectl label ns {privileged_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}- || true")
-                    first_control_plane.sudo(f"kubectl label ns {privileged_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}-version- || true")
-            elif is_install and plugin in baseline_plugins.keys():
-                # delete 'pod-security.kubernetes.io/enforce: baseline' for kubernetes dashboard
-                cluster.log.debug(f"Delete PSS labels from namespace {baseline_plugins[plugin]}")
-                for mode in valid_modes:
-                    first_control_plane.sudo(f"kubectl label ns {baseline_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}- || true")
-                    first_control_plane.sudo(f"kubectl label ns {baseline_plugins[plugin]} "
-                                             f"pod-security.kubernetes.io/{mode}-version- || true")
+    for namespace, profile in builtin.get_namespace_to_necessary_pss_profiles(cluster).items():
+        target_labels = get_labels_to_ensure_profile(cluster.inventory, profile)
+        if manage_type in ["apply", "install"] and target_labels:
+            cluster.log.debug(f"Set PSS labels for profile {profile} on namespace {namespace}")
+            command = "kubectl label ns {namespace} {lk}={lv} --overwrite"
+
+        else:  # manage_type == "delete" or default labels are not necessary
+            cluster.log.debug(f"Delete PSS labels from namespace {namespace}")
+            command = "kubectl label ns {namespace} {lk}- || true"
+            target_labels = _get_default_labels(profile)
+
+        for lk, lv in target_labels.items():
+            first_control_plane.sudo(command.format(namespace=namespace, lk=lk, lv=lv))
 
     procedure_config = cluster.procedure_inventory["pss"]
     namespaces = procedure_config.get("namespaces")

kubemarine/plugins/builtin.py

@@ -32,6 +32,17 @@
 }
 
 
+def _is_manifest_enabled(inventory: dict, manifest_identity: Identity) -> bool:
+    plugin_name = manifest_identity.plugin_name
+    if not inventory["plugins"][plugin_name]["install"]:
+        return False
+
+    if manifest_identity == Identity("calico", "apiserver") and not calico.is_apiserver_enabled(inventory):
+        return False
+
+    return True
+
+
 def _get_manifest_installation_step(inventory: dict, manifest_identity: Identity) -> Optional[dict]:
     plugin_name = manifest_identity.plugin_name
     items = inventory['plugins'][plugin_name]['installation']['procedures']
@@ -61,11 +72,7 @@ def _get_manifest_installation_step(inventory: dict, manifest_identity: Identity
 
 def verify_inventory(inventory: dict, cluster: KubernetesCluster) -> dict:
     for manifest_identity, processor_provider in MANIFEST_PROCESSOR_PROVIDERS.items():
-        plugin_name = manifest_identity.plugin_name
-        if not inventory["plugins"][plugin_name]["install"]:
-            continue
-
-        if manifest_identity == Identity("calico", "apiserver") and not calico.is_apiserver_enabled(inventory):
+        if not _is_manifest_enabled(inventory, manifest_identity):
             continue
 
         config = _get_manifest_installation_step(inventory, manifest_identity)
@@ -78,7 +85,7 @@ def verify_inventory(inventory: dict, cluster: KubernetesCluster) -> dict:
             processor.validate_inventory()
         else:
             cluster.log.warning(f"Invocation of plugins.builtin.apply_yaml for {manifest_identity.repr_id()} "
-                                f"is not found for {plugin_name!r} plugin. "
+                                f"is not found for {manifest_identity.plugin_name!r} plugin. "
                                 f"Such configuration is obsolete, and support for it may be stopped in future releases.")
 
     return inventory
@@ -100,12 +107,24 @@ def apply_yaml(cluster: KubernetesCluster, plugin_name: str,
                original_yaml_path: Optional[str] = None,
                destination_name: Optional[str] = None) -> None:
     manifest_identity = Identity(plugin_name, manifest_id)
-    if manifest_identity == Identity("calico", "apiserver") and not calico.is_apiserver_enabled(cluster.inventory):
-        cluster.log.debug("Calico API server is disabled. Skip installing of the manifest.")
+    if not _is_manifest_enabled(cluster.inventory, manifest_identity):
+        cluster.log.debug(f"Skip installing of the {manifest_identity.repr_id()} for {plugin_name!r} plugin.")
         return
 
     processor = get_manifest_processor(cluster.log, cluster.inventory, manifest_identity,
                                        original_yaml_path, destination_name)
 
     manifest = processor.enrich()
     processor.apply(cluster, manifest)
+
+
+def get_namespace_to_necessary_pss_profiles(cluster: KubernetesCluster) -> Dict[str, str]:
+    result = {}
+    for manifest_identity in MANIFEST_PROCESSOR_PROVIDERS:
+        if not _is_manifest_enabled(cluster.inventory, manifest_identity):
+            continue
+
+        processor = get_manifest_processor(cluster.log, cluster.inventory, manifest_identity)
+        result.update(processor.get_namespace_to_necessary_pss_profiles())
+
+    return result

kubemarine/plugins/calico.py

@@ -418,12 +418,11 @@ def get_enrichment_functions(self) -> List[EnrichmentFunction]:
             self.enrich_clusterrole_calico_crds,
         ]
 
+    def get_namespace_to_necessary_pss_profiles(self) -> Dict[str, str]:
+        return {'calico-apiserver': 'baseline'}
+
     def enrich_namespace_calico_apiserver(self, manifest: Manifest) -> None:
-        key = "Namespace_calico-apiserver"
-        rbac = self.inventory['rbac']
-        if rbac['admission'] == 'pss' and rbac['pss']['pod-security'] == 'enabled' \
-                and rbac['pss']['defaults']['enforce'] == 'restricted':
-            self.assign_default_pss_labels(manifest, key, 'baseline')
+        self.assign_default_pss_labels(manifest, 'calico-apiserver')
 
     def enrich_deployment_calico_apiserver(self, manifest: Manifest) -> None:
         key = "Deployment_calico-apiserver"

kubemarine/plugins/kubernetes_dashboard.py

@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from typing import List, Optional
+from typing import List, Optional, Dict
 
 from kubemarine.core import summary, utils, log
 from kubemarine.core.cluster import KubernetesCluster
@@ -70,12 +70,11 @@ def get_enrichment_functions(self) -> List[EnrichmentFunction]:
             self.enrich_deployment_dashboard_metrics_scraper,
         ]
 
+    def get_namespace_to_necessary_pss_profiles(self) -> Dict[str, str]:
+        return {'kubernetes-dashboard': 'baseline'}
+
     def enrich_namespace_kubernetes_dashboard(self, manifest: Manifest) -> None:
-        key = "Namespace_kubernetes-dashboard"
-        rbac = self.inventory['rbac']
-        if rbac['admission'] == 'pss' and rbac['pss']['pod-security'] == 'enabled' \
-                and rbac['pss']['defaults']['enforce'] == 'restricted':
-            self.assign_default_pss_labels(manifest, key, 'baseline')
+        self.assign_default_pss_labels(manifest, 'kubernetes-dashboard')
 
     def enrich_deployment_kubernetes_dashboard(self, manifest: Manifest) -> None:
         key = "Deployment_kubernetes-dashboard"

kubemarine/plugins/local_path_provisioner.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from textwrap import dedent
-from typing import List, Optional
+from typing import List, Optional, Dict
 
 import yaml
 
@@ -59,12 +59,11 @@ def get_enrichment_functions(self) -> List[EnrichmentFunction]:
             self.enrich_configmap_local_path_config,
         ]
 
+    def get_namespace_to_necessary_pss_profiles(self) -> Dict[str, str]:
+        return {'local-path-storage': 'privileged'}
+
     def enrich_namespace_local_path_storage(self, manifest: Manifest) -> None:
-        key = "Namespace_local-path-storage"
-        rbac = self.inventory['rbac']
-        if rbac['admission'] == 'pss' and rbac['pss']['pod-security'] == 'enabled' \
-                and rbac['pss']['defaults']['enforce'] != 'privileged':
-            self.assign_default_pss_labels(manifest, key, 'privileged')
+        self.assign_default_pss_labels(manifest, 'local-path-storage')
 
     def add_clusterrolebinding_local_path_provisioner_privileged_psp(self, manifest: Manifest) -> None:
         # TODO add only if psp is enabled?

kubemarine/plugins/manifest.py

@@ -14,7 +14,7 @@
 
 import io
 from dataclasses import dataclass
-from typing import Callable, Optional, List, IO, Tuple, cast
+from typing import Callable, Optional, List, IO, Tuple, cast, Dict
 
 import ruamel.yaml
 import os
@@ -196,6 +196,12 @@ def get_enrichment_functions(self) -> List[EnrichmentFunction]:
         """
         pass
 
+    def get_namespace_to_necessary_pss_profiles(self) -> Dict[str, str]:
+        """
+        :return: Minimal PSS profiles for each namespace to set labels to.
+        """
+        return {}
+
     def get_version(self) -> str:
         version: str = self.inventory['plugins'][self.plugin_name]['version']
         return version
@@ -300,18 +306,21 @@ def _get_destination(self, custom_destination_name: Optional[str]) -> str:
 
         return f'{self.manifest_identity.name}-{self.get_version()}.yaml'
 
-    def assign_default_pss_labels(self, manifest: Manifest, key: str, profile: str) -> None:
-        source_yaml = manifest.get_obj(key, patch=True)
-        labels: dict = source_yaml['metadata'].setdefault('labels', {})
-        labels.update({
-            'pod-security.kubernetes.io/enforce': profile,
-            'pod-security.kubernetes.io/enforce-version': 'latest',
-            'pod-security.kubernetes.io/audit': profile,
-            'pod-security.kubernetes.io/audit-version': 'latest',
-            'pod-security.kubernetes.io/warn': profile,
-            'pod-security.kubernetes.io/warn-version': 'latest',
-        })
-        self.log.verbose(f"The {key} has been patched in 'metadata.labels' with pss labels for {profile!r} profile")
+    def assign_default_pss_labels(self, manifest: Manifest, namespace: str) -> None:
+        key = f"Namespace_{namespace}"
+        rbac = self.inventory['rbac']
+        if rbac['admission'] == 'pss' and rbac['pss']['pod-security'] == 'enabled':
+            from kubemarine import admission
+
+            profile = self.get_namespace_to_necessary_pss_profiles()[namespace]
+            target_labels = admission.get_labels_to_ensure_profile(self.inventory, profile)
+            if not target_labels:
+                return
+
+            source_yaml = manifest.get_obj(key, patch=True)
+            labels: dict = source_yaml['metadata'].setdefault('labels', {})
+            labels.update(target_labels)
+            self.log.verbose(f"The {key} has been patched in 'metadata.labels' with pss labels for {profile!r} profile")
 
     def find_container_for_patch(self, manifest: Manifest, key: str,
                                  *,

kubemarine/plugins/nginx_ingress.py

@@ -14,7 +14,7 @@
 
 import io
 import ipaddress
-from typing import Optional, List
+from typing import Optional, List, Dict
 
 from kubemarine.core import utils, log
 from kubemarine.core.cluster import KubernetesCluster
@@ -216,12 +216,11 @@ def get_enrichment_functions(self) -> List[EnrichmentFunction]:
             self.enrich_service_ingress_nginx_controller,
         ]
 
+    def get_namespace_to_necessary_pss_profiles(self) -> Dict[str, str]:
+        return {'ingress-nginx': 'privileged'}
+
     def enrich_namespace_ingress_nginx(self, manifest: Manifest) -> None:
-        key = "Namespace_ingress-nginx"
-        rbac = self.inventory['rbac']
-        if rbac['admission'] == 'pss' and rbac['pss']['pod-security'] == 'enabled' \
-                and rbac['pss']['defaults']['enforce'] != 'privileged':
-            self.assign_default_pss_labels(manifest, key, 'privileged')
+        self.assign_default_pss_labels(manifest, 'ingress-nginx')
 
     def enrich_configmap_ingress_nginx_controller(self, manifest: Manifest) -> None:
         key = "ConfigMap_ingress-nginx-controller"