From 79b1d56363a74494416137721cd1703f52d13be0 Mon Sep 17 00:00:00 2001 From: Roja Reddy Sareddy Date: Tue, 4 Aug 2026 17:10:12 -0700 Subject: [PATCH 1/2] Bug fix in Hyperpod recipe rendering --- src/sagemaker/hyperpod/cli/recipe_utils.py | 5 +- .../test_recipe_utils_template_rendering.py | 137 ++++++++++++++++++ 2 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 test/unit_tests/cli/test_recipe_utils_template_rendering.py diff --git a/src/sagemaker/hyperpod/cli/recipe_utils.py b/src/sagemaker/hyperpod/cli/recipe_utils.py index 8306d34f..63aa10c8 100644 --- a/src/sagemaker/hyperpod/cli/recipe_utils.py +++ b/src/sagemaker/hyperpod/cli/recipe_utils.py @@ -6,7 +6,7 @@ import click import boto3 import sys -from jinja2 import Template +from jinja2.sandbox import SandboxedEnvironment from kubernetes import client, config from pathlib import Path from typing import Dict, Any, Tuple, Optional @@ -320,7 +320,8 @@ def _submit_k8s_resources(custom_api, rendered_yaml: str) -> None: def _render_k8s_template(template_content: str, config_data: Dict[str, Any]) -> str: """Render Kubernetes template with configuration data.""" - template = Template(template_content) + env = SandboxedEnvironment() + template = env.from_string(template_content) return template.render(**config_data) diff --git a/test/unit_tests/cli/test_recipe_utils_template_rendering.py b/test/unit_tests/cli/test_recipe_utils_template_rendering.py new file mode 100644 index 00000000..a9667504 --- /dev/null +++ b/test/unit_tests/cli/test_recipe_utils_template_rendering.py @@ -0,0 +1,137 @@ +"""Unit tests for recipe_utils template rendering. + +These tests verify that the Jinja2 template rendering in _render_k8s_template +uses a sandboxed environment that restricts template operations to safe +constructs while still allowing legitimate template functionality. +""" + +import pytest +from jinja2.exceptions import SecurityError + +from sagemaker.hyperpod.cli.recipe_utils import _render_k8s_template + + +class TestRenderK8sTemplate: + """Tests for _render_k8s_template sandboxed rendering.""" + + def test_variable_substitution(self): + """Normal variable substitution should work.""" + template = "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: {{ name }}\n namespace: {{ namespace }}" + config = {"name": "my-job", "namespace": "default"} + result = _render_k8s_template(template, config) + assert "name: my-job" in result + assert "namespace: default" in result + + def test_filter_usage(self): + """Jinja2 filters should work normally.""" + template = "name: {{ name | upper }}" + config = {"name": "my-job"} + result = _render_k8s_template(template, config) + assert "name: MY-JOB" in result + + def test_conditional(self): + """Jinja2 conditionals should work normally.""" + template = "{% if gpu %}accelerator: nvidia{% endif %}" + config = {"gpu": True} + result = _render_k8s_template(template, config) + assert "accelerator: nvidia" in result + + def test_loop(self): + """Jinja2 loops should work normally.""" + template = "{% for item in items %}{{ item }}\n{% endfor %}" + config = {"items": ["a", "b", "c"]} + result = _render_k8s_template(template, config) + assert "a" in result + assert "b" in result + assert "c" in result + + def test_blocks_os_popen_via_cycler(self): + """Access to os.popen through object traversal should be blocked.""" + template = "{{ cycler.__init__.__globals__.os.popen('whoami').read() }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_dunder_init_access(self): + """Access to __init__ on objects should be blocked.""" + template = "{{ ''.__class__.__init__.__globals__ }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_dunder_class_chaining(self): + """Chaining from __class__ to internal attributes should be blocked.""" + template = "{{ ''.__class__.__init__.__globals__ }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_subclasses_access(self): + """Access to __subclasses__ should be blocked.""" + template = "{{ ''.__class__.__mro__[1].__subclasses__() }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_globals_access(self): + """Access to __globals__ should be blocked.""" + template = "{{ config.__init__.__globals__['os'] }}" + config = {"config": {}} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_import_via_builtins(self): + """Attempt to access modules via __builtins__ should be blocked.""" + template = ( + "{{ ''.__class__.__init__.__globals__['__builtins__']['__import__']('os').popen('id').read() }}" + ) + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_mro_traversal(self): + """MRO traversal to reach internal classes should be blocked.""" + template = "{{ [].__class__.__mro__ }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_realistic_k8s_template(self): + """A realistic Kubernetes manifest template should render correctly.""" + template = """apiVersion: batch/v1 +kind: Job +metadata: + name: {{ name }} + namespace: {{ namespace }} +spec: + template: + spec: + containers: + - name: training + image: {{ image }} + resources: + limits: + nvidia.com/gpu: {{ gpu_count }} + restartPolicy: Never""" + config = { + "name": "sft-llama-job", + "namespace": "hyperpod", + "image": "763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.0", + "gpu_count": 8, + } + result = _render_k8s_template(template, config) + assert "name: sft-llama-job" in result + assert "namespace: hyperpod" in result + assert "nvidia.com/gpu: 8" in result + assert "763104351884" in result + + def test_blocks_code_execution_in_template(self): + """Arbitrary code execution through template internals should be blocked.""" + template = ( + 'apiVersion: v1\nkind: ConfigMap\nmetadata:\n' + ' name: {{ cycler.__init__.__globals__.os.popen(\'echo pwned\').read() }}\n' + ' namespace: {{ namespace }}' + ) + config = {"namespace": "default"} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) From 7363b21f8e1e5424215b6d83c15a91ebdcb0cbfc Mon Sep 17 00:00:00 2001 From: Roja Reddy Sareddy Date: Wed, 5 Aug 2026 16:42:05 -0700 Subject: [PATCH 2/2] Bug fix in Hyperpod recipe rendering --- .../test_recipe_utils_template_rendering.py | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) diff --git a/test/unit_tests/cli/test_recipe_utils_template_rendering.py b/test/unit_tests/cli/test_recipe_utils_template_rendering.py index a9667504..93042620 100644 --- a/test/unit_tests/cli/test_recipe_utils_template_rendering.py +++ b/test/unit_tests/cli/test_recipe_utils_template_rendering.py @@ -135,3 +135,204 @@ def test_blocks_code_execution_in_template(self): config = {"namespace": "default"} with pytest.raises(SecurityError): _render_k8s_template(template, config) + + # ------------------------------------------------------------------- + # Additional sandbox bypass vector tests + # + # The SandboxedEnvironment blocks unsafe attribute access via two mechanisms: + # 1. Dot notation (e.g., obj.__init__): raises SecurityError when the sandbox + # detects access to internal attributes on real objects. + # 2. |attr() filter (e.g., obj|attr("__init__")): returns Undefined (renders + # as empty string) for internal attributes. This is safe because Undefined + # cannot be called, iterated, or used to access further attributes. + # + # Both mechanisms prevent exploitation. Tests below verify neither path + # leaks internal state or enables code execution. + # ------------------------------------------------------------------- + + def test_blocks_attr_filter_returns_undefined(self): + """The |attr() filter returns Undefined for internal attributes. + + {{ ""|attr("__class__") }} uses the attr filter to access dunder + attributes indirectly. The sandbox intercepts this and returns Undefined + which renders as an empty string — no data leak occurs. + """ + template = '{{ ""|attr("__class__")|attr("__init__")|attr("__globals__") }}' + config = {} + result = _render_k8s_template(template, config) + # Should render as empty string (Undefined) — no internal data leaked + assert result == "" + assert "function" not in result + assert "module" not in result + assert "os" not in result + + def test_attr_filter_cannot_reach_callable(self): + """The attr() chain cannot produce a callable to execute code. + + Even though attr() doesn't raise, the resulting Undefined cannot be + called, preventing actual exploitation. + """ + from jinja2.exceptions import UndefinedError + # Attempting to call the result of attr() on an Undefined raises UndefinedError + template = '{{ ""|attr("__class__")|attr("__subclasses__")() }}' + config = {} + with pytest.raises(UndefinedError): + _render_k8s_template(template, config) + + def test_attr_filter_get_method_blocked(self): + """Cannot use .get() on Undefined returned by attr() to extract values.""" + from jinja2.exceptions import UndefinedError + template = '{% set g = ""|attr("__class__")|attr("__init__")|attr("__globals__") %}{{ g.get("os") }}' + config = {} + with pytest.raises(UndefinedError): + _render_k8s_template(template, config) + + def test_blocks_lipsum_globals_access(self): + """Access to __globals__ via lipsum builtin should be blocked. + + lipsum is a built-in Jinja2 global (like cycler, joiner, namespace); + it can be used as an entry point for traversal. + """ + template = "{{ lipsum.__globals__['os'].popen('id').read() }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_joiner_globals_access(self): + """Access to __globals__ via joiner builtin should be blocked.""" + template = "{{ joiner.__init__.__globals__['os'].popen('id').read() }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_namespace_init_globals(self): + """Access to __globals__ via namespace builtin should be blocked.""" + template = "{{ namespace.__init__.__globals__ }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_config_data_object_traversal(self): + """Objects passed via config_data should not allow internal traversal.""" + template = "{{ obj.__class__.__init__.__globals__ }}" + config = {"obj": {"key": "value"}} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_format_string_attr_construction_is_safe(self): + """Format string tricks to construct dunder names are blocked by attr(). + + Even if you dynamically build "__class__" via string concat and pass it + to |attr(), the sandbox still returns Undefined — no data leak. + """ + template = '{% set name = "__cla" ~ "ss__" %}{{ ""|attr(name) }}' + config = {} + result = _render_k8s_template(template, config) + assert result == "" + assert "class" not in result.lower() or result == "" + + def test_format_constructed_full_chain_is_safe(self): + """Dynamically constructed attr names cannot bypass sandbox protection.""" + template = '{% set a = "__ini" %}{% set b = "t__" %}{{ ""|attr("__class__")|attr(a~b)|attr("__globals__") }}' + config = {} + result = _render_k8s_template(template, config) + assert result == "" + assert "os" not in result + assert "module" not in result + + def test_map_filter_attribute_returns_undefined(self): + """The |map(attribute=) filter returns Undefined for dunder access. + + {{ items|map(attribute="__class__")|list }} uses filter parameters + to perform attribute access, but sandbox blocks it returning Undefined. + """ + template = '{{ ["a","b"]|map(attribute="__class__")|list }}' + config = {} + result = _render_k8s_template(template, config) + # map with blocked attribute returns Undefined for each item + assert "str" not in result + assert "type" not in result + + def test_blocks_map_filter_deep_attribute(self): + """Deep attribute access via |map should be blocked.""" + template = '{{ ["a"]|map(attribute="__class__.__init__.__globals__")|list }}' + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_set_tag_namespace_globals(self): + """{% set %} assignment with namespace.__init__.__globals__ should be blocked. + + Self-referential trick using set to assign internal objects. + """ + template = "{% set x = namespace.__init__.__globals__ %}{{ x }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_set_with_cycler_traversal(self): + """{% set %} combined with builtin object traversal should be blocked.""" + template = "{% set x = cycler.__init__.__globals__ %}{{ x }}" + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_join_filter_attr_construction_is_safe(self): + """Using |join to construct attribute names still returns Undefined.""" + template = '{% set parts = ["__","class","__"] %}{{ ""|attr(parts|join) }}' + config = {} + result = _render_k8s_template(template, config) + assert result == "" + + def test_getitem_dunder_returns_undefined(self): + """Accessing __class__ via bracket notation returns Undefined (safe). + + The sandbox intercepts __getitem__ access to internal attribute names + and returns Undefined rather than the actual attribute. + """ + template = '{{ ""["__class__"] }}' + config = {} + result = _render_k8s_template(template, config) + # Bracket notation for dunder attrs also returns Undefined (renders empty) + assert "str" not in result + assert "type" not in result + + def test_getitem_cannot_chain_to_globals(self): + """Bracket notation chaining to reach __globals__ raises SecurityError.""" + template = '{{ ""["__class__"]["__init__"]["__globals__"] }}' + config = {} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_blocks_request_like_object_traversal(self): + """If any complex objects are passed in config, traversal should be blocked.""" + class FakeRequest: + pass + + template = "{{ req.__class__.__init__.__globals__ }}" + config = {"req": FakeRequest()} + with pytest.raises(SecurityError): + _render_k8s_template(template, config) + + def test_attr_filter_no_data_leak_comprehensive(self): + """Comprehensive test that no attr() variant leaks internal Python state.""" + dangerous_templates = [ + '{{ ""|attr("__class__")|attr("__init__")|attr("__globals__") }}', + '{{ []|attr("__class__")|attr("__mro__") }}', + '{{ ""|attr("__class__")|attr("__subclasses__") }}', + '{% set x = ""|attr("__class__") %}{{ x|attr("__init__") }}', + ] + dangerous_indicators = ["function", "module", "os", "subprocess", "popen", + "builtins", "