Make mongodb work without admin privelege (#2023)

Author: artem-shelkovnikov

connectors/sources/mongo.py

@@ -12,6 +12,7 @@
 from bson import DBRef, Decimal128, ObjectId
 from fastjsonschema import JsonSchemaValueException
 from motor.motor_asyncio import AsyncIOMotorClient
+from pymongo.errors import OperationFailure
 
 from connectors.filtering.validation import (
     AdvancedRulesValidator,
@@ -256,24 +257,49 @@ async def validate_config(self):
         await super().validate_config()
 
         client = self.client
+
+        user = self.configuration["user"]
         configured_database_name = self.configuration["database"]
         configured_collection_name = self.configuration["collection"]
 
-        existing_database_names = await client.list_database_names()
+        # First check if collection is accessible
+        try:
+            # This works on both standalone and Managed mongo in the same way
+            await client[configured_database_name].validate_collection(
+                configured_collection_name
+            )
+            return
+        except OperationFailure:
+            self._logger.warning(
+                f"Unable to access '{configured_database_name}.{configured_collection_name}' as user '{user}'"
+            )
+
+        # If it's not accessible, try to make a good user-friendly error message
+        try:
+            # We will try to access some databases/collections to give a friendly message
+            # That will suggest the name of existing collection - but only if the user
+            # that we use to log in into MongoDB has access to it
+            existing_database_names = await client.list_database_names()
 
-        self._logger.debug(f"Existing databases: {existing_database_names}")
+            self._logger.debug(f"Existing databases: {existing_database_names}")
 
-        if configured_database_name not in existing_database_names:
-            msg = f"Database ({configured_database_name}) does not exist. Existing databases: {', '.join(existing_database_names)}"
-            raise ConfigurableFieldValueError(msg)
+            if configured_database_name not in existing_database_names:
+                msg = f"Database '{configured_database_name}' does not exist. Existing databases: {', '.join(existing_database_names)}"
+                raise ConfigurableFieldValueError(msg)
 
-        database = client[configured_database_name]
+            database = client[configured_database_name]
 
-        existing_collection_names = await database.list_collection_names()
-        self._logger.debug(
-            f"Existing collections in {configured_database_name}: {existing_collection_names}"
-        )
+            existing_collection_names = await database.list_collection_names()
+            self._logger.debug(
+                f"Existing collections in {configured_database_name}: {existing_collection_names}"
+            )
 
-        if configured_collection_name not in existing_collection_names:
-            msg = f"Collection ({configured_collection_name}) does not exist within database {configured_database_name}. Existing collections: {', '.join(existing_collection_names)}"
-            raise ConfigurableFieldValueError(msg)
+            if configured_collection_name not in existing_collection_names:
+                msg = f"Collection '{configured_collection_name}' does not exist within database '{configured_database_name}'. Existing collections: {', '.join(existing_collection_names)}"
+                raise ConfigurableFieldValueError(msg)
+        except OperationFailure as e:
+            # This happens if the user has no access to operations to list collection/database names
+            # Managed MongoDB never gets here, but if we're running against a standalone mongo
+            # Then this code can trigger
+            msg = f"Database '{configured_database_name}' or collection '{configured_collection_name}' is not accessible by user '{user}'. Verify that these database and collection exist, and specified user has access to it"
+            raise ConfigurableFieldValueError(msg) from e

tests/sources/test_mongo.py

@@ -12,7 +12,7 @@
 import pytest
 from bson import DBRef, ObjectId
 from bson.decimal128 import Decimal128
-from pymongo.errors import ServerSelectionTimeoutError
+from pymongo.errors import OperationFailure
 
 from connectors.protocol import Filter
 from connectors.source import ConfigurableFieldValueError
@@ -261,7 +261,13 @@ def future_with_result(result):
 
 
 @pytest.mark.asyncio
-async def test_validate_config_when_database_name_invalid_then_raises_exception():
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.validate_collection",
+    side_effect=OperationFailure("Unauthorized"),
+)
+async def test_validate_config_when_database_name_invalid_then_raises_exception(
+    patch_validate_collection,
+):
     server_database_names = ["hello", "world"]
     configured_database_name = "something"
 
@@ -283,7 +289,13 @@ async def test_validate_config_when_database_name_invalid_then_raises_exception(
 
 
 @pytest.mark.asyncio
-async def test_validate_config_when_collection_name_invalid_then_raises_exception():
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.validate_collection",
+    side_effect=OperationFailure("Unauthorized"),
+)
+async def test_validate_config_when_collection_name_invalid_then_raises_exception(
+    patch_validate_collection,
+):
     server_database_names = ["hello"]
     server_collection_names = ["first", "second"]
     configured_database_name = "hello"
@@ -310,25 +322,76 @@ async def test_validate_config_when_collection_name_invalid_then_raises_exceptio
 
 
 @pytest.mark.asyncio
-async def test_validate_config_when_configuration_valid_then_does_not_raise():
-    server_database_names = ["hello"]
-    server_collection_names = ["first", "second"]
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.validate_collection",
+    side_effect=OperationFailure("Unauthorized"),
+)
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorClient.list_database_names",
+    return_value=future_with_result(["hello"]),
+)
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.list_collection_names",
+    return_value=future_with_result(["first"]),
+)
+async def test_validate_config_when_collection_access_unauthorized(
+    patch_validate_collection, patch_list_database_names, patch_list_collection_names
+):
     configured_database_name = "hello"
     configured_collection_name = "second"
 
-    with mock.patch(
-        "motor.motor_asyncio.AsyncIOMotorClient.list_database_names",
-        return_value=future_with_result(server_database_names),
-    ), mock.patch(
-        "motor.motor_asyncio.AsyncIOMotorDatabase.list_collection_names",
-        return_value=future_with_result(server_collection_names),
-    ):
+    with pytest.raises(ConfigurableFieldValueError) as e:
+        async with create_mongo_source(
+            database=configured_database_name,
+            collection=configured_collection_name,
+        ) as source:
+            await source.validate_config()
+
+    assert e is not None
+
+
+@pytest.mark.asyncio
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.validate_collection",
+    side_effect=OperationFailure("Unauthorized"),
+)
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorClient.list_database_names",
+    side_effect=OperationFailure("Unauthorized"),
+)
+async def test_validate_config_when_collection_access_unauthorized_and_no_admin_access(
+    patch_validate_collection, patch_list_database_names
+):
+    configured_database_name = "hello"
+    configured_collection_name = "second"
+
+    with pytest.raises(ConfigurableFieldValueError) as e:
         async with create_mongo_source(
             database=configured_database_name,
             collection=configured_collection_name,
         ) as source:
             await source.validate_config()
 
+    assert e is not None
+
+
+@pytest.mark.asyncio
+@mock.patch(
+    "motor.motor_asyncio.AsyncIOMotorDatabase.validate_collection",
+    return_value=future_with_result(None),
+)
+async def test_validate_config_when_configuration_valid_then_does_not_raise(
+    patch_validate_connection,
+):
+    configured_database_name = "hello"
+    configured_collection_name = "second"
+
+    async with create_mongo_source(
+        database=configured_database_name,
+        collection=configured_collection_name,
+    ) as source:
+        await source.validate_config()
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
@@ -347,22 +410,6 @@ async def test_serialize(raw, output):
         assert source.serialize(raw) == output
 
 
-@pytest.mark.asyncio
-@mock.patch("ssl.SSLContext.load_verify_locations")
-async def test_ssl_connection_with_invalid_certificate(mock_ssl):
-    mock_ssl.return_value = True
-    async with create_mongo_source(
-        ssl_enabled=True,
-        ssl_ca="-----BEGIN CERTIFICATE----- Invalid-Certificate -----END CERTIFICATE-----",
-    ) as source:
-        with mock.patch(
-            "motor.motor_asyncio.AsyncIOMotorClient.list_database_names",
-            side_effect=ServerSelectionTimeoutError(),
-        ):
-            with pytest.raises(ServerSelectionTimeoutError):
-                await source.validate_config()
-
-
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "certificate_value, tls_insecure",
@@ -379,19 +426,12 @@ async def test_ssl_connection_with_invalid_certificate(mock_ssl):
         ),
     ],
 )
-@mock.patch("ssl.SSLContext.load_verify_locations")
+@mock.patch("ssl.SSLContext.load_verify_locations", return_value=True)
 async def test_ssl_with_successful_connection(
     mock_ssl, certificate_value, tls_insecure
 ):
-    mock_ssl.return_value = True
     async with create_mongo_source(
-        ssl_enabled=True, ssl_ca=certificate_value, tls_insecure=tls_insecure
-    ) as source:
-        with mock.patch(
-            "motor.motor_asyncio.AsyncIOMotorClient.list_database_names",
-            return_value=future_with_result(["db"]),
-        ), mock.patch(
-            "motor.motor_asyncio.AsyncIOMotorDatabase.list_collection_names",
-            return_value=future_with_result(["col"]),
-        ):
-            await source.validate_config()
+        ssl_enabled=True,
+        ssl_ca="-----BEGIN CERTIFICATE----- Invalid-Certificate -----END CERTIFICATE-----",
+    ):
+        assert True