Validate connection port is a valid network port number

airflow-core/src/airflow/api_fastapi/core_api/datamodels/connections.py

@@ -191,7 +191,7 @@ class ConnectionBody(StrictBaseModel):
     host: str | None = Field(default=None)
     login: str | None = Field(default=None)
     schema_: str | None = Field(None, alias="schema")
-    port: int | None = Field(default=None)
+    port: int | None = Field(default=None, ge=1, le=65535)
     password: str | None = Field(default=None)
     extra: str | None = Field(default=None)
     team_name: str | None = Field(max_length=50, default=None)

airflow-core/src/airflow/models/connection.py

@@ -28,8 +28,9 @@
 from urllib.parse import parse_qsl, quote, unquote, urlencode, urlsplit
 
 from sqlalchemy import ForeignKey, Integer, String, Text, select
-from sqlalchemy.orm import Mapped, mapped_column, reconstructor
+from sqlalchemy.orm import Mapped, mapped_column, reconstructor, validates
 
+from airflow._shared.configuration import parse_and_validate_port
 from airflow._shared.module_loading import import_string
 from airflow._shared.secrets_backend.base import call_secrets_backend_method
 from airflow._shared.secrets_masker import mask_secret
@@ -219,6 +220,10 @@ def _validate_extra(extra, conn_id) -> None:
             raise ValueError(f"Encountered non-JSON in `extra` field for connection {conn_id!r}.")
         return None
 
+    @validates("port")
+    def validate_port(self, key, value):
+        return parse_and_validate_port(value)
+
     @reconstructor
     def on_db_load(self):
         if self.password:

airflow-core/tests/unit/models/test_connection.py

@@ -540,3 +540,26 @@ def test_get_conn_id_to_team_name_mapping(self, testing_team: Team, session: Ses
             "test_conn2": None,
         }
         clear_db_connections()
+
+    def test_port_validation(self):
+        """Test that Connection model validates the port field correctly."""
+        # Valid ports
+        assert Connection(conn_id="test_port_1", port=80).port == 80
+        assert Connection(conn_id="test_port_2", port="8080").port == 8080
+        assert Connection(conn_id="test_port_3", port=None).port is None
+        assert Connection(conn_id="test_port_4", port="").port is None
+        assert Connection(conn_id="test_port_5", port="  ").port is None
+
+        # Invalid ports - out of range
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_1", port=70000)
+
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_2", port=0)
+
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_3", port=-80)
+
+        # Invalid ports - type errors
+        with pytest.raises(ValueError, match="Expected integer value for `port`"):
+            Connection(conn_id="test_port_fail_4", port="invalid_port")

shared/configuration/src/airflow_shared/configuration/__init__.py

@@ -22,7 +22,9 @@
 __all__ = [
     "AirflowConfigException",
     "AirflowConfigParser",
+    "parse_and_validate_port",
 ]
 
+from .connection import parse_and_validate_port
 from .exceptions import AirflowConfigException
 from .parser import AirflowConfigParser

shared/configuration/src/airflow_shared/configuration/connection.py

@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+
+def parse_and_validate_port(port: int | str | None) -> int | None:
+    """Parse and validate connection port range."""
+    if port is None:
+        return None
+    if isinstance(port, str) and not port.strip():
+        return None
+    try:
+        port_val = int(port)
+    except (ValueError, TypeError):
+        raise ValueError(f"Expected integer value for `port`, but got {port!r} instead.")
+    if not (1 <= port_val <= 65535):
+        raise ValueError(
+            f"The `port` field must be a value between 1 and 65535, but got {port_val!r} instead."
+        )
+    return port_val

task-sdk/src/airflow/sdk/definitions/connection.py

@@ -25,6 +25,7 @@
 
 import attrs
 
+from airflow.sdk._shared.configuration import parse_and_validate_port
 from airflow.sdk.exceptions import AirflowException, AirflowNotFoundException, AirflowRuntimeError, ErrorType
 from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime
 
@@ -153,6 +154,9 @@ def __init__(self, *, conn_id: str, uri: str | None = None, **kwargs) -> None:
         else:
             self.__dict__.update(attrs.asdict(self.from_uri(uri, conn_id=conn_id), recurse=False))
 
+    def __attrs_post_init__(self) -> None:
+        self.port = parse_and_validate_port(self.port)
+
     def get_uri(self) -> str:
         """Generate and return connection in URI format."""
         from urllib.parse import parse_qsl

task-sdk/tests/task_sdk/definitions/test_connection.py

@@ -421,3 +421,41 @@ def test_from_uri_roundtrip(self):
             original_extra = json.loads(conn_from_original.extra)
             roundtrip_extra = json.loads(conn_from_roundtrip.extra)
             assert original_extra == roundtrip_extra
+
+
+class TestConnectionPortValidation:
+    """Test connection port validation in the task-sdk."""
+
+    def test_port_validation(self):
+        """Test that Connection model validates the port field correctly."""
+        # Valid ports
+        assert Connection(conn_id="test_port_1", port=80).port == 80
+        assert Connection(conn_id="test_port_2", port="8080").port == 8080
+        assert Connection(conn_id="test_port_3", port=None).port is None
+
+        # Invalid ports - out of range
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_1", port=70000)
+
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_2", port=0)
+
+        with pytest.raises(ValueError, match="The `port` field must be a value between 1 and 65535"):
+            Connection(conn_id="test_port_fail_3", port=-80)
+
+        # Invalid ports - type errors
+        with pytest.raises(ValueError, match="Expected integer value for `port`"):
+            Connection(conn_id="test_port_fail_4", port="invalid_port")
+
+    def test_from_uri_port_validation(self):
+        """Test that Connection.from_uri validates the port field correctly."""
+        # Valid port from URI
+        assert Connection.from_uri("postgres://host:5432/db", conn_id="test").port == 5432
+
+        # Invalid port from URI - out of range
+        with pytest.raises(ValueError, match="Port out of range 0-65535"):
+            Connection.from_uri("postgres://host:70000/db", conn_id="test")
+
+        # Invalid port from URI - type/invalid format
+        with pytest.raises(ValueError, match="Port could not be cast to integer value"):
+            Connection.from_uri("postgres://host:abc/db", conn_id="test")