fix(ingestion): protect all connectors from scheme-prefixed hostPort by stripping in model_post_init

Addresses gitar-bot review feedback on PR open-metadata#27191: moving strip_hostport_scheme
to get_connection_url_common left several connectors that parse hostPort manually
still vulnerable to the original crash (open-metadata#24348):

  - cassandra/connection.py: host, port = connection.hostPort.split(':')
  - db2/connection.py: connection.hostPort.split(':')[0]
  - databricks/auth.py: connection.hostPort.split(':')[0]
  - redshift/connection.py: connection.hostPort.split(':')[0]
  - microsoftfabric/connection.py: connection.hostPort.split(':')[0]
  - builders.py (IAM auth): connection.hostPort.split(':')

Restore strip_hostport_scheme call to model_post_init so scheme stripping
happens at construction time for ALL *Connection classes, protecting every
downstream split site automatically.

Exclude AirflowConnection and OpenMetadataConnection via _HOSTPORT_URL_ALLOWED
since their hostPort field legitimately stores a full URL (e.g.
http://localhost:8080 and http://localhost:8585/api).

The call in get_connection_url_common is kept for defence-in-depth; it is
idempotent because strip_hostport_scheme short-circuits when '://' is absent.

Add three new unit tests to test_connection_builders.py covering:
- URL scheme stripping at model construction
- ValueError raised for invalid port at construction time
- URL-allowed connections are not stripped

Author: RajdeepKushwaha5

ingestion/src/metadata/ingestion/models/custom_pydantic.py

@@ -122,6 +122,13 @@ def strip_hostport_scheme(raw: str) -> str:
 # pinned to the original underscored symbol while the helper was private.
 _strip_hostport_scheme = strip_hostport_scheme
 
+# Connection classes whose hostPort field legitimately stores a full URL
+# (e.g. "http://airflow:8080" or "http://localhost:8585/api") and must NOT
+# have their scheme stripped by model_post_init.
+_HOSTPORT_URL_ALLOWED: frozenset = frozenset(
+    {"AirflowConnection", "OpenMetadataConnection"}
+)
+
 
 class BaseModel(PydanticBaseModel):
     """
@@ -134,14 +141,31 @@ def model_post_init(self, context: Any, /):
         This function is used to parse the FilterPattern fields for the Connection classes.
         This is needed because dict is defined in the JSON schema for the FilterPattern field,
         but a FilterPattern object is required in the generated code.
+
+        Additionally, for Connection classes that store a plain ``hostname[:port]``
+        in ``hostPort``, any accidental URL scheme prefix (e.g. ``http://``) is
+        stripped here so that all downstream connector code — including connectors
+        that call ``connection.hostPort.split(":")`` directly — receives a clean value.
         """
         # pylint: disable=import-outside-toplevel
+        if not self.__class__.__name__.endswith("Connection"):
+            # Only process Connection classes
+            return
+        if not hasattr(self, "__pydantic_fields__"):
+            return
+
+        # Strip accidental URL schemes from hostPort at model construction time.
+        # This protects connectors that split hostPort manually (e.g. Cassandra,
+        # Databricks, Redshift, DB2, Microsoft Fabric) without going through
+        # get_connection_url_common. ValueError for genuinely invalid inputs
+        # (e.g. non-numeric port) propagates immediately with a clear message.
+        if (
+            hasattr(self, "hostPort")
+            and self.__class__.__name__ not in _HOSTPORT_URL_ALLOWED
+        ):
+            self.hostPort = strip_hostport_scheme(self.hostPort)
+
         try:
-            if not self.__class__.__name__.endswith("Connection"):
-                # Only parse FilterPattern for Connection classes
-                return
-            if not hasattr(self, "__pydantic_fields__"):
-                return
             for field in self.__pydantic_fields__:
                 if field.endswith("FilterPattern"):
                     from metadata.generated.schema.type.filterPattern import (

ingestion/tests/unit/test_connection_builders.py

@@ -19,11 +19,15 @@
 from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
     MysqlConnection,
 )
+from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
+    OpenMetadataConnection,
+)
 from metadata.ingestion.connections.builders import (
     get_connection_args_common,
     get_connection_options_dict,
     init_empty_connection_arguments,
 )
+from metadata.ingestion.models.custom_pydantic import _HOSTPORT_URL_ALLOWED
 
 
 class ConnectionBuilderTest(TestCase):
@@ -79,3 +83,64 @@ def test_init_empty_connection_arguments(self):
 
         self.assertEqual(new_args.root.get("hello"), "world")
         self.assertIsNone(new_args.root.get("not there"))
+
+    def test_model_post_init_strips_url_scheme_from_hostport(self):
+        """
+        Verify that model_post_init strips URL schemes from hostPort at
+        construction time so connectors that call hostPort.split(":") directly
+        receive a clean hostname[:port] value.
+        """
+        # http:// prefix should be stripped
+        conn = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password="pass"),
+            hostPort="http://localhost:3306",
+        )
+        self.assertEqual(conn.hostPort, "localhost:3306")
+
+        # https:// prefix should be stripped
+        conn2 = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password="pass"),
+            hostPort="https://myhost:5432",
+        )
+        self.assertEqual(conn2.hostPort, "myhost:5432")
+
+        # Already clean value should pass through unchanged
+        conn3 = MysqlConnection(
+            username="user",
+            authType=BasicAuth(password="pass"),
+            hostPort="localhost:3306",
+        )
+        self.assertEqual(conn3.hostPort, "localhost:3306")
+
+    def test_model_post_init_raises_for_invalid_port_in_hostport(self):
+        """
+        Verify that a non-numeric port in a scheme-prefixed hostPort raises
+        at model construction time (fail-fast) rather than producing a
+        confusing error deep in connector code.
+        """
+        with self.assertRaises(Exception):
+            MysqlConnection(
+                username="user",
+                authType=BasicAuth(password="pass"),
+                hostPort="http://localhost:abc",
+            )
+
+    def test_url_allowed_connections_not_stripped(self):
+        """
+        Verify that connection classes in _HOSTPORT_URL_ALLOWED (AirflowConnection
+        and OpenMetadataConnection) are excluded from hostPort scheme stripping,
+        since their hostPort field legitimately stores a full URL.
+        """
+        # Confirm the exclusion set contains the expected class names
+        self.assertIn("AirflowConnection", _HOSTPORT_URL_ALLOWED)
+        self.assertIn("OpenMetadataConnection", _HOSTPORT_URL_ALLOWED)
+
+        # OpenMetadataConnection.hostPort is a plain str field that expects a
+        # full URL like "http://localhost:8585/api" — it must NOT be stripped.
+        ometa = OpenMetadataConnection(
+            hostPort="http://localhost:8585/api",
+        )
+        self.assertIn("http", ometa.hostPort)
+        self.assertIn("localhost", ometa.hostPort)