refactor: replace str.format with safe_jinja2/safe_str_format (#2315)

Author: piglei

apiserver/paasng/paasng/accessories/servicehub/remote/manager.py

@@ -70,6 +70,7 @@
 from paasng.platform.applications.models import Application, ApplicationEnvironment, ModuleEnvironment
 from paasng.platform.engine.models import EngineApp
 from paasng.platform.modules.models import Module
+from paasng.utils import safe_jinja2
 
 if TYPE_CHECKING:
     import datetime
@@ -330,14 +331,12 @@ def get_instance(self) -> ServiceInstanceObj:
         )
 
     def render_params(self, params_tmpl: Dict) -> Dict:
-        """ "Render params dict by current rel's context, Available keys:
+        """Render params dict by current rel's context, Available keys:
 
-        Database objects:
-
-        - `engine_app`: current EngineApp object
-        - `application`: current Application object
-        - `module`: current Module object
-        - `env`: current ModuleEnvironment object
+        - `application/module/env/engine_app`: current object(s) to provision.
+        - `cluster_info`: The cluster information of the current env.
+        - `app_developers`: The application's developers.
+        - `bk_monitor_space_id`: The space id of the bkmonitor space, if exists.
         """
         result = {}
         cluster_info = EnvClusterInfo(self.db_env)
@@ -352,7 +351,8 @@ def render_params(self, params_tmpl: Dict) -> Dict:
             bk_monitor_space_id = space.space_uid
 
         for key, tmpl_str in params_tmpl.items():
-            result[key] = tmpl_str.format(
+            # `str.format()` was previously used. It has been changed to use jinja2 for better security.
+            result[key] = safe_jinja2.StrFormatCompatTemplate(tmpl_str).render(
                 engine_app=self.db_engine_app,
                 application=self.db_application,
                 module=self.db_module,

apiserver/paasng/paasng/accessories/smart_advisor/utils.py

@@ -25,6 +25,7 @@
 from paasng.accessories.smart_advisor.advisor import DocumentaryLinkAdvisor
 from paasng.accessories.smart_advisor.tagging import get_deployment_tags
 from paasng.platform.engine.models.deployment import Deployment
+from paasng.utils.text import basic_str_format
 
 
 @dataclass
@@ -39,7 +40,7 @@ def render_links(self, params: Dict[str, Any]):
         """Render helper's "link" field with given params"""
         for helper in self.helpers:
             if "link" in helper:
-                helper["link"] = helper["link"].format(**params)
+                helper["link"] = basic_str_format(helper["link"], params)
 
 
 def get_default_failure_hint() -> DeploymentFailureHint:

apiserver/paasng/paasng/misc/tools/serializers.py

@@ -169,11 +169,11 @@ def validate(self, value):
         return value
 
 
-class PackageStashRequestSLZ(serializers.Serializer):
+class ToolPackageStashInputSLZ(serializers.Serializer):
     """Handle package for S-mart build"""
 
     package = serializers.FileField(help_text="待构建的应用源码包")
 
 
-class PackageStashResponseSLZ(serializers.Serializer):
+class ToolPackageStashOutputSLZ(serializers.Serializer):
     signature = serializers.CharField(help_text="数字签名")

apiserver/paasng/paasng/misc/tools/views.py

@@ -38,7 +38,7 @@
 from paasng.utils.yaml import IndentDumper
 
 from .app_desc import transform_app_desc_spec2_to_spec3
-from .serializers import AppDescSpec2Serializer, PackageStashRequestSLZ, PackageStashResponseSLZ
+from .serializers import AppDescSpec2Serializer, ToolPackageStashInputSLZ, ToolPackageStashOutputSLZ
 
 
 class AppDescTransformAPIView(APIView):
@@ -87,13 +87,13 @@ class SMartBuilderViewSet(viewsets.ViewSet):
     permission_classes = [IsAuthenticated]
 
     @swagger_auto_schema(
-        request_body=PackageStashRequestSLZ,
-        response_serializer=PackageStashResponseSLZ,
+        request_body=ToolPackageStashInputSLZ,
+        response_serializer=ToolPackageStashOutputSLZ,
         tags=["S-Mart 包构建"],
     )
     def upload(self, request):
         """上传一个待构建的源码包，校验通过后将其暂存起来"""
-        slz = PackageStashRequestSLZ(data=request.data)
+        slz = ToolPackageStashInputSLZ(data=request.data)
         slz.is_valid(raise_exception=True)
 
         with generate_temp_dir() as tmp_dir:
@@ -118,7 +118,7 @@ def upload(self, request):
             # Store as prepared package for later build
             PreparedSourcePackage(request, namespace=self._get_store_namespace(app_desc.code)).store(filepath)
 
-        return Response(PackageStashResponseSLZ({"signature": stat.sha256_signature}).data)
+        return Response(ToolPackageStashOutputSLZ({"signature": stat.sha256_signature}).data)
 
     @staticmethod
     def _validate_app_desc(app_desc: ApplicationDesc):

apiserver/paasng/paasng/plat_admin/numbers/app.py

@@ -52,6 +52,7 @@
 from paasng.platform.sourcectl.repo_controller import BaseGitRepoController
 from paasng.platform.sourcectl.source_types import get_sourcectl_names, get_sourcectl_type
 from paasng.platform.sourcectl.svn.server_config import get_bksvn_config
+from paasng.utils.text import basic_str_format
 
 try:
     from paasng.infras.legacydb_te.models import LApplication, LApplicationUseRecord
@@ -212,7 +213,9 @@ def get_tag_display_name(app: Application):
     def get_market_address(application: Application) -> Optional[str]:
         """获取市场访问地址，兼容精简版应用。普通应用如未打开市场开关，返回 None"""
         region = get_region(application.region)
-        addr = region.basic_info.link_production_app.format(code=application.code, region=application.region)
+        addr = basic_str_format(
+            region.basic_info.link_production_app, {"code": application.code, "region": application.region}
+        )
         if not application.engine_enabled:
             return addr
 

apiserver/paasng/paasng/platform/engine/views/configvar.py

@@ -396,9 +396,14 @@ class ConflictedConfigVarsViewSet(viewsets.ViewSet, ApplicationCodeInPathMixin):
 
     permission_classes = [IsAuthenticated, application_perm_class(AppAction.BASIC_DEVELOP)]
 
+    @swagger_auto_schema(
+        responses={200: ConflictedKeyOutputSLZ(many=True)},
+    )
     def get_user_conflicted_keys(self, request, code, module_name):
-        """获取当前模块中有冲突的环境变量 Key 列表，“冲突”指用户自定义变量与平台内置变量同名。
-        不同类型的应用，平台处理冲突变量的行为有所不同，本接口返回的 key 列表主要作引导和提示用。
+        """获取当前模块中有冲突的环境变量 Key 列表
+
+        “冲突”指用户自定义变量与平台内置变量同名。 不同类型的应用，平台处理冲突变量的行为有所不同，
+        本接口返回的 key 列表主要作引导和提示用。
 
         客户端展示建议：
 

apiserver/paasng/paasng/utils/safe_jinja2.py

@@ -19,11 +19,19 @@
 
 import os
 import typing as t
+from functools import partial
 
 import jinja2
 from jinja2.sandbox import SandboxedEnvironment
 
 
