diff --git a/pyiceberg/io/__init__.py b/pyiceberg/io/__init__.py index c44e105e62..9cae219f7f 100644 --- a/pyiceberg/io/__init__.py +++ b/pyiceberg/io/__init__.py @@ -69,6 +69,7 @@ def _is_local_path(path: str) -> bool: S3_SESSION_TOKEN = "s3.session-token" S3_REGION = "s3.region" S3_RESOLVE_REGION = "s3.resolve-region" +S3_SSL_CA_CERT = "s3.ssl.ca-cert" S3_PROXY_URI = "s3.proxy-uri" S3_CONNECT_TIMEOUT = "s3.connect-timeout" S3_REQUEST_TIMEOUT = "s3.request-timeout" diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 1e59107da3..e7cc5c5926 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -116,6 +116,7 @@ S3_ROLE_SESSION_NAME, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, + S3_SSL_CA_CERT, FileIO, InputFile, InputStream, @@ -453,6 +454,16 @@ def _initialize_fs(self, scheme: str, netloc: str | None = None) -> FileSystem: else: raise ValueError(f"Unrecognized filesystem type in URI: {scheme}") + def _set_tls_ca_file_path(self, client_kwargs: dict[str, Any]) -> None: + if tls_ca_file_path := self.properties.get(S3_SSL_CA_CERT): + min_pyarrow_version_supporting_tls_ca_file_path = (21, 0) + if tuple(map(int, pa.__version__.split(".")[:2])) < min_pyarrow_version_supporting_tls_ca_file_path: + raise ImportError( + "pyarrow version >= 21.0.0 required for S3FileSystem tls_ca_file_path support, " + f"but found version {pa.__version__}." + ) + client_kwargs["tls_ca_file_path"] = tls_ca_file_path + def _initialize_oss_fs(self) -> FileSystem: from pyarrow.fs import S3FileSystem @@ -483,6 +494,8 @@ def _initialize_oss_fs(self) -> FileSystem: if s3_anonymous := self.properties.get(S3_ANONYMOUS): client_kwargs["anonymous"] = strtobool(s3_anonymous) + self._set_tls_ca_file_path(client_kwargs) + return S3FileSystem(**client_kwargs) def _initialize_s3_fs(self, netloc: str | None) -> FileSystem: @@ -537,6 +550,8 @@ def _initialize_s3_fs(self, netloc: str | None) -> FileSystem: if s3_anonymous := self.properties.get(S3_ANONYMOUS): client_kwargs["anonymous"] = strtobool(s3_anonymous) + self._set_tls_ca_file_path(client_kwargs) + return S3FileSystem(**client_kwargs) def _initialize_azure_fs(self) -> FileSystem: diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index 892d8e54eb..9a303ea084 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -63,7 +63,7 @@ Or, ) from pyiceberg.expressions.literals import literal -from pyiceberg.io import S3_RETRY_STRATEGY_IMPL, InputStream, OutputStream, load_file_io +from pyiceberg.io import S3_RETRY_STRATEGY_IMPL, S3_SSL_CA_CERT, InputStream, OutputStream, load_file_io from pyiceberg.io.pyarrow import ( ICEBERG_SCHEMA, PYARROW_PARQUET_FIELD_ID_KEY, @@ -382,6 +382,7 @@ def test_pyarrow_s3_session_properties() -> None: "s3.secret-access-key": "password", "s3.region": "us-east-1", "s3.session-token": "s3.session-token", + S3_SSL_CA_CERT: "/path/to/ca.pem", **UNIFIED_AWS_SESSION_PROPERTIES, } @@ -399,9 +400,26 @@ def test_pyarrow_s3_session_properties() -> None: secret_key="password", region="us-east-1", session_token="s3.session-token", + tls_ca_file_path="/path/to/ca.pem", ) +def test_pyarrow_s3_ssl_ca_cert_requires_supported_pyarrow_version() -> None: + session_properties: Properties = { + S3_SSL_CA_CERT: "/path/to/ca.pem", + } + + with ( + patch("pyiceberg.io.pyarrow.pa.__version__", "20.0.0"), + patch("pyarrow.fs.S3FileSystem"), + patch("pyarrow.fs.resolve_s3_region") as mock_s3_region_resolver, + ): + mock_s3_region_resolver.side_effect = OSError("S3 bucket is not found") + + with pytest.raises(ImportError, match="pyarrow version >= 21.0.0 required"): + PyArrowFileIO(properties=session_properties).new_input(location=f"s3://warehouse/{uuid.uuid4()}") + + def test_pyarrow_s3_session_properties_with_anonymous() -> None: session_properties: Properties = { "s3.anonymous": "true",