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 fee330e1fd1d6..4491487c3bb10 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
@@ -136,3 +136,26 @@ class ConnectionBody(StrictBaseModel):
     port: int | None = Field(default=None)
     password: str | None = Field(default=None)
     extra: str | None = Field(default=None)
+
+    @field_validator("extra")
+    @classmethod
+    def validate_extra(cls, v: str | None) -> str | None:
+        """
+        Validate that `extra` field is a JSON-encoded Python dict.
+
+        If `extra` field is not a valid JSON, it will be returned as is.
+        """
+        if v is None:
+            return v
+        if v == "":
+            return "{}"  # Backward compatibility: treat "" as empty JSON object
+        try:
+            extra_dict = json.loads(v)
+            if not isinstance(extra_dict, dict):
+                raise ValueError("The `extra` field must be a valid JSON object (e.g., {'key': 'value'})")
+        except json.JSONDecodeError:
+            raise ValueError(
+                "The `extra` field must be a valid JSON object (e.g., {'key': 'value'}), "
+                "but encountered non-JSON in `extra` field"
+            )
+        return v
diff --git a/airflow-core/src/airflow/api_fastapi/logging/decorators.py b/airflow-core/src/airflow/api_fastapi/logging/decorators.py
index 1e5ad9fbd930e..ad65c8caac8d3 100644
--- a/airflow-core/src/airflow/api_fastapi/logging/decorators.py
+++ b/airflow-core/src/airflow/api_fastapi/logging/decorators.py
@@ -40,9 +40,12 @@ def _mask_connection_fields(extra_fields):
     for k, v in extra_fields.items():
         if k == "extra" and v:
             try:
-                extra = json.loads(v)
-                extra = {k: secrets_masker.redact(v, k) for k, v in extra.items()}
-                result[k] = dict(extra)
+                parsed_extra = json.loads(v)
+                if isinstance(parsed_extra, dict):
+                    masked_extra = {ek: secrets_masker.redact(ev, ek) for ek, ev in parsed_extra.items()}
+                    result[k] = masked_extra
+                else:
+                    result[k] = "Expected JSON object in `extra` field, got non-dict JSON"
             except json.JSONDecodeError:
                 result[k] = "Encountered non-JSON in `extra` field"
         else:
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 5ca5f6756a6bd..c023d446a427c 100644
--- a/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Connections/ConnectionForm.tsx
@@ -98,7 +98,16 @@ const ConnectionForm = ({
 
   const validateAndPrettifyJson = (value: string) => {
     try {
-      const parsedJson = JSON.parse(value) as JSON;
+      if (value.trim() === "") {
+        setErrors((prev) => ({ ...prev, conf: undefined }));
+
+        return value;
+      }
+      const parsedJson = JSON.parse(value) as Record<string, unknown>;
+
+      if (typeof parsedJson !== "object" || Array.isArray(parsedJson)) {
+        throw new TypeError('extra fields must be a valid JSON object (e.g., {"key": "value"})');
+      }
 
       setErrors((prev) => ({ ...prev, conf: undefined }));
       const formattedJson = JSON.stringify(parsedJson, undefined, 2);
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 5f210278be81c..9afac754292c9 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
@@ -1245,3 +1245,15 @@ def test_should_respond_401(self, unauthenticated_test_client):
     def test_should_respond_403(self, unauthorized_test_client):
         response = unauthorized_test_client.patch("/connections", json={})
         assert response.status_code == 403
+
+
+class TestPostConnectionExtraBackwardCompatibility(TestConnectionEndpoint):
+    def test_post_should_accept_empty_string_as_extra(self, test_client, session):
+        body = {"connection_id": TEST_CONN_ID, "conn_type": TEST_CONN_TYPE, "extra": ""}
+
+        response = test_client.post("/connections", json=body)
+        assert response.status_code == 201
+
+        connection = session.query(Connection).filter_by(conn_id=TEST_CONN_ID).first()
+        assert connection is not None
+        assert connection.extra == "{}"  # Backward compatibility: treat "" as empty JSON object