sanitize kubernetes messages

src/servala/core/models/service.py

@@ -1,4 +1,5 @@
 import copy
+import html
 import json
 import re
 
@@ -607,29 +608,60 @@ class ServiceInstance(ServalaModelMixin, models.Model):
 
     @classmethod
     def _format_kubernetes_error(cls, error_message):
-        """
-        Format Kubernetes API error messages for better user experience.
-        Converts validation error arrays into unordered lists.
-        """
-        # Pattern to match validation errors in brackets
+        if not error_message:
+            return {"message": "", "errors": None, "has_list": False}
+
+        error_message = str(error_message).strip()
+
+        # Pattern to match validation errors in brackets like [error1, error2, error3]
         pattern = r"\[([^\]]+)\]"
         match = re.search(pattern, error_message)
 
         if not match:
-            return error_message
+            return {"message": error_message, "errors": None, "has_list": False}
 
-        errors_text = match.group(1)
-        # Split by comma and clean up each error
-        errors = [error.strip() for error in errors_text.split(",")]
+        errors_text = match.group(1).strip()
 
-        if len(errors) > 1:
-            # Format as HTML unordered list
-            error_list = "".join(f"<li>{error}</li>" for error in errors)
-            # Replace the bracketed section with the formatted list
-            formatted_message = re.sub(pattern, f"<ul>{error_list}</ul>", error_message)
-            return mark_safe(formatted_message)
+        if "," not in errors_text:
+            return {"message": error_message, "errors": None, "has_list": False}
 
-        return error_message
+        errors = [error.strip().strip("\"'") for error in errors_text.split(",")]
+        errors = [error for error in errors if error]
+
+        if len(errors) <= 1:
+            return {"message": error_message, "errors": None, "has_list": False}
+
+        base_message = re.sub(pattern, "", error_message).strip()
+        base_message = base_message.rstrip(":").strip()
+
+        return {"message": base_message, "errors": errors, "has_list": True}
+
+    @classmethod
+    def _safe_format_error(cls, error_data):
+        if not isinstance(error_data, dict):
+            return html.escape(str(error_data))
+
+        if not error_data.get("has_list", False):
+            return html.escape(error_data.get("message", ""))
+
+        message = html.escape(error_data.get("message", ""))
+        errors = error_data.get("errors", [])
+
+        if not errors:
+            return message
+
+        escaped_errors = [html.escape(str(error)) for error in errors]
+        error_items = "".join(f"<li>{error}</li>" for error in escaped_errors)
+
+        if message:
+            return mark_safe(f"{message}<ul>{error_items}</ul>")
+        else:
+            return mark_safe(f"<ul>{error_items}</ul>")
+
+    @classmethod
+    def format_error_for_display(cls, error_message):
+        error_data = cls._format_kubernetes_error(error_message)
+        return cls._safe_format_error(error_data)
 
     @classmethod
     def create_instance(cls, name, organization, context, created_by, spec_data):
@@ -684,18 +716,21 @@ class ServiceInstance(ServalaModelMixin, models.Model):
                 try:
                     error_body = json.loads(e.body)
                     reason = error_body.get("message", str(e))
-                    formatted_reason = cls._format_kubernetes_error(reason)
+                    error_data = cls._format_kubernetes_error(reason)
+                    formatted_reason = cls._safe_format_error(error_data)
                     message = _("Error reported by control plane: {reason}").format(
                         reason=formatted_reason
                     )
                     raise ValidationError(organization.add_support_message(message))
                 except (ValueError, TypeError):
-                    formatted_error = cls._format_kubernetes_error(str(e))
+                    error_data = cls._format_kubernetes_error(str(e))
+                    formatted_error = cls._safe_format_error(error_data)
                     message = _("Error reported by control plane: {error}").format(
                         error=formatted_error
                     )
                     raise ValidationError(organization.add_support_message(message))
-            formatted_error = cls._format_kubernetes_error(str(e))
+            error_data = cls._format_kubernetes_error(str(e))
+            formatted_error = cls._safe_format_error(error_data)
             message = _("Error creating instance: {error}").format(
                 error=formatted_error
             )
@@ -727,19 +762,22 @@ class ServiceInstance(ServalaModelMixin, models.Model):
             try:
                 error_body = json.loads(e.body)
                 reason = error_body.get("message", str(e))
-                formatted_reason = self._format_kubernetes_error(reason)
+                error_data = self._format_kubernetes_error(reason)
+                formatted_reason = self._safe_format_error(error_data)
                 message = _(
                     "Error reported by control plane while updating instance: {reason}"
                 ).format(reason=formatted_reason)
                 raise ValidationError(self.organization.add_support_message(message))
             except (ValueError, TypeError):
-                formatted_error = self._format_kubernetes_error(str(e))
+                error_data = self._format_kubernetes_error(str(e))
+                formatted_error = self._safe_format_error(error_data)
                 message = _(
                     "Error reported by control plane while updating instance: {error}"
                 ).format(error=formatted_error)
                 raise ValidationError(self.organization.add_support_message(message))
         except Exception as e:
-            formatted_error = self._format_kubernetes_error(str(e))
+            error_data = self._format_kubernetes_error(str(e))
+            formatted_error = self._safe_format_error(error_data)
             message = _("Error updating instance: {error}").format(
                 error=formatted_error
             )

src/servala/frontend/templates/includes/k8s_error.html

@@ -0,0 +1,14 @@
+{% if show_error %}
+<div class="{{ css_class }}">
+    {% if has_list %}
+        {% if message %}{{ message }}{% endif %}
+        <ul>
+        {% for error in errors %}
+            <li>{{ error }}</li>
+        {% endfor %}
+        </ul>
+    {% else %}
+        {{ message }}
+    {% endif %}
+</div>
+{% endif %}

src/servala/frontend/templatetags/error_filters.py

@@ -0,0 +1,43 @@
+"""
+Template filters for safe error formatting.
+"""
+
+import html
+from django import template
+from django.utils.safestring import mark_safe
+
+register = template.Library()
+
+
+@register.filter
+def format_k8s_error(error_data):
+    """
+    Template filter to safely format Kubernetes error data.
+    Usage: {{ error_data|format_k8s_error }}
+
+    Args:
+        error_data: Dictionary with structure from _format_kubernetes_error method
+                   or a simple string
+
+    Returns:
+        Safely formatted HTML string
+    """
+    if not error_data:
+        return ""
+
+    if not error_data.get("has_list", False):
+        return html.escape(error_data.get("message", ""))
+
+    message = html.escape(error_data.get("message", ""))
+    errors = error_data.get("errors", [])
+
+    if not errors:
+        return message
+
+    escaped_errors = [html.escape(str(error)) for error in errors]
+    error_items = "".join(f"<li>{error}</li>" for error in escaped_errors)
+
+    if message:
+        return mark_safe(f"{message}<ul>{error_items}</ul>")
+    else:
+        return mark_safe(f"<ul>{error_items}</ul>")