Reject connection secrets without a connection type

shared/secrets_backend/src/airflow_shared/secrets_backend/base.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import inspect
+import json
 from abc import ABC
 from collections.abc import Callable
 
@@ -122,6 +123,11 @@ def _get_connection_class(self) -> type:
     def _deserialize_connection_value(conn_class: type, conn_id: str, value: str):
         value = value.strip()
         if value[0] == "{":
+            connection_data = json.loads(value)
+            conn_type = connection_data.get("conn_type")
+            uri = connection_data.get("uri")
+            if not (isinstance(conn_type, str) and conn_type.strip() or isinstance(uri, str) and uri.strip()):
+                raise ValueError("Connection secret JSON must include a non-empty 'conn_type' or 'uri'.")
             return conn_class.from_json(value=value, conn_id=conn_id)  # type: ignore[attr-defined]
 
         # TODO: Only sdk has from_uri defined on it. Is it worthwhile developing the core path or not?

shared/secrets_backend/tests/secrets_backend/test_base.py

@@ -124,6 +124,39 @@ def test_deserialize_connection_json(self, sample_conn_json):
         assert conn.conn_id == "test_conn"
         assert conn._kwargs["conn_type"] == "mysql"
 
+    @pytest.mark.parametrize(
+        "value",
+        [
+            '{"host": "example.com"}',
+            '{"conn_type": null}',
+            '{"conn_type": ""}',
+            '{"conn_type": "   "}',
+        ],
+    )
+    def test_deserialize_connection_json_requires_conn_type(self, value):
+        backend = _TestBackend()
+        backend._set_connection_class(MockConnection)
+
+        with pytest.raises(
+            ValueError, match="Connection secret JSON must include a non-empty 'conn_type' or 'uri'."
+        ):
+            backend.deserialize_connection("test_conn", value)
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            '{"uri": "http://example.com"}',
+            '{"conn_type": null, "uri": "http://example.com"}',
+        ],
+    )
+    def test_deserialize_connection_json_with_uri_does_not_require_conn_type(self, value):
+        backend = _TestBackend()
+        backend._set_connection_class(MockConnection)
+
+        conn = backend.deserialize_connection("test_conn", value)
+
+        assert conn.uri == "http://example.com"
+
     def test_deserialize_connection_uri(self, sample_conn_uri):
         """Test deserialize_connection with URI format through _TestBackend."""
         backend = _TestBackend()