Core: Validate implementation classes loaded from properties

pyiceberg/catalog/__init__.py

@@ -331,6 +331,8 @@ def _import_catalog(name: str, catalog_impl: str, properties: Properties) -> Cat
         module_name, class_name = ".".join(path_parts[:-1]), path_parts[-1]
         module = importlib.import_module(module_name)
         class_ = getattr(module, class_name)
+        if not isinstance(class_, type) or not issubclass(class_, Catalog):
+            raise ValueError(f"py-catalog-impl should be a subclass of Catalog, got: {catalog_impl}")
         return class_(name, **properties)
     except ModuleNotFoundError:
         logger.warning(f"Could not initialize Catalog: {catalog_impl}", exc_info=logger.isEnabledFor(logging.DEBUG))

pyiceberg/catalog/rest/auth.py

@@ -383,6 +383,8 @@ def create(cls, class_or_name: str, config: dict[str, Any]) -> AuthManager:
                 manager_cls = getattr(module, class_name)
             except Exception as err:
                 raise ValueError(f"Could not load AuthManager class for '{class_or_name}'") from err
+            if not isinstance(manager_cls, type) or not issubclass(manager_cls, AuthManager):
+                raise ValueError(f"auth.impl should be a subclass of AuthManager, got: {class_or_name}")
 
         return manager_cls(**config)
 

pyiceberg/io/__init__.py

@@ -340,6 +340,8 @@ def _import_file_io(io_impl: str, properties: Properties) -> FileIO | None:
         module_name, class_name = ".".join(path_parts[:-1]), path_parts[-1]
         module = importlib.import_module(module_name)
         class_ = getattr(module, class_name)
+        if not isinstance(class_, type) or not issubclass(class_, FileIO):
+            raise ValueError(f"py-io-impl should be a subclass of FileIO, got: {io_impl}")
         return class_(properties)
     except ModuleNotFoundError:
         logger.warning(f"Could not initialize FileIO: {io_impl}", exc_info=logger.isEnabledFor(logging.DEBUG))

pyiceberg/io/pyarrow.py

@@ -232,6 +232,8 @@ def _import_retry_strategy(impl: str) -> S3RetryStrategy | None:
         module_name, class_name = ".".join(path_parts[:-1]), path_parts[-1]
         module = importlib.import_module(module_name)
         class_ = getattr(module, class_name)
+        if not isinstance(class_, type) or not issubclass(class_, S3RetryStrategy):
+            raise ValueError(f"retry-strategy-impl should be a subclass of S3RetryStrategy, got: {impl}")
         return class_()
     except (ModuleNotFoundError, AttributeError):
         warnings.warn(f"Could not initialize S3 retry strategy: {impl}", stacklevel=2)

pyiceberg/table/locations.py

@@ -178,6 +178,10 @@ def _import_location_provider(
         module_name, class_name = ".".join(path_parts[:-1]), path_parts[-1]
         module = importlib.import_module(module_name)
         class_ = getattr(module, class_name)
+        if not isinstance(class_, type) or not issubclass(class_, LocationProvider):
+            raise ValueError(
+                f"write.py-location-provider.impl should be a subclass of LocationProvider, got: {location_provider_impl}"
+            )
         return class_(table_location, table_properties)
     except ModuleNotFoundError:
         logger.warning(

tests/catalog/test_base.py

@@ -47,6 +47,11 @@ def test_load_catalog_impl_not_full_path() -> None:
     assert "py-catalog-impl should be full path (module.CustomCatalog), got: CustomCatalog" in str(exc_info.value)
 
 
+def test_load_catalog_impl_wrong_type() -> None:
+    with pytest.raises(ValueError, match="py-catalog-impl should be a subclass of Catalog"):
+        load_catalog("catalog", **{"py-catalog-impl": "pyiceberg.io.FileIO"})
+
+
 def test_load_catalog_impl_does_not_exist() -> None:
     with pytest.raises(ValueError) as exc_info:
         load_catalog("catalog", **{"py-catalog-impl": "pyiceberg.does.not.exist.Catalog"})

tests/catalog/test_rest.py

@@ -2501,6 +2501,24 @@ def test_rest_catalog_with_custom_auth_type() -> None:
     assert "Could not load AuthManager class for 'dummy.nonexistent.package'" in str(e.value)
 
 
+def test_rest_catalog_with_custom_auth_type_wrong_type() -> None:
+    # Given
+    catalog_properties = {
+        "uri": TEST_URI,
+        "auth": {
+            "type": "custom",
+            "impl": "pyiceberg.io.FileIO",
+            "custom": {
+                "property1": "one",
+                "property2": "two",
+            },
+        },
+    }
+    with pytest.raises(ValueError) as e:
+        RestCatalog("rest", **catalog_properties)  # type: ignore
+    assert "auth.impl should be a subclass of AuthManager, got: pyiceberg.io.FileIO" in str(e.value)
+
+
 def test_rest_catalog_with_custom_basic_auth_type(rest_mock: Mocker) -> None:
     # Given
     catalog_properties = {

tests/catalog/test_rest_auth.py

@@ -22,7 +22,14 @@
 import requests
 from requests_mock import Mocker
 
-from pyiceberg.catalog.rest.auth import AuthManagerAdapter, BasicAuthManager, EntraAuthManager, GoogleAuthManager, NoopAuthManager
+from pyiceberg.catalog.rest.auth import (
+    AuthManagerAdapter,
+    AuthManagerFactory,
+    BasicAuthManager,
+    EntraAuthManager,
+    GoogleAuthManager,
+    NoopAuthManager,
+)
 
 TEST_URI = "https://iceberg-test-catalog/"
 GOOGLE_CREDS_URI = "https://oauth2.googleapis.com/token"
@@ -61,6 +68,11 @@ def test_noop_auth_header(rest_mock: Mocker) -> None:
     assert "Authorization" not in actual_headers
 
 
+def test_auth_manager_factory_wrong_type() -> None:
+    with pytest.raises(ValueError, match="auth.impl should be a subclass of AuthManager"):
+        AuthManagerFactory.create("pyiceberg.io.FileIO", {})
+
+
 def test_basic_auth_header(rest_mock: Mocker) -> None:
     username = "testuser"
     password = "testpassword"

tests/io/test_io.py

@@ -280,6 +280,11 @@ def test_import_file_io() -> None:
     assert isinstance(_import_file_io(ARROW_FILE_IO, {}), PyArrowFileIO)
 
 
+def test_import_file_io_wrong_type() -> None:
+    with pytest.raises(ValueError, match="py-io-impl should be a subclass of FileIO"):
+        _import_file_io("pyiceberg.table.locations.SimpleLocationProvider", {})
+
+
 def test_import_file_io_does_not_exist(caplog: Any) -> None:
     import logging
 

tests/io/test_pyarrow.py

@@ -3309,6 +3309,12 @@ def test_retry_strategy() -> None:
         io.new_input("s3://bucket/path/to/file")
 
 
+def test_retry_strategy_wrong_type() -> None:
+    io = PyArrowFileIO(properties={S3_RETRY_STRATEGY_IMPL: "pyiceberg.io.FileIO"})
+    with pytest.raises(ValueError, match="retry-strategy-impl should be a subclass of S3RetryStrategy"):
+        io.new_input("s3://bucket/path/to/file")
+
+
 def test_retry_strategy_not_found() -> None:
     io = PyArrowFileIO(properties={S3_RETRY_STRATEGY_IMPL: "pyiceberg.DoesNotExist"})
     with pytest.warns(UserWarning, match="Could not initialize S3 retry strategy: pyiceberg.DoesNotExist"):

tests/table/test_locations.py

@@ -60,6 +60,14 @@ def test_custom_location_provider() -> None:
     assert provider.new_data_location("my_file") == "custom_location_provider/my_file"
 
 
+def test_custom_location_provider_wrong_type() -> None:
+    with pytest.raises(ValueError, match="write.py-location-provider.impl should be a subclass of LocationProvider"):
+        load_location_provider(
+            table_location="table_location",
+            table_properties={"write.py-location-provider.impl": "pyiceberg.io.FileIO"},
+        )
+
+
 def test_custom_location_provider_single_path() -> None:
     with pytest.raises(ValueError, match=r"write\.py-location-provider\.impl should be full path"):
         load_location_provider(table_location="table_location", table_properties={"write.py-location-provider.impl": "not_found"})