Sanitize the conn_id to disallow potential script execution

airflow/models/connection.py

@@ -19,6 +19,7 @@
 
 import json
 import logging
+import re
 import warnings
 from json import JSONDecodeError
 from urllib.parse import parse_qsl, quote, unquote, urlencode, urlsplit
@@ -35,6 +36,7 @@
 from airflow.utils.module_loading import import_string
 
 log = logging.getLogger(__name__)
+_RE_COMPILED = re.compile("^[\w\@\#\$\%\&\!\(\)\*\-]{1,200}$")
 
 
 def parse_netloc_to_hostname(*args, **kwargs):
@@ -43,6 +45,23 @@ def parse_netloc_to_hostname(*args, **kwargs):
     return _parse_netloc_to_hostname(*args, **kwargs)
 
 
+def sanitize_conn_id(conn_id: str | None) -> str | None:
+    """
+    Sanitises the connection id and allows only specific characters to be within.
+
+    The character selection prevents the injection of javascript or executable bits
+    in order to avoid any awkward behavior in the front-end.
+
+    :param conn_id: The connection id to sanitize.
+    :return: the sanitized string, `None` otherwise.
+    """
+    res = None
+    if conn_id is None or (res := re.match(_RE_COMPILED, conn_id)) is None:
+        log.warning("We failed to match the conn_id or it was None")
+
+    return res if res is None else conn_id
+
+
 # Python automatically converts all letters to lowercase in hostname
 # See: https://issues.apache.org/jira/browse/AIRFLOW-3615
 def _parse_netloc_to_hostname(uri_parts):
@@ -112,7 +131,7 @@ def __init__(
         uri: str | None = None,
     ):
         super().__init__()
-        self.conn_id = conn_id
+        self.conn_id = sanitize_conn_id(conn_id)
         self.description = description
         if extra and not isinstance(extra, str):
             extra = json.dumps(extra)