fix(server,k8s): propagate template container securityContext to pods

server/opensandbox_server/services/k8s/batchsandbox_provider.py

@@ -61,6 +61,26 @@
 logger = logging.getLogger(__name__)
 
 
+def _merge_security_context(
+    template_sc: Dict[str, Any], runtime_sc: Dict[str, Any]
+) -> Dict[str, Any]:
+    """Merge the template's container securityContext into the runtime one.
+
+    Nested dicts (capabilities, seccompProfile, ...) merge recursively so a
+    template member on one key (e.g. capabilities.add) survives even when the
+    runtime populates another key of the same field (e.g. capabilities.drop from
+    network-policy wiring). On actual conflicting leaves, the runtime value wins.
+    """
+    merged = dict(template_sc)
+    for key, runtime_value in runtime_sc.items():
+        template_value = merged.get(key)
+        if isinstance(runtime_value, dict) and isinstance(template_value, dict):
+            merged[key] = _merge_security_context(template_value, runtime_value)
+        else:
+            merged[key] = runtime_value
+    return merged
+
+
 class BatchSandboxProvider(WorkloadProvider):
     """Workload provider for BatchSandbox CRDs."""
 
@@ -163,7 +183,7 @@ def create_workload(
                 annotations=annotations,
             )
 
-        extra_volumes, extra_mounts = self._extract_template_pod_extras()
+        extra_volumes, extra_mounts, extra_security_context = self._extract_template_pod_extras()
 
         if windows_profile:
             validate_windows_profile_resource_limits(resource_limits)
@@ -289,7 +309,9 @@ def create_workload(
             batchsandbox["spec"].pop("expireTime", None)
         else:
             batchsandbox["spec"]["expireTime"] = expires_at.isoformat()
-        self._merge_pod_spec_extras(batchsandbox, extra_volumes, extra_mounts)
+        self._merge_pod_spec_extras(
+            batchsandbox, extra_volumes, extra_mounts, extra_security_context
+        )
         merged_pod_spec = batchsandbox.get("spec", {}).get("template", {}).get("spec", {})
         ensure_egress_runtime_compatible(
             network_policy,
@@ -424,14 +446,17 @@ def _create_workload_from_pool(
             "kind": "BatchSandbox",
         }
 
-    def _extract_template_pod_extras(self) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]]]:
-        """Extract extra template volumes and mounts for runtime merge."""
+    def _extract_template_pod_extras(
+        self,
+    ) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]], Optional[Dict[str, Any]]]:
+        """Extract extra template volumes, mounts, and container securityContext for runtime merge."""
         template = self.template_manager.get_base_template()
         spec = template.get("spec", {}) if isinstance(template, dict) else {}
         template_spec = spec.get("template", {}).get("spec", {})
         extra_volumes = template_spec.get("volumes", []) or []
 
         extra_mounts: list[Dict[str, Any]] = []
+        extra_security_context: Optional[Dict[str, Any]] = None
         containers = template_spec.get("containers", []) or []
         if containers:
             target = None
@@ -442,20 +467,24 @@ def _extract_template_pod_extras(self) -> tuple[list[Dict[str, Any]], list[Dict[
             if target is None:
                 target = containers[0]
             extra_mounts = target.get("volumeMounts", []) or []
+            security_context = target.get("securityContext")
+            if isinstance(security_context, dict):
+                extra_security_context = security_context
 
         if not isinstance(extra_volumes, list):
             extra_volumes = []
         if not isinstance(extra_mounts, list):
             extra_mounts = []
-        return extra_volumes, extra_mounts
+        return extra_volumes, extra_mounts, extra_security_context
 
     def _merge_pod_spec_extras(
         self,
         batchsandbox: Dict[str, Any],
         extra_volumes: list[Dict[str, Any]],
         extra_mounts: list[Dict[str, Any]],
+        extra_security_context: Optional[Dict[str, Any]] = None,
     ) -> None:
-        """Merge template-provided volumes and mounts into runtime pod spec."""
+        """Merge template-provided volumes, mounts, and securityContext into runtime pod spec."""
         try:
             spec = batchsandbox["spec"]["template"]["spec"]
         except KeyError:
@@ -478,6 +507,18 @@ def _merge_pod_spec_extras(
         if not containers or not isinstance(containers, list):
             return
         main_container = containers[0]
+        if extra_security_context and isinstance(main_container, dict):
+            # The template's container securityContext is a base default: merge it
+            # into the runtime container's own securityContext (runtime leaves win,
+            # nested dicts merge so template members like capabilities.add survive),
+            # and fill the whole context when the runtime sets none.
+            runtime_security_context = main_container.get("securityContext")
+            if isinstance(runtime_security_context, dict):
+                main_container["securityContext"] = _merge_security_context(
+                    extra_security_context, runtime_security_context
+                )
+            else:
+                main_container["securityContext"] = extra_security_context
         mounts = main_container.get("volumeMounts", []) or []
         if isinstance(mounts, list) and extra_mounts:
             existing = {m.get("name") for m in mounts if isinstance(m, dict)}

server/tests/k8s/test_batchsandbox_provider.py

@@ -644,6 +644,198 @@ def test_create_workload_dedupes_template_volume_and_mount_names(
         assert mount_names.count("opensandbox-bin") == 1
         assert "sandbox-shared-data" in mount_names
 
+    def test_create_workload_applies_template_container_security_context(self, mock_k8s_client, tmp_path):
+        template_file = tmp_path / "template.yaml"
+        template_file.write_text(
+            """
+spec:
+  template:
+    spec:
+      containers:
+        - name: sandbox
+          image: ubuntu:latest
+          securityContext:
+            runAsNonRoot: true
+            seccompProfile:
+              type: Unconfined
+"""
+        )
+        provider = BatchSandboxProvider(
+            mock_k8s_client, _app_config_with_template(str(template_file))
+        )
+        mock_k8s_client.create_custom_object.return_value = {
+            "metadata": {"name": "sandbox-test", "uid": "uid"}
+        }
+
+        provider.create_workload(
+            sandbox_id="test-id",
+            namespace="test-ns",
+            image_spec=ImageSpec(uri="python:3.11"),
+            entrypoint=["/bin/bash"],
+            env={},
+            resource_limits={},
+            labels={},
+            expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
+            execd_image="execd:latest",
+        )
+
+        body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
+        container = body["spec"]["template"]["spec"]["containers"][0]
+
+        # Template image must not override the runtime image, but its container
+        # securityContext should be propagated to the generated Pod.
+        assert container["name"] == "sandbox"
+        assert container["image"] == "python:3.11"
+        assert container["securityContext"] == {
+            "runAsNonRoot": True,
+            "seccompProfile": {"type": "Unconfined"},
+        }
+
+    def test_create_workload_merges_template_security_context_with_runtime_network_policy(
+        self, mock_k8s_client, tmp_path
+    ):
+        template_file = tmp_path / "template.yaml"
+        template_file.write_text(
+            """
+spec:
+  template:
+    spec:
+      containers:
+        - name: sandbox
+          image: ubuntu:latest
+          securityContext:
+            runAsNonRoot: true
+"""
+        )
+        provider = BatchSandboxProvider(
+            mock_k8s_client, _app_config_with_template(str(template_file))
+        )
+        mock_k8s_client.create_custom_object.return_value = {
+            "metadata": {"name": "sandbox-test", "uid": "uid"}
+        }
+
+        provider.create_workload(
+            sandbox_id="test-id",
+            namespace="test-ns",
+            image_spec=ImageSpec(uri="python:3.11"),
+            entrypoint=["/bin/bash"],
+            env={},
+            resource_limits={},
+            labels={},
+            expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
+            execd_image="execd:latest",
+            network_policy=NetworkPolicy(default_action="deny", egress=[]),
+            egress_image="opensandbox/egress:v1.1.6",
+        )
+
+        body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
+        container = body["spec"]["template"]["spec"]["containers"][0]
+
+        # Runtime-provided securityContext keys (network policy capabilities) win;
+        # template-provided keys supplement rather than replace them.
+        assert container["securityContext"] == {
+            "runAsNonRoot": True,
+            "capabilities": {"drop": ["NET_ADMIN"]},
+        }
+
+    def test_create_workload_merges_template_nested_capabilities_with_runtime_network_policy(
+        self, mock_k8s_client, tmp_path
+    ):
+        template_file = tmp_path / "template.yaml"
+        template_file.write_text(
+            """
+spec:
+  template:
+    spec:
+      containers:
+        - name: sandbox
+          image: ubuntu:latest
+          securityContext:
+            capabilities:
+              add:
+                - SYS_PTRACE
+"""
+        )
+        provider = BatchSandboxProvider(
+            mock_k8s_client, _app_config_with_template(str(template_file))
+        )
+        mock_k8s_client.create_custom_object.return_value = {
+            "metadata": {"name": "sandbox-test", "uid": "uid"}
+        }
+
+        provider.create_workload(
+            sandbox_id="test-id",
+            namespace="test-ns",
+            image_spec=ImageSpec(uri="python:3.11"),
+            entrypoint=["/bin/bash"],
+            env={},
+            resource_limits={},
+            labels={},
+            expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
+            execd_image="execd:latest",
+            network_policy=NetworkPolicy(default_action="deny", egress=[]),
+            egress_image="opensandbox/egress:v1.1.6",
+        )
+
+        body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
+        container = body["spec"]["template"]["spec"]["containers"][0]
+
+        # Nested dicts merge: the template's capabilities.add survives even though
+        # network-policy wiring populates a different member (capabilities.drop).
+        assert container["securityContext"] == {
+            "capabilities": {
+                "add": ["SYS_PTRACE"],
+                "drop": ["NET_ADMIN"],
+            },
+        }
+
+    def test_create_workload_runtime_capabilities_win_over_template_conflicts(
+        self, mock_k8s_client, tmp_path
+    ):
+        template_file = tmp_path / "template.yaml"
+        template_file.write_text(
+            """
+spec:
+  template:
+    spec:
+      containers:
+        - name: sandbox
+          image: ubuntu:latest
+          securityContext:
+            capabilities:
+              drop:
+                - ALL
+"""
+        )
+        provider = BatchSandboxProvider(
+            mock_k8s_client, _app_config_with_template(str(template_file))
+        )
+        mock_k8s_client.create_custom_object.return_value = {
+            "metadata": {"name": "sandbox-test", "uid": "uid"}
+        }
+
+        provider.create_workload(
+            sandbox_id="test-id",
+            namespace="test-ns",
+            image_spec=ImageSpec(uri="python:3.11"),
+            entrypoint=["/bin/bash"],
+            env={},
+            resource_limits={},
+            labels={},
+            expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
+            execd_image="execd:latest",
+            network_policy=NetworkPolicy(default_action="deny", egress=[]),
+            egress_image="opensandbox/egress:v1.1.6",
+        )
+
+        body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
+        container = body["spec"]["template"]["spec"]["containers"][0]
+
+        # Conflicting leaves keep the runtime value (network-policy requirement).
+        assert container["securityContext"] == {
+            "capabilities": {"drop": ["NET_ADMIN"]},
+        }
+
     def test_create_workload_sets_resource_limits_and_requests(self, mock_k8s_client):
         provider = BatchSandboxProvider(mock_k8s_client)
         mock_k8s_client.create_custom_object.return_value = {