test_secrets_consistency: cleanup the code after ben review, add subPath

support

Author: gaelgatelement

tests/manifests/test_secrets_consistency.py

@@ -22,6 +22,18 @@ def get_configmap(templates, configmap_name):
     raise ValueError(f"ConfigMap {configmap_name} not found")
 
 
+def get_secret(templates, secret_name):
+    """
+    Get the content of a Secret with the given name.
+    :param secret_name: The name of the Secret to retrieve.
+    :return: A string containing the content of the Secret, or an empty string if not found.
+    """
+    for t in templates:
+        if t["kind"] == "Secret" and t["metadata"]["name"] == secret_name:
+            return t
+    raise ValueError(f"Secret {secret_name} not found")
+
+
 def get_volume_from_mount(template, volume_mount):
     """
     Get a specific volume mount from a given template.
@@ -48,7 +60,6 @@ async def test_secrets_consistency(templates):
     This test checks if each secret is correctly associated with its respective volume and container,
     ensuring that no inconsistencies or missing configurations exist.
     """
-    secrets = [t for t in templates if t["kind"] == "Secret"]
     workloads = [t for t in templates if t["kind"] in ("Deployment", "StatefulSet")]
     for template in workloads:
         # Gather all containers and initContainers from the template spec
@@ -58,47 +69,60 @@ async def test_secrets_consistency(templates):
 
         for container in containers:
             # Determine which secrets are mounted by this container
-            mounted_secrets = []
+            mounted_secret_keys = []
             mounted_config_maps = []
+            secrets_mount_paths = []
+            uses_rendered_config = False
 
             for volume_mount in container.get("volumeMounts", []):
                 current_volume = get_volume_from_mount(template, volume_mount)
                 if "secret" in current_volume:
                     # Extract the paths where this volume's secrets are mounted
-                    for secret in secrets:
-                        if current_volume["secret"]["secretName"] == secret["metadata"]["name"]:
-                            # When secret data is empty, `data:` is None, so use `get_or_empty`
-                            for key in get_or_empty(secret, "data"):
-                                mounted_path = f"{volume_mount['mountPath']}/{key}"
-                                mounted_secrets.append(mounted_path)
-                            break
+                    secret = get_secret(templates, current_volume["secret"]["secretName"])
+                    if "subPath" in volume_mount:
+                        # When using subPath, the key is mounted as the mountPath itself
+                        mounted_secret_keys.append(f"{volume_mount['mountPath']}")
                     else:
-                        raise ValueError(
-                            f"Secret name '{current_volume['secret']['secretName']}' does not match any secret"
-                        )
+                        # When secret data is empty, `data:` is None, so use `get_or_empty`
+                        for key in get_or_empty(secret, "data"):
+                            # Without subPath, the key will be present as child of the mount path
+                            mounted_path = f"{volume_mount['mountPath']}/{key}"
+                            mounted_secret_keys.append(mounted_path)
+                    secrets_mount_paths.append(volume_mount["mountPath"])
                 elif "configMap" in current_volume:
                     # Parse config map content
                     mounted_config_maps.append(get_configmap(templates, current_volume["configMap"]["name"]))
+                elif "emptyDir" in current_volume and current_volume["name"] == "rendered-config":
+                    # We can't verify rendered-config, it's generated at runtime
+                    uses_rendered_config = True
 
-            for volume_mount in container.get("volumeMounts", []):
-                # Only consider volumes that are secrets
-                current_volume = get_volume_from_mount(template, volume_mount)
-                if "secret" not in current_volume:
-                    continue
-                # Parse container commands to find potential mounted secrets
-                # Make sure that potential mounted secrets are present in mounted secrets
-                for matches in re.findall(
-                    rf"{volume_mount['mountPath']}/([A-Za-z0-9._]+)", "\n".join(container.get("command", []))
-                ):
-                    assert f"{volume_mount['mountPath']}/{matches}" in mounted_secrets, (
-                        f"{volume_mount['mountPath']}/{matches} used in container {container['name']} "
+            # We look for all secrets mountPath in configs and commands
+            # And using a regex, make sure that patterns `<mount path>/<some key>`
+            # refers <some key> to an existing mounted secret key
+            for mount_path in set(secrets_mount_paths):
+                mount_path_found = False
+                # Parse container commands to find paths which would match a mounted secret
+                # Make sure that paths which match are actually present in mounted secrets
+                for matches in re.findall(rf"{mount_path}/([^\s\n);]+)", "\n".join(container.get("command", []))):
+                    assert f"{mount_path}/{matches}" in mounted_secret_keys, (
+                        f"{mount_path}/{matches} used in container {container['name']} "
                         + "but it is not found from any mounted secret"
                     )
+                    mount_path_found = True
+                # Parse container configmaps to find paths which would match a mounted secret
+                # Make sure that paths which match are actually present in mounted secrets
                 for cm in mounted_config_maps:
                     for data, content in cm["data"].items():
-                        for matches in re.findall(rf"{volume_mount['mountPath']}/([A-Za-z0-9._]+)", content):
-                            assert f"{volume_mount['mountPath']}/{matches}" in mounted_secrets, (
-                                f"{volume_mount['mountPath']}/{matches} used in config {cm['metadata']['name']}/{data} "
+                        for matches in re.findall(rf"{mount_path}/([^\s\n);]+)", content):
+                            assert f"{mount_path}/{matches}" in mounted_secret_keys, (
+                                f"{mount_path}/{matches} used in "
+                                f"config {cm['metadata']['name']}/{data} "
                                 f"mounted in container {container['name']} "
                                 + "but it is not found from any mounted secret"
                             )
+                            mount_path_found = True
+                if not mount_path_found and not uses_rendered_config:
+                    raise AssertionError(
+                        f"{volume_mount['mountPath']} used in container {container['name']} "
+                        "but no config or command is using it"
+                    )