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 f2ac1f7a9403a..fee330e1fd1d6 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
@@ -17,12 +17,15 @@
 
 from __future__ import annotations
 
+import json
 from collections import abc
 from typing import Annotated
 
-from pydantic import Field
+from pydantic import Field, field_validator
+from pydantic_core.core_schema import ValidationInfo
 
 from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.sdk.execution_time.secrets_masker import redact
 
 
 # Response Models
@@ -39,6 +42,26 @@ class ConnectionResponse(BaseModel):
     password: str | None
     extra: str | None
 
+    @field_validator("password", mode="after")
+    @classmethod
+    def redact_password(cls, v: str | None, field_info: ValidationInfo) -> str | None:
+        if v is None:
+            return None
+        return redact(v, field_info.field_name)
+
+    @field_validator("extra", mode="before")
+    @classmethod
+    def redact_extra(cls, v: str | None) -> str | None:
+        if v is None:
+            return None
+        try:
+            extra_dict = json.loads(v)
+            redacted_dict = redact(extra_dict)
+            return json.dumps(redacted_dict)
+        except json.JSONDecodeError:
+            # we can't redact fields in an unstructured `extra`
+            return v
+
 
 class ConnectionCollectionResponse(BaseModel):
     """Connection Collection serializer for responses."""
diff --git a/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx b/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx
index 8452e738445e4..866d555c06857 100644
--- a/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx
@@ -56,7 +56,7 @@ const ConnectionForm = ({
   const { conf: extra, setConf } = useParamStore();
   const {
     control,
-    formState: { isDirty, isValid },
+    formState: { isValid },
     handleSubmit,
     reset,
     watch,
@@ -92,14 +92,6 @@ const ConnectionForm = ({
     mutateConnection(data);
   };
 
-  const hasChanges = () => {
-    if (isDirty) {
-      return true;
-    }
-
-    return JSON.stringify(JSON.parse(extra)) !== JSON.stringify(JSON.parse(initialConnection.extra));
-  };
-
   const validateAndPrettifyJson = (value: string) => {
     try {
       const parsedJson = JSON.parse(value) as JSON;
@@ -240,7 +232,7 @@ const ConnectionForm = ({
           <Spacer />
           <Button
             colorPalette="blue"
-            disabled={Boolean(errors.conf) || formErrors || isPending || !isValid || !hasChanges()}
+            disabled={Boolean(errors.conf) || formErrors || isPending || !isValid}
             onClick={() => void handleSubmit(onSubmit)()}
           >
             <FiSave /> Save
diff --git a/airflow-core/src/airflow/ui/src/queries/useEditConnection.tsx b/airflow-core/src/airflow/ui/src/queries/useEditConnection.tsx
index d76e34bd57b55..6f7face3a325b 100644
--- a/airflow-core/src/airflow/ui/src/queries/useEditConnection.tsx
+++ b/airflow-core/src/airflow/ui/src/queries/useEditConnection.tsx
@@ -60,9 +60,7 @@ export const useEditConnection = (
   const editConnection = (requestBody: ConnectionBody) => {
     const updateMask: Array<string> = [];
 
-    if (
-      JSON.stringify(JSON.parse(requestBody.extra)) !== JSON.stringify(JSON.parse(initialConnection.extra))
-    ) {
+    if (requestBody.extra !== initialConnection.extra) {
       updateMask.push("extra");
     }
     if (requestBody.conn_type !== initialConnection.conn_type) {
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 4741a03f67c63..ddff799431cf7 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
@@ -158,6 +158,35 @@ def test_get_should_respond_200_with_extra(self, test_client, session):
         assert body["conn_type"] == TEST_CONN_TYPE
         assert body["extra"] == '{"extra_key": "extra_value"}'
 
+    @pytest.mark.enable_redact
+    def test_get_should_respond_200_with_extra_redacted(self, test_client, session):
+        self.create_connection()
+        connection = session.query(Connection).first()
+        connection.extra = '{"password": "test-password"}'
+        session.commit()
+        response = test_client.get(f"/connections/{TEST_CONN_ID}")
+        assert response.status_code == 200
+        body = response.json()
+        assert body["connection_id"] == TEST_CONN_ID
+        assert body["conn_type"] == TEST_CONN_TYPE
+        assert body["extra"] == '{"password": "***"}'
+
+    @pytest.mark.enable_redact
+    def test_get_should_not_overmask_short_password_value_in_extra(self, test_client, session):
+        connection = Connection(
+            conn_id=TEST_CONN_ID, conn_type="generic", login="a", password="a", extra='{"key": "value"}'
+        )
+        session.add(connection)
+        session.commit()
+
+        response = test_client.get(f"/connections/{TEST_CONN_ID}")
+        assert response.status_code == 200
+        body = response.json()
+        assert body["connection_id"] == TEST_CONN_ID
+        assert body["conn_type"] == "generic"
+        assert body["login"] == "a"
+        assert body["extra"] == '{"key": "value"}'
+
 
 class TestGetConnections(TestConnectionEndpoint):
     @pytest.mark.parametrize(
@@ -279,6 +308,7 @@ def test_post_should_respond_already_exist(self, test_client, body):
         assert "detail" in response_json
         assert list(response_json["detail"].keys()) == ["reason", "statement", "orig_error", "message"]
 
+    @pytest.mark.enable_redact
     @pytest.mark.parametrize(
         "body, expected_response",
         [
@@ -291,7 +321,21 @@ def test_post_should_respond_already_exist(self, test_client, body):
                     "extra": None,
                     "host": None,
                     "login": None,
-                    "password": "test-password",
+                    "password": "***",
+                    "port": None,
+                    "schema": None,
+                },
+            ),
+            (
+                {"connection_id": TEST_CONN_ID, "conn_type": TEST_CONN_TYPE, "password": "?>@#+!_%()#"},
+                {
+                    "connection_id": TEST_CONN_ID,
+                    "conn_type": TEST_CONN_TYPE,
+                    "description": None,
+                    "extra": None,
+                    "host": None,
+                    "login": None,
+                    "password": "***",
                     "port": None,
                     "schema": None,
                 },
@@ -307,23 +351,21 @@ def test_post_should_respond_already_exist(self, test_client, body):
                     "connection_id": TEST_CONN_ID,
                     "conn_type": TEST_CONN_TYPE,
                     "description": None,
-                    "extra": '{"password": "test-password"}',
+                    "extra": '{"password": "***"}',
                     "host": None,
                     "login": None,
-                    "password": "A!rF|0wi$aw3s0m3",
+                    "password": "***",
                     "port": None,
                     "schema": None,
                 },
             ),
         ],
     )
-    def test_post_should_response_201_password_not_masked(
-        self, test_client, body, expected_response, session
-    ):
+    def test_post_should_response_201_redacted_password(self, test_client, body, expected_response, session):
         response = test_client.post("/connections", json=body)
         assert response.status_code == 201
         assert response.json() == expected_response
-        _check_last_log(session, dag_id=None, event="post_connection", logical_date=None)
+        _check_last_log(session, dag_id=None, event="post_connection", logical_date=None, check_masked=True)
 
 
 class TestPatchConnection(TestConnectionEndpoint):
@@ -733,7 +775,22 @@ def test_patch_should_respond_404(self, test_client, body):
                     "extra": None,
                     "host": "some_host_a",
                     "login": "some_login",
-                    "password": "test-password",
+                    "password": "***",
+                    "port": 8080,
+                    "schema": None,
+                },
+                {"update_mask": ["password"]},
+            ),
+            (
+                {"connection_id": TEST_CONN_ID, "conn_type": TEST_CONN_TYPE, "password": "?>@#+!_%()#"},
+                {
+                    "connection_id": TEST_CONN_ID,
+                    "conn_type": TEST_CONN_TYPE,
+                    "description": "some_description_a",
+                    "extra": None,
+                    "host": "some_host_a",
+                    "login": "some_login",
+                    "password": "***",
                     "port": 8080,
                     "schema": None,
                 },
@@ -750,10 +807,10 @@ def test_patch_should_respond_404(self, test_client, body):
                     "connection_id": TEST_CONN_ID,
                     "conn_type": TEST_CONN_TYPE,
                     "description": "some_description_a",
-                    "extra": '{"password": "test-password"}',
+                    "extra": '{"password": "***"}',
                     "host": "some_host_a",
                     "login": "some_login",
-                    "password": "A!rF|0wi$aw3s0m3",
+                    "password": "***",
                     "port": 8080,
                     "schema": None,
                 },
@@ -761,14 +818,14 @@ def test_patch_should_respond_404(self, test_client, body):
             ),
         ],
     )
-    def test_patch_should_response_200_password_not_masked(
+    def test_patch_should_response_200_redacted_password(
         self, test_client, session, body, expected_response, update_mask
     ):
         self.create_connections()
         response = test_client.patch(f"/connections/{TEST_CONN_ID}", json=body, params=update_mask)
         assert response.status_code == 200
         assert response.json() == expected_response
-        _check_last_log(session, dag_id=None, event="patch_connection", logical_date=None)
+        _check_last_log(session, dag_id=None, event="patch_connection", logical_date=None, check_masked=True)
 
 
 class TestConnection(TestConnectionEndpoint):