fix: sanitize template variables before YAML validation

This commit is contained in:
2026-05-22 21:01:59 +00:00
parent ae41a64e66
commit 765cb965e6
+23 -6
View File
@@ -1,3 +1,4 @@
import re
import uuid
from datetime import datetime
@@ -7,6 +8,11 @@ from pydantic import BaseModel, ConfigDict, field_validator, model_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
def _sanitize_template_vars(template: str) -> str:
"""Replace template variables like {{VAR}} with placeholders to avoid YAML parsing errors."""
return re.sub(r"\{\{[A-Za-z_][A-Za-z0-9_]*\}\}", "__PLACEHOLDER__", template)
from src.auth.dependencies import get_current_user_id, get_db_session
from src.models.tool_type import ToolType
from src.models.user import User
@@ -65,8 +71,12 @@ class ToolTypeCreate(BaseModel):
if v is None:
raise ValueError("compose_template is required when definition_type is 'compose'")
# Replace template variables with dummy values before YAML validation
# to avoid YAML parsing errors with {{VAR}} syntax
sanitized = _sanitize_template_vars(v)
try:
parsed = yaml.safe_load(v)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
@@ -145,7 +155,8 @@ class ToolTypeCreate(BaseModel):
# Validate that default_port is exposed in compose template (only if requires_port)
if self.requires_port and self.definition_type == "compose" and self.compose_template:
try:
parsed = yaml.safe_load(self.compose_template)
sanitized = _sanitize_template_vars(self.compose_template)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError:
return self
@@ -215,8 +226,11 @@ class ToolTypeUpdate(BaseModel):
if definition_type and definition_type != "compose":
return v
# Replace template variables with dummy values before YAML validation
sanitized = _sanitize_template_vars(v)
try:
parsed = yaml.safe_load(v)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError as e:
raise ValueError(f"Invalid YAML: {e}")
@@ -427,7 +441,8 @@ async def update_tool_type(
template = update_data.get("compose_template", tool_type.compose_template)
if template:
try:
parsed = yaml.safe_load(template)
sanitized = _sanitize_template_vars(template)
parsed = yaml.safe_load(sanitized)
except yaml.YAMLError:
parsed = None
@@ -518,7 +533,8 @@ async def validate_tool_type_template(
errors.append("Compose template is required")
else:
try:
parsed = yaml.safe_load(data.compose_template)
sanitized = _sanitize_template_vars(data.compose_template)
parsed = yaml.safe_load(sanitized)
if not isinstance(parsed, dict):
errors.append("Compose template must be a YAML mapping")
elif "services" not in parsed:
@@ -575,7 +591,8 @@ async def validate_tool_type(
errors.append("Compose template is empty")
else:
try:
parsed = yaml.safe_load(tool_type.compose_template)
sanitized = _sanitize_template_vars(tool_type.compose_template)
parsed = yaml.safe_load(sanitized)
if not isinstance(parsed, dict):
errors.append("Compose template must be a YAML mapping")
elif "services" not in parsed: