fix(t2i): validate template content to prevent Jinja2 SSTI injection (#8077)

* fix(t2i): validate template content to prevent Jinja2 SSTI injection

* fix(t2i): add error feedback

* fix(test): update assertion to match previous commits

* style: format code

astrbot/core/utils/t2i/template_manager.py

@@ -1,10 +1,52 @@
 # astrbot/core/utils/t2i/template_manager.py
 
+import logging
 import os
+import re
 import shutil
 
 from astrbot.core.utils.astrbot_path import get_astrbot_data_path, get_astrbot_path
 
+logger = logging.getLogger("astrbot")
+
+_ALLOWED_VARS = frozenset({"text", "version", "shiki_runtime"})
+
+_SSTI_BLACKLIST: list[tuple[str, re.Pattern]] = [
+    (
+        "dunder_chain",
+        re.compile(
+            r"__\s*(class|globals|init|mro|base|bases|subclasses|reduce|getitem|builtins|import|self|func|code|reduce_ex)__"
+        ),
+    ),
+    (
+        "dangerous_builtins",
+        re.compile(
+            r"\b(import\s+(?!url)|os\.\w+|subprocess\.|\.popen\(|eval\(|exec\()"
+        ),
+    ),
+    ("flask_context", re.compile(r"\{\{.*?\b(config|request|session|g)\b.*?\}\}")),
+]
+
+_VAR_RE = re.compile(r"\{\{\s*(\w+)\s*(\|[^}]*)?\}\}")
+
+
+def validate_template_content(content: str, *, strict: bool = False) -> None:
+    for label, pattern in _SSTI_BLACKLIST:
+        if pattern.search(content):
+            logger.warning(f"SSTI validation blocked template: matched rule [{label}]")
+            raise ValueError(f"Template contains forbidden pattern ({label}).")
+    if strict:
+        for m in _VAR_RE.finditer(content):
+            var = m.group(1)
+            if var not in _ALLOWED_VARS:
+                logger.warning(
+                    f"SSTI validation blocked template: unauthorized variable '{var}'"
+                )
+                raise ValueError(
+                    f"Unauthorized Jinja2 variable '{var}'; "
+                    f"allowed: {', '.join(sorted(_ALLOWED_VARS))}."
+                )
+
 
 class TemplateManager:
     """负责管理 t2i HTML 模板的 CRUD 和重置操作。
@@ -86,6 +128,7 @@ class TemplateManager:
 
     def create_template(self, name: str, content: str) -> None:
         """在用户目录中创建一个新的模板文件。"""
+        validate_template_content(content, strict=True)
         path = self._get_user_template_path(name)
         if os.path.exists(path):
             raise FileExistsError("同名模板已存在。")
@@ -97,6 +140,7 @@ class TemplateManager:
         如果更新的是一个内置模板，此操作实际上会在用户目录中创建一个修改后的副本，
         从而实现对内置模板的“覆盖”。
         """
+        validate_template_content(content, strict=True)
         path = self._get_user_template_path(name)
         with open(path, "w", encoding="utf-8") as f:
             f.write(content)

dashboard/src/components/shared/T2ITemplateEditor.vue

@@ -243,10 +243,12 @@
 import { ref, computed, nextTick, watch } from 'vue'
 import { VueMonacoEditor } from '@guolao/vue-monaco-editor'
 import { useI18n, useModuleI18n } from '@/i18n/composables'
+import { useToast } from '@/utils/toast'
 import axios from 'axios'
 
 const { t } = useI18n()
 const { tm } = useModuleI18n('core.shared')
+const toast = useToast()
 
 // --- 响应式数据 ---
 const dialog = ref(false)
@@ -448,8 +450,9 @@ const saveTemplate = async () => {
       })
     }
   } catch (error) {
-    console.error('保存模板失败:', error)
-    // 可以在此添加错误提示
+    const msg = error?.response?.data?.message || error?.message || String(error)
+    console.error('保存模板失败:', msg)
+    toast.error(msg)
   } finally {
     saveLoading.value = false
   }
@@ -461,7 +464,9 @@ const setActiveTemplate = async (name) => {
     await axios.post('/api/t2i/templates/set_active', { name })
     activeTemplate.value = name
   } catch (error) {
-    console.error(`应用模板 '${name}' 失败:`, error)
+    const msg = error?.response?.data?.message || error?.message || String(error)
+    console.error(`应用模板 '${name}' 失败:`, msg)
+    toast.error(msg)
   } finally {
     applyLoading.value = false
   }
@@ -482,7 +487,9 @@ const confirmDelete = async () => {
     await loadInitialData()
     selectedTemplate.value = 'base'
   } catch (error) {
-    console.error(`删除模板 '${selectedTemplate.value}' 失败:`, error)
+    const msg = error?.response?.data?.message || error?.message || String(error)
+    console.error(`删除模板失败:`, msg)
+    toast.error(msg)
   } finally {
     saveLoading.value = false
   }
@@ -500,7 +507,9 @@ const confirmReset = async () => {
         await setActiveTemplate('base')
     }
   } catch (error) {
-    console.error('重置模板失败:', error)
+    const msg = error?.response?.data?.message || error?.message || String(error)
+    console.error('重置模板失败:', msg)
+    toast.error(msg)
   } finally {
     resetLoading.value = false
   }

tests/test_dashboard.py

@@ -1026,7 +1026,7 @@ async def test_t2i_set_active_template_syncs_all_configs(
             "/api/t2i/templates/create",
             json={
                 "name": template_name,
-                "content": "<html><body>{{ content }}</body></html>",
+                "content": "<html><body>{{ text }}</body></html>",
             },
             headers=authenticated_header,
         )
@@ -1093,7 +1093,7 @@ async def test_t2i_reset_default_template_syncs_all_configs(
             "/api/t2i/templates/create",
             json={
                 "name": template_name,
-                "content": "<html><body>{{ content }} reset</body></html>",
+                "content": "<html><body>{{ text }} reset</body></html>",
             },
             headers=authenticated_header,
         )
@@ -1166,7 +1166,7 @@ async def test_t2i_update_active_template_reloads_all_schedulers(
             "/api/t2i/templates/create",
             json={
                 "name": template_name,
-                "content": "<html><body>{{ content }} v1</body></html>",
+                "content": "<html><body>{{ text }} v1</body></html>",
             },
             headers=authenticated_header,
         )
@@ -1187,7 +1187,7 @@ async def test_t2i_update_active_template_reloads_all_schedulers(
 
         response = await test_client.put(
             f"/api/t2i/templates/{template_name}",
-            json={"content": "<html><body>{{ content }} v2</body></html>"},
+            json={"content": "<html><body>{{ text }} v2</body></html>"},
             headers=authenticated_header,
         )
         assert response.status_code == 200