iceberg/io/opendal/
s3.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::sync::Arc;
20
21use async_trait::async_trait;
22use opendal::services::S3Config;
23use opendal::{Configurator, Operator};
24pub use reqsign::{AwsCredential, AwsCredentialLoad};
25use reqwest::Client;
26use url::Url;
27
28use crate::io::config::{
29    CLIENT_REGION, S3_ACCESS_KEY_ID, S3_ALLOW_ANONYMOUS, S3_ASSUME_ROLE_ARN,
30    S3_ASSUME_ROLE_EXTERNAL_ID, S3_ASSUME_ROLE_SESSION_NAME, S3_DISABLE_CONFIG_LOAD,
31    S3_DISABLE_EC2_METADATA, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY,
32    S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE,
33};
34use crate::io::is_truthy;
35use crate::{Error, ErrorKind, Result};
36
37/// Parse iceberg props to s3 config.
38pub(crate) fn s3_config_parse(mut m: HashMap<String, String>) -> Result<S3Config> {
39    let mut cfg = S3Config::default();
40    if let Some(endpoint) = m.remove(S3_ENDPOINT) {
41        cfg.endpoint = Some(endpoint);
42    };
43    if let Some(access_key_id) = m.remove(S3_ACCESS_KEY_ID) {
44        cfg.access_key_id = Some(access_key_id);
45    };
46    if let Some(secret_access_key) = m.remove(S3_SECRET_ACCESS_KEY) {
47        cfg.secret_access_key = Some(secret_access_key);
48    };
49    if let Some(session_token) = m.remove(S3_SESSION_TOKEN) {
50        cfg.session_token = Some(session_token);
51    };
52    if let Some(region) = m.remove(S3_REGION) {
53        cfg.region = Some(region);
54    };
55    if let Some(region) = m.remove(CLIENT_REGION) {
56        cfg.region = Some(region);
57    };
58    if let Some(path_style_access) = m.remove(S3_PATH_STYLE_ACCESS) {
59        cfg.enable_virtual_host_style = !is_truthy(path_style_access.to_lowercase().as_str());
60    };
61    if let Some(arn) = m.remove(S3_ASSUME_ROLE_ARN) {
62        cfg.role_arn = Some(arn);
63    }
64    if let Some(external_id) = m.remove(S3_ASSUME_ROLE_EXTERNAL_ID) {
65        cfg.external_id = Some(external_id);
66    };
67    if let Some(session_name) = m.remove(S3_ASSUME_ROLE_SESSION_NAME) {
68        cfg.role_session_name = Some(session_name);
69    };
70    let s3_sse_key = m.remove(S3_SSE_KEY);
71    if let Some(sse_type) = m.remove(S3_SSE_TYPE) {
72        match sse_type.to_lowercase().as_str() {
73            // No Server Side Encryption
74            "none" => {}
75            // S3 SSE-S3 encryption (S3 managed keys). https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingServerSideEncryption.html
76            "s3" => {
77                cfg.server_side_encryption = Some("AES256".to_string());
78            }
79            // S3 SSE KMS, either using default or custom KMS key. https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingKMSEncryption.html
80            "kms" => {
81                cfg.server_side_encryption = Some("aws:kms".to_string());
82                cfg.server_side_encryption_aws_kms_key_id = s3_sse_key;
83            }
84            // S3 SSE-C, using customer managed keys. https://docs.aws.amazon.com/AmazonS3/latest/dev/ServerSideEncryptionCustomerKeys.html
85            "custom" => {
86                cfg.server_side_encryption_customer_algorithm = Some("AES256".to_string());
87                cfg.server_side_encryption_customer_key = s3_sse_key;
88                cfg.server_side_encryption_customer_key_md5 = m.remove(S3_SSE_MD5);
89            }
90            _ => {
91                return Err(Error::new(
92                    ErrorKind::DataInvalid,
93                    format!(
94                        "Invalid {S3_SSE_TYPE}: {sse_type}. Expected one of (custom, kms, s3, none)"
95                    ),
96                ));
97            }
98        }
99    };
100
101    if let Some(allow_anonymous) = m.remove(S3_ALLOW_ANONYMOUS)
102        && is_truthy(allow_anonymous.to_lowercase().as_str())
103    {
104        cfg.allow_anonymous = true;
105    }
106    if let Some(disable_ec2_metadata) = m.remove(S3_DISABLE_EC2_METADATA)
107        && is_truthy(disable_ec2_metadata.to_lowercase().as_str())
108    {
109        cfg.disable_ec2_metadata = true;
110    };
111    if let Some(disable_config_load) = m.remove(S3_DISABLE_CONFIG_LOAD)
112        && is_truthy(disable_config_load.to_lowercase().as_str())
113    {
114        cfg.disable_config_load = true;
115    };
116
117    Ok(cfg)
118}
119
120/// Build new opendal operator from give path.
121pub(crate) fn s3_config_build(
122    cfg: &S3Config,
123    customized_credential_load: &Option<CustomAwsCredentialLoader>,
124    path: &str,
125) -> Result<Operator> {
126    let url = Url::parse(path)?;
127    let bucket = url.host_str().ok_or_else(|| {
128        Error::new(
129            ErrorKind::DataInvalid,
130            format!("Invalid s3 url: {path}, missing bucket"),
131        )
132    })?;
133
134    let mut builder = cfg
135        .clone()
136        .into_builder()
137        // Set bucket name.
138        .bucket(bucket);
139
140    if let Some(customized_credential_load) = customized_credential_load {
141        builder = builder
142            .customized_credential_load(customized_credential_load.clone().into_opendal_loader());
143    }
144
145    Ok(Operator::new(builder)?.finish())
146}
147
148/// Custom AWS credential loader.
149/// This can be used to load credentials from a custom source, such as the AWS SDK.
150///
151/// This should be set as an extension on `FileIOBuilder`.
152#[derive(Clone)]
153pub struct CustomAwsCredentialLoader(Arc<dyn AwsCredentialLoad>);
154
155impl std::fmt::Debug for CustomAwsCredentialLoader {
156    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
157        f.debug_struct("CustomAwsCredentialLoader")
158            .finish_non_exhaustive()
159    }
160}
161
162impl CustomAwsCredentialLoader {
163    /// Create a new custom AWS credential loader.
164    pub fn new(loader: Arc<dyn AwsCredentialLoad>) -> Self {
165        Self(loader)
166    }
167
168    /// Convert this loader into an opendal compatible loader for customized AWS credentials.
169    pub fn into_opendal_loader(self) -> Box<dyn AwsCredentialLoad> {
170        Box::new(self)
171    }
172}
173
174#[async_trait]
175impl AwsCredentialLoad for CustomAwsCredentialLoader {
176    async fn load_credential(&self, client: Client) -> anyhow::Result<Option<AwsCredential>> {
177        self.0.load_credential(client).await
178    }
179}
iceberg/io/opendal/s3.rs

iceberg/io/opendal/
s3.rs