+class NoCallableSandboxedEnvironment(SandboxedEnvironment):
+    """A SandboxedEnvironment that does not allow any function calls."""
+
+    def is_safe_callable(self, obj: t.Any) -> bool:
+        return False
+
+
 def _get_file_environment(
     searchpath: t.Union[str, "os.PathLike[str]", t.Sequence[t.Union[str, "os.PathLike[str]"]]],
     trim_blocks: bool = False,
@@ -34,13 +42,24 @@ def _get_file_environment(
 FileEnvironment = _get_file_environment
 
 
-_default_env = SandboxedEnvironment()
+_default_env = NoCallableSandboxedEnvironment()
 
 
-def _safe_template(source, *args, **kwargs):
+def _safe_template(source, env_cls=NoCallableSandboxedEnvironment, *args, **kwargs):
     if not args and not kwargs:
         return _default_env.from_string(source)
-    return SandboxedEnvironment(*args, **kwargs).from_string(source)
+    return env_cls(*args, **kwargs).from_string(source)
 
 
 Template = _safe_template
+
+
+# StrFormatCompatTemplate is a Template that uses a single brace `{}` as delimiters, which is
+# compatible with str.format() and can be used as a replacement when the template is untrusted.
+StrFormatCompatTemplate = partial(
+    _safe_template,
+    # Disable any function/method calls because supporting `obj.delete()` can be dangerous.
+    env_cls=NoCallableSandboxedEnvironment,
+    variable_start_string="{",
+    variable_end_string="}",
+)

apiserver/paasng/paasng/utils/text.py

@@ -17,6 +17,7 @@
 
 import random
 import re
+import string
 import uuid
 from typing import Collection
 
@@ -124,3 +125,42 @@ def calculate_percentage(x: float, y: float, decimal_places: int = 2) -> str:
     # 否则，将结果转换为百分制，并保留指定位数的小数
     else:
         return "{:.{decimal_places}%}".format(result, decimal_places=decimal_places)
+
+
+class BraceOnlyTemplate(string.Template):
+    """A template class similar to `string.Template`, but use `{` as the delimiter and
+    only supports single-braces formatted variables: `{<name>}`.
+
+    - Use '{{' to escape a single brace
+    - Only '{' need to be escaped, '}' is treated as a normal character
+    """
+
+    # Override the delimiter property because `string.Template` uses it
+    delimiter = "{"
+
+    delim = delimiter
+    # The identifier is copied from `string.Template`
+    id = r"(?a:[_a-z][_a-z0-9]*)"
+
+    # "named" and "braced" patterns are modified
+    pattern = rf"""
+        {delim}(?:
+            (?P<escaped>{delim})  |   # Escape sequence of two delimiters
+            (?P<named>{id})}}     |   # delimiter and a Python identifier, **modified to ends with a bracket**
+            (?P<braced>\b\B)      |   # delimiter and a braced identifier, **modified to never match anything**
+            (?P<invalid>)             # Other ill-formed delimiter exprs
+        )
+        """  # type: ignore
+
+
+def basic_str_format(template: str, context: dict[str, str]) -> str:
+    """This function is similar to `str.format`, but only support basic string substitution,
+    features such as attribute access and slicing are not supported.
+
+    It's recommended to use this function when the template is untrusted.
+
+    :param template: The template string.
+    :param context: The context dictionary.
+    :return: The formatted string.
+    """
+    return BraceOnlyTemplate(template).substitute(**context)

apiserver/paasng/tests/paasng/test_utils/test_text.py

@@ -17,7 +17,14 @@
 
 import pytest
 
-from paasng.utils.text import calculate_percentage, camel_to_snake, remove_suffix, strip_html_tags
+from paasng.utils.text import (
+    BraceOnlyTemplate,
+    basic_str_format,
+    calculate_percentage,
+    camel_to_snake,
+    remove_suffix,
+    strip_html_tags,
+)
 
 
 class TestStripHTMLTags:
@@ -101,3 +108,43 @@ def test_calculate_percentage(x, y, decimal_places, expected):
     else:
         # 否则，应该返回期望值
         assert calculate_percentage(x, y, decimal_places) == expected
+
+
+class TestBraceOnlyTemplate:
+    @pytest.mark.parametrize(
+        ("template_str", "kwargs", "expected"),
+        [
+            # Basic substitution
+            ("Hello {name}, welcome to {place}!", {"name": "foo", "place": "bar"}, "Hello foo, welcome to bar!"),
+            # Variable inside word
+            ("prefix_{var}_suffix", {"var": "test"}, "prefix_test_suffix"),
+            # Underscore variable
+            ("{underscore_var}", {"underscore_var": "value"}, "value"),
+        ],
+    )
+    def test_various_valid_patterns(self, template_str, kwargs, expected):
+        template = BraceOnlyTemplate(template_str)
+        assert template.substitute(**kwargs) == expected
+
+    def test_escaped_braces(self):
+        # Only "{" needs to be escaped
+        template = BraceOnlyTemplate("Use {{ to escape braces, like {{name}")
+        assert template.substitute(name="foo") == "Use { to escape braces, like {name}"
+
+    def test_no_substitution_needed(self):
+        template = BraceOnlyTemplate("No variables here!")
+        assert template.substitute() == "No variables here!"
+
+    def test_missing_variable_raises_error(self):
+        template = BraceOnlyTemplate("Hello {name}!")
+        with pytest.raises(KeyError):
+            template.substitute()
+
+
+class Test__basic_str_format:
+    def test_basic(self):
+        assert basic_str_format("Hello {name}!", {"name": "foo"}) == "Hello foo!"
+
+    def test_index_access_should_fail(self):
+        with pytest.raises(ValueError, match="Invalid placeholder .*"):
+            basic_str_format("Hello {names[0]}!", {"names": "foobar"})