[confcom] Derive image platform and add --platform validation

Author: MahatiC

src/confcom/azext_confcom/security_policy.py

@@ -5,6 +5,7 @@
 
 import copy
 import json
+import subprocess
 import warnings
 from enum import Enum, auto
 from typing import Any, Dict, List, Optional, Tuple, Union
@@ -41,8 +42,9 @@
                                          process_fragment_imports,
                                          process_mounts,
                                          process_mounts_from_config,
-                                         readable_diff)
-from azext_confcom.lib.images import get_image_platform  # pylint: disable=unused-import
+                                         readable_diff,
+                                         find_value_in_params_and_vars)
+from azext_confcom.lib.images import get_image_platform
 from azext_confcom.lib.defaults import get_debug_mode_exec_procs
 from knack.log import get_logger
 from tqdm import tqdm
@@ -669,6 +671,76 @@ def set_images(self, images: List[ContainerImage]) -> None:
         self._images = images
 
 
+def _get_image_platforms_from_docker(image_name: str) -> List[str]:
+    """Detect all platforms an image supports using docker manifest inspect.
+
+    If the image reference is a digest (@sha256:...), strip it back to the
+    tag so that multi-platform manifest list lookups work correctly.
+    """
+    # Strip digest references — manifest inspect needs a tag, not a digest
+    if "@sha256:" in image_name:
+        image_name = image_name.split("@sha256:")[0]
+        if ":" not in image_name.split("/")[-1]:
+            image_name += ":latest"
+
+    try:
+        result = subprocess.run(
+            ["docker", "manifest", "inspect", image_name],
+            capture_output=True, text=True, timeout=30, check=True,
+        )
+        data = json.loads(result.stdout)
+        platforms = []
+        for manifest in data.get("manifests", []):
+            plat = manifest.get("platform", {})
+            os_name = plat.get("os", "unknown")
+            arch = plat.get("architecture", "unknown")
+            if os_name != "unknown" and arch != "unknown":
+                platforms.append(f"{os_name}/{arch}")
+        if platforms:
+            return platforms
+    except (subprocess.CalledProcessError, subprocess.TimeoutExpired,
+            json.JSONDecodeError, FileNotFoundError):
+        pass
+    return []
+
+
+def validate_image_platform(image_name: str, platform: str) -> None:
+    """Validate that the image supports the specified platform.
+
+    Uses docker manifest inspect for multi-platform detection,
+    then falls back to get_image_platform (docker pull) for single-platform.
+    """
+    supported = _get_image_platforms_from_docker(image_name)
+
+    # Fall back to single-platform detection via docker pull
+    if not supported:
+        try:
+            detected = get_image_platform(image_name)
+            if detected:
+                supported = [detected]
+        except (ValueError, KeyError, AttributeError):
+            pass
+
+    if not supported:
+        logger.warning(
+            "Could not detect supported platforms for image '%s'. "
+            "Skipping platform validation.", image_name,
+        )
+        return
+
+    if len(supported) == 1 and supported[0] != platform:
+        eprint(
+            f'Image "{image_name}" only supports platform "{supported[0]}", '
+            f'which does not match the specified platform "{platform}".'
+        )
+
+    if len(supported) > 1 and platform not in supported:
+        eprint(
+            f'Image "{image_name}" supports platforms {supported}, '
+            f'which does not include the specified platform "{platform}".'
+        )
+
+
 # pylint: disable=R0914,
 def load_policy_from_arm_template_str(
     template_data: str,
@@ -830,6 +902,10 @@ def load_policy_from_arm_template_str(
                     f'Field ["{config.ACI_FIELD_TEMPLATE_IMAGE}"] is empty or cannot be found'
                 )
 
+            # Resolve ARM parameters/variables to get the real image name for validation
+            resolved_image = find_value_in_params_and_vars(all_params, all_vars, image_name)
+            validate_image_platform(resolved_image, platform)
+
             exec_processes = []
             extract_probe(exec_processes, image_properties, config.ACI_FIELD_CONTAINERS_READINESS_PROBE)
             extract_probe(exec_processes, image_properties, config.ACI_FIELD_CONTAINERS_LIVENESS_PROBE)
@@ -920,6 +996,8 @@ def load_policy_from_image_name(
 
     containers = []
     for image_name in image_names:
+        validate_image_platform(image_name, platform)
+
         container = {}
         # assign just the fields that are expected
         # the values will come when calling
@@ -1037,6 +1115,8 @@ def load_policy_from_json(
                 f'Field ["{config.ACI_FIELD_TEMPLATE_IMAGE}"] is empty or cannot be found'
             )
 
+        validate_image_platform(image_name, platform)
+
         container_name = case_insensitive_dict_get(
             container, config.ACI_FIELD_CONTAINERS_NAME
         ) or image_name
@@ -1250,6 +1330,8 @@ def load_policy_from_virtual_node_yaml_str(
             if not image:
                 eprint("Container does not have an image field")
 
+            validate_image_platform(image, platform)
+
             # env vars
             envs = process_env_vars_from_yaml(
                 container,