diff --git a/airflow-core/newsfragments/68541.bugfix.rst b/airflow-core/newsfragments/68541.bugfix.rst
new file mode 100644
index 0000000000000..02ab3fee730d5
--- /dev/null
+++ b/airflow-core/newsfragments/68541.bugfix.rst
@@ -0,0 +1 @@
+Validate that ``Connection.port`` is within the TCP/UDP range (1-65535) on all write paths (model constructor, ``from_json``, REST API, and CLI). Port ``0`` and negative values are now rejected. Read paths and ORM loads are unchanged so existing rows with out-of-range values continue to load.
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/connections.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/connections.py
index 34cfe47334893..be8305d19b053 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/connections.py
+++ b/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)
diff --git a/airflow-core/src/airflow/cli/cli_config.py b/airflow-core/src/airflow/cli/cli_config.py
index 99d7faf30bca6..897d914b95397 100644
--- a/airflow-core/src/airflow/cli/cli_config.py
+++ b/airflow-core/src/airflow/cli/cli_config.py
@@ -848,7 +848,7 @@ def string_lower_type(val):
 ARG_CONN_SCHEMA = Arg(
     ("--conn-schema",), help="Connection schema, optional when adding a connection", type=str
 )
-ARG_CONN_PORT = Arg(("--conn-port",), help="Connection port, optional when adding a connection", type=str)
+ARG_CONN_PORT = Arg(("--conn-port",), help="Connection port (1-65535), optional when adding a connection", type=int)
 ARG_CONN_EXTRA = Arg(
     ("--conn-extra",), help="Connection `Extra` field, optional when adding a connection", type=str
 )
diff --git a/airflow-core/src/airflow/models/connection.py b/airflow-core/src/airflow/models/connection.py
index 1b4b0f8f86768..ce2de3139de5f 100644
--- a/airflow-core/src/airflow/models/connection.py
+++ b/airflow-core/src/airflow/models/connection.py
@@ -190,6 +190,8 @@ def __init__(
             self.port = port
             self.extra = extra
 
+        self._validate_port(self.port)
+
         if conn_id is not None:
             sanitized_id = sanitize_conn_id(conn_id)
             if sanitized_id is not None:
@@ -202,6 +204,13 @@ def __init__(
             mask_secret(quote(self.password))
         self.team_name = team_name
 
+    @staticmethod
+    def _validate_port(port: int | None) -> None:
+        if port is not None and (port < 1 or port > 65535):
+            raise ValueError(
+                f"The connection port must be between 1 and 65535, but got {port!r}."
+            )
+
     @staticmethod
     def _validate_extra(extra, conn_id) -> None:
         """Verify that ``extra`` is a JSON-encoded Python dict."""
@@ -591,6 +600,7 @@ def from_json(cls, value, conn_id=None) -> Connection:
                 kwargs["port"] = int(port)
             except ValueError:
                 raise ValueError(f"Expected integer value for `port`, but got {port!r} instead.")
+            cls._validate_port(kwargs["port"])
         return Connection(conn_id=conn_id, **kwargs)
 
     def as_json(self) -> str:
diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
index 6930039e1abbc..41057c706e3af 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
@@ -352,6 +352,19 @@ def test_should_respond_403(self, unauthorized_test_client):
             {"connection_id": "iam_not@#$_connection_id", "conn_type": TEST_CONN_TYPE},
         ],
     )
+    @pytest.mark.parametrize("port", [0, -1, 65536, 99999])
+    def test_post_should_respond_422_for_out_of_range_port(self, test_client, port):
+        """REST API must reject port values outside 1-65535 (closes #68382)."""
+        response = test_client.post(
+            "/connections",
+            json={
+                "connection_id": TEST_CONN_ID,
+                "conn_type": TEST_CONN_TYPE,
+                "port": port,
+            },
+        )
+        assert response.status_code == 422
+
     def test_post_should_respond_422_for_invalid_conn_id(self, test_client, body):
         response = test_client.post("/connections", json=body)
         assert response.status_code == 422
diff --git a/airflow-core/tests/unit/cli/commands/test_connection_command.py b/airflow-core/tests/unit/cli/commands/test_connection_command.py
index c62dfb4c5b357..24a3404f5a297 100644
--- a/airflow-core/tests/unit/cli/commands/test_connection_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_connection_command.py
@@ -687,6 +687,36 @@ def test_cli_connections_add_duplicate(self):
                 self.parser.parse_args(["connections", "add", conn_id, f"--conn-uri={TEST_URL}"])
             )
 
+    @pytest.mark.parametrize("port", ["0", "-1", "65536", "99999"])
+    def test_cli_connections_add_rejects_out_of_range_port(self, port):
+        """CLI ``connections add`` must reject port values outside 1-65535 (closes #68382)."""
+        with pytest.raises((SystemExit, ValueError)):
+            connection_command.connections_add(
+                self.parser.parse_args(
+                    [
+                        "connections",
+                        "add",
+                        "new_conn",
+                        "--conn-type=http",
+                        "--conn-host=example.com",
+                        f"--conn-port={port}",
+                    ]
+                )
+            )
+
+    def test_cli_connections_add_rejects_non_integer_port(self):
+        """CLI ``--conn-port`` is typed as int — argparse should reject non-integer strings."""
+        with pytest.raises(SystemExit):
+            self.parser.parse_args(
+                [
+                    "connections",
+                    "add",
+                    "new_conn",
+                    "--conn-type=http",
+                    "--conn-port=not-a-number",
+                ]
+            )
+
     def test_cli_connections_add_delete_with_missing_parameters(self):
         # Attempt to add without providing conn_uri
         with pytest.raises(
diff --git a/airflow-core/tests/unit/models/test_connection.py b/airflow-core/tests/unit/models/test_connection.py
index 94cabe5e4daf4..9d68bfe12de5a 100644
--- a/airflow-core/tests/unit/models/test_connection.py
+++ b/airflow-core/tests/unit/models/test_connection.py
@@ -424,6 +424,47 @@ def test_extra_dejson(self):
             "headers": {"Content-Type": "application/json", "X-Requested-By": "Airflow"},
         }
 
+    @pytest.mark.parametrize("port", [1, 22, 5432, 65535])
+    def test_port_within_valid_range_is_accepted(self, port):
+        """Ports in the valid TCP/UDP range (1-65535) should be accepted."""
+        conn = Connection(conn_id="test_conn", conn_type="http", host="example.com", port=port)
+        assert conn.port == port
+
+    def test_port_none_is_accepted(self):
+        """``port`` is optional, ``None`` must remain a valid value."""
+        conn = Connection(conn_id="test_conn", conn_type="http", host="example.com", port=None)
+        assert conn.port is None
+
+    @pytest.mark.parametrize("port", [0, -1, 65536, 99999])
+    def test_port_out_of_range_is_rejected(self, port):
+        """Ports outside 1-65535 should raise ValueError on Connection() (closes #68382)."""
+        with pytest.raises(ValueError, match="must be between 1 and 65535"):
+            Connection(conn_id="test_conn", conn_type="http", host="example.com", port=port)
+
+    @pytest.mark.parametrize("port", [0, -1, 99999])
+    def test_from_json_rejects_out_of_range_port(self, port):
+        """``Connection.from_json`` should also enforce the port range."""
+        import json as _json
+
+        payload = _json.dumps({"conn_type": "http", "host": "example.com", "port": port})
+        with pytest.raises(ValueError, match="must be between 1 and 65535"):
+            Connection.from_json(payload, conn_id="test_conn")
+
+    def test_orm_load_tolerates_legacy_invalid_port(self):
+        """Existing rows with invalid port values must still load via SQLAlchemy.
+
+        SQLAlchemy uses ``@reconstructor`` (``on_db_load``) when loading rows,
+        which bypasses ``__init__`` — so legacy persisted invalid data should
+        continue to load without raising. This guards against breaking
+        existing installations (issue #68382 review comment).
+        """
+        conn = Connection.__new__(Connection)
+        conn.conn_id = "legacy_conn"
+        conn.conn_type = "http"
+        conn.port = 99999
+        conn.on_db_load()
+        assert conn.port == 99999
+
     @mock.patch("airflow.sdk.Connection.get")
     def test_get_connection_from_secrets_task_sdk_success(self, mock_get):
         """Test the get_connection_from_secrets method with Task SDK success path."""
diff --git a/task-sdk/src/airflow/sdk/definitions/connection.py b/task-sdk/src/airflow/sdk/definitions/connection.py
index 06a95a3b868b8..ca3ed3e72e146 100644
--- a/task-sdk/src/airflow/sdk/definitions/connection.py
+++ b/task-sdk/src/airflow/sdk/definitions/connection.py
@@ -118,11 +118,19 @@ class Connection:
     schema: str | None = None
     login: str | None = None
     password: str | None = None
-    port: int | None = None
+    port: int | None = attrs.field(default=None)
     extra: str | None = None
 
     EXTRA_KEY = "__extra__"
 
+    @port.validator
+    def _validate_port(self, attribute, value):
+        if value is not None and (value < 1 or value > 65535):
+            raise ValueError(
+                f"The connection port must be between 1 and 65535, but got {value!r}."
+            )
+
+
     @overload
     def __init__(self, *, conn_id: str, uri: str) -> None: ...
 
@@ -360,6 +368,11 @@ def from_json(cls, value, conn_id=None) -> Connection:
                 kwargs["port"] = int(port)
             except ValueError:
                 raise ValueError(f"Expected integer value for `port`, but got {port!r} instead.")
+            port_val = kwargs["port"]
+            if port_val < 1 or port_val > 65535:
+                raise ValueError(
+                    f"The connection port must be between 1 and 65535, but got {port_val!r}."
+                )
         return cls(conn_id=conn_id, **kwargs)
 
     def as_json(self) -> str:
diff --git a/task-sdk/tests/task_sdk/definitions/test_connection.py b/task-sdk/tests/task_sdk/definitions/test_connection.py
index 5746b1b14f75c..0f31c6e36bb15 100644
--- a/task-sdk/tests/task_sdk/definitions/test_connection.py
+++ b/task-sdk/tests/task_sdk/definitions/test_connection.py
@@ -208,6 +208,30 @@ def test_from_json_without_conn_type(self):
         assert connection.password == "secret"
         assert connection.schema == "production"
 
+    @pytest.mark.parametrize("port", [1, 22, 5432, 65535])
+    def test_port_within_valid_range_is_accepted(self, port):
+        """Ports in the valid TCP/UDP range (1-65535) should be accepted."""
+        conn = Connection(conn_id="test_conn", conn_type="http", host="example.com", port=port)
+        assert conn.port == port
+
+    def test_port_none_is_accepted(self):
+        """``port`` is optional, ``None`` must remain a valid value."""
+        conn = Connection(conn_id="test_conn", conn_type="http", host="example.com", port=None)
+        assert conn.port is None
+
+    @pytest.mark.parametrize("port", [0, -1, 65536, 99999])
+    def test_port_out_of_range_is_rejected(self, port):
+        """Ports outside 1-65535 should raise ValueError on Connection() (closes #68382)."""
+        with pytest.raises(ValueError, match="must be between 1 and 65535"):
+            Connection(conn_id="test_conn", conn_type="http", host="example.com", port=port)
+
+    @pytest.mark.parametrize("port", [0, -1, 99999])
+    def test_from_json_rejects_out_of_range_port(self, port):
+        """``Connection.from_json`` should also enforce the port range."""
+        payload = json.dumps({"conn_type": "http", "host": "example.com", "port": port})
+        with pytest.raises(ValueError, match="must be between 1 and 65535"):
+            Connection.from_json(payload, conn_id="test_conn")
+
     def test_extra_dejson_property(self):
         """Test that extra_dejson property correctly deserializes JSON extra field."""
         connection = Connection(