You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/08/07 13:50:06 UTC
[GitHub] [arrow-rs] tustvold opened a new pull request, #2352: Replace rusoto with custom implementation (#2176)
tustvold opened a new pull request, #2352:
URL: https://github.com/apache/arrow-rs/pull/2352
_Draft as builds on #2344, needs further testing, and I may split up_
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #2350
Part of #2176
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
See #2176
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
Replaces rusoto with a custom implementation based on reqwest
# Are there any user-facing changes?
Not that I'm aware of, but there is a possibility
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1209213210
https://github.com/influxdata/influxdb_iox/pull/5342 has been deployed and appears to be working correctly, testing both the Instance Metadata credentials and the WebIdentity
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on pull request #2352: Replace rusoto with custom implementation for AWS (#2176)
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1214883180
I don't have any objections, though I haven't reviewed this PR in detail either. I will confirm we deployed these changes in our IOx environment and they seem to be working great
cc @roeap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r940076078
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,493 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &'static str = "x-amz-date";
+const HASH_HEADER: &'static str = "x-amz-content-sha256";
+const TOKEN_HEADER: &'static str = "x-amz-security-token";
+const AUTH_HEADER: &'static str = "authorization";
+
+const ALL_HEADERS: &[&'static str; 4] =
+ &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
Review Comment:
Previous comment about this being ludicrous still stands :laughing:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r941048252
##########
object_store/src/aws/mod.rs:
##########
@@ -0,0 +1,631 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for S3
+//!
+//! ## Multi-part uploads
+//!
+//! Multi-part uploads can be initiated with the [ObjectStore::put_multipart] method.
+//! Data passed to the writer is automatically buffered to meet the minimum size
+//! requirements for a part. Multiple parts are uploaded concurrently.
+//!
+//! If the writer fails for any reason, you may have parts uploaded to AWS but not
+//! used that you may be charged for. Use the [ObjectStore::abort_multipart] method
+//! to abort the upload and drop those unneeded parts. In addition, you may wish to
+//! consider implementing [automatic cleanup] of unused parts that are older than one
+//! week.
+//!
+//! [automatic cleanup]: https://aws.amazon.com/blogs/aws/s3-lifecycle-management-update-support-for-multipart-uploads-and-delete-markers/
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{DateTime, Utc};
+use futures::stream::BoxStream;
+use futures::TryStreamExt;
+use snafu::{OptionExt, ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use tracing::info;
+
+use crate::aws::client::{S3Client, S3Config};
+use crate::aws::credential::{AwsCredential, CredentialProvider};
+use crate::multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart};
+use crate::{
+ GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Path, Result,
+ RetryConfig, StreamExt,
+};
+
+mod client;
+mod credential;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+ #[snafu(display("Last-Modified Header missing from response"))]
+ MissingLastModified,
+
+ #[snafu(display("Content-Length Header missing from response"))]
+ MissingContentLength,
+
+ #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+ InvalidLastModified {
+ last_modified: String,
+ source: chrono::ParseError,
+ },
+
+ #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+ InvalidContentLength {
+ content_length: String,
+ source: std::num::ParseIntError,
+ },
+
+ #[snafu(display("Missing region"))]
+ MissingRegion,
+
+ #[snafu(display("Missing bucket name"))]
+ MissingBucketName,
+
+ #[snafu(display("Missing AccessKeyId"))]
+ MissingAccessKeyId,
+
+ #[snafu(display("Missing SecretAccessKey"))]
+ MissingSecretAccessKey,
+
+ #[snafu(display("ETag Header missing from response"))]
+ MissingEtag,
+
+ #[snafu(display("Received header containing non-ASCII data"))]
+ BadHeader { source: reqwest::header::ToStrError },
+
+ #[snafu(display("Error reading token file: {}", source))]
+ ReadTokenFile { source: std::io::Error },
+}
+
+impl From<Error> for super::Error {
+ fn from(err: Error) -> Self {
+ Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ }
+ }
+}
+
+/// Interface for [Amazon S3](https://aws.amazon.com/s3/).
+#[derive(Debug)]
+pub struct AmazonS3 {
+ client: Arc<S3Client>,
Review Comment:
It helps with encapsulation, especially as many methods on `ObjectStore` map to some combination of actual upstream requests - e.g. the various types of Put request. Also made it easier to split into its own module
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r939671445
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,493 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &'static str = "x-amz-date";
+const HASH_HEADER: &'static str = "x-amz-content-sha256";
+const TOKEN_HEADER: &'static str = "x-amz-security-token";
+const AUTH_HEADER: &'static str = "authorization";
+
+const ALL_HEADERS: &[&'static str; 4] =
+ &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+ /// Sign a request <https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html>
+ fn sign(mut self, credential: &AwsCredential, region: &str, service: &str) -> Self {
+ // Hack around lack of access to underlying request
+ // https://github.com/seanmonstar/reqwest/issues/1212
+ let mut request = self
+ .try_clone()
+ .expect("not stream")
+ .build()
+ .expect("request valid");
+
+ let date = Utc::now();
+ let signer = RequestSigner {
+ date,
+ credential,
+ service,
+ region,
+ };
+
+ signer.sign(&mut request);
+
+ for header in ALL_HEADERS {
+ if let Some(val) = request.headers_mut().remove(*header) {
+ self = self.header(*header, val)
+ }
+ }
+ self
+ }
+}
+
+fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) -> ring::hmac::Tag {
+ let key = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, secret.as_ref());
+ ring::hmac::sign(&key, bytes.as_ref())
+}
+
+/// Computes the SHA256 digest of `body` returned as a hex encoded string
+fn hex_digest(bytes: &[u8]) -> String {
+ let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
+ hex_encode(digest.as_ref())
+}
+
+/// Returns `bytes` as a lower-case hex encoded string
+fn hex_encode(bytes: &[u8]) -> String {
+ use std::fmt::Write;
+ let mut out = String::with_capacity(bytes.len() * 2);
+ for byte in bytes {
+ // String writing is infallible
+ let _ = write!(out, "{:02x}", byte);
+ }
+ out
+}
+
+/// Canonicalizes headers into the AWS Canonical Form.
+///
+/// <http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
+fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
+ let mut headers = BTreeMap::<&str, Vec<&str>>::new();
+ let mut value_count = 0;
+ let mut value_bytes = 0;
+ let mut key_bytes = 0;
+
+ for (key, value) in header_map {
+ let key = key.as_str();
+ if ["authorization", "content-length", "user-agent"].contains(&key) {
+ continue;
+ }
+
+ let value = std::str::from_utf8(value.as_bytes()).unwrap();
+ key_bytes += key.len();
+ value_bytes += value.len();
+ value_count += 1;
+ headers.entry(key).or_default().push(value);
+ }
+
+ let mut signed_headers = String::with_capacity(key_bytes + headers.len());
+ let mut canonical_headers =
+ String::with_capacity(key_bytes + value_bytes + headers.len() + value_count);
+
+ for (header_idx, (name, values)) in headers.into_iter().enumerate() {
+ if header_idx != 0 {
+ signed_headers.push(';');
+ }
+
+ signed_headers.push_str(name);
+ canonical_headers.push_str(name);
+ canonical_headers.push(':');
+ for (value_idx, value) in values.into_iter().enumerate() {
+ if value_idx != 0 {
+ canonical_headers.push(',');
+ }
+ canonical_headers.push_str(value.trim());
+ }
+ canonical_headers.push('\n');
+ }
+
+ (signed_headers, canonical_headers)
+}
+
+#[derive(Debug)]
+pub enum CredentialProvider {
+ Static {
+ credential: Arc<AwsCredential>,
+ },
+ /// https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html
+ Instance {
+ cache: TokenCache<Arc<AwsCredential>>,
+ },
+ WebIdentity {
+ cache: TokenCache<Arc<AwsCredential>>,
+ token: String,
+ role_arn: String,
+ session_name: String,
+ endpoint: String,
+ },
+}
Review Comment:
I have not yet tested InstanceProvider or WebIdentityProvider, they are therefore probably slightly broken atm
##########
object_store/src/multipart.rs:
##########
@@ -26,23 +27,19 @@ type BoxedTryFuture<T> = Pin<Box<dyn Future<Output = Result<T, io::Error>> + Sen
/// A trait that can be implemented by cloud-based object stores
/// and used in combination with [`CloudMultiPartUpload`] to provide
/// multipart upload support
-///
-/// Note: this does not use AsyncTrait as the lifetimes are difficult to manage
-pub(crate) trait CloudMultiPartUploadImpl {
+#[async_trait]
Review Comment:
I took the opportunity to clean up this interface a bit, this isn't public and so this isn't a breaking change. I may split this out into another PR
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,493 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &'static str = "x-amz-date";
+const HASH_HEADER: &'static str = "x-amz-content-sha256";
+const TOKEN_HEADER: &'static str = "x-amz-security-token";
+const AUTH_HEADER: &'static str = "authorization";
+
+const ALL_HEADERS: &[&'static str; 4] =
+ &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
Review Comment:
Everything about this process is ludicrous... TLS exists... Why oh why would you do this :sob:
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,493 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &'static str = "x-amz-date";
+const HASH_HEADER: &'static str = "x-amz-content-sha256";
+const TOKEN_HEADER: &'static str = "x-amz-security-token";
+const AUTH_HEADER: &'static str = "authorization";
+
+const ALL_HEADERS: &[&'static str; 4] =
+ &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+ /// Sign a request <https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html>
+ fn sign(mut self, credential: &AwsCredential, region: &str, service: &str) -> Self {
+ // Hack around lack of access to underlying request
+ // https://github.com/seanmonstar/reqwest/issues/1212
+ let mut request = self
+ .try_clone()
+ .expect("not stream")
+ .build()
+ .expect("request valid");
+
+ let date = Utc::now();
+ let signer = RequestSigner {
+ date,
+ credential,
+ service,
+ region,
+ };
+
+ signer.sign(&mut request);
+
+ for header in ALL_HEADERS {
+ if let Some(val) = request.headers_mut().remove(*header) {
+ self = self.header(*header, val)
+ }
+ }
+ self
+ }
+}
+
+fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) -> ring::hmac::Tag {
+ let key = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, secret.as_ref());
+ ring::hmac::sign(&key, bytes.as_ref())
+}
+
+/// Computes the SHA256 digest of `body` returned as a hex encoded string
+fn hex_digest(bytes: &[u8]) -> String {
+ let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
+ hex_encode(digest.as_ref())
+}
+
+/// Returns `bytes` as a lower-case hex encoded string
+fn hex_encode(bytes: &[u8]) -> String {
+ use std::fmt::Write;
+ let mut out = String::with_capacity(bytes.len() * 2);
+ for byte in bytes {
+ // String writing is infallible
+ let _ = write!(out, "{:02x}", byte);
+ }
+ out
+}
+
+/// Canonicalizes headers into the AWS Canonical Form.
+///
+/// <http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
+fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
+ let mut headers = BTreeMap::<&str, Vec<&str>>::new();
+ let mut value_count = 0;
+ let mut value_bytes = 0;
+ let mut key_bytes = 0;
+
+ for (key, value) in header_map {
+ let key = key.as_str();
+ if ["authorization", "content-length", "user-agent"].contains(&key) {
+ continue;
+ }
+
+ let value = std::str::from_utf8(value.as_bytes()).unwrap();
+ key_bytes += key.len();
+ value_bytes += value.len();
+ value_count += 1;
+ headers.entry(key).or_default().push(value);
+ }
+
+ let mut signed_headers = String::with_capacity(key_bytes + headers.len());
+ let mut canonical_headers =
+ String::with_capacity(key_bytes + value_bytes + headers.len() + value_count);
+
+ for (header_idx, (name, values)) in headers.into_iter().enumerate() {
+ if header_idx != 0 {
+ signed_headers.push(';');
+ }
+
+ signed_headers.push_str(name);
+ canonical_headers.push_str(name);
+ canonical_headers.push(':');
+ for (value_idx, value) in values.into_iter().enumerate() {
+ if value_idx != 0 {
+ canonical_headers.push(',');
+ }
+ canonical_headers.push_str(value.trim());
+ }
+ canonical_headers.push('\n');
+ }
+
+ (signed_headers, canonical_headers)
+}
+
+#[derive(Debug)]
+pub enum CredentialProvider {
+ Static {
+ credential: Arc<AwsCredential>,
+ },
+ /// https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html
+ Instance {
+ cache: TokenCache<Arc<AwsCredential>>,
+ },
+ WebIdentity {
+ cache: TokenCache<Arc<AwsCredential>>,
+ token: String,
+ role_arn: String,
+ session_name: String,
+ endpoint: String,
+ },
+}
+
+impl CredentialProvider {
+ pub async fn get_credential(
+ &self,
+ client: &Client,
+ retry_config: &RetryConfig,
+ ) -> Result<Arc<AwsCredential>> {
+ match self {
+ CredentialProvider::Static { credential } => Ok(Arc::clone(&credential)),
+ CredentialProvider::Instance { cache } => {
+ cache
+ .get_or_insert_with(|| {
+ instance_creds(client, retry_config).map_err(|source| {
+ crate::Error::Generic {
+ store: "S3",
+ source,
+ }
+ })
+ })
+ .await
+ }
+ CredentialProvider::WebIdentity {
+ cache,
+ token,
+ role_arn,
+ session_name,
+ endpoint,
+ } => {
+ cache
+ .get_or_insert_with(|| {
+ web_identity(
+ client,
+ retry_config,
+ token,
+ role_arn,
+ session_name,
+ endpoint,
+ )
+ .map_err(|source| {
+ crate::Error::Generic {
+ store: "S3",
+ source,
+ }
+ })
+ })
+ .await
+ }
+ }
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct InstanceCredentials {
+ access_key_id: String,
+ secret_access_key: String,
+ token: String,
+ expiration: DateTime<Utc>,
+}
+
+impl From<InstanceCredentials> for AwsCredential {
+ fn from(s: InstanceCredentials) -> Self {
+ Self {
+ key_id: s.access_key_id,
+ secret_key: s.secret_access_key,
+ token: Some(s.token),
+ }
+ }
+}
+
+/// https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html#instance-metadata-security-credentials
+async fn instance_creds(
+ client: &Client,
+ retry_config: &RetryConfig,
+) -> Result<TemporaryToken<Arc<AwsCredential>>, StdError> {
+ const AWS_CREDENTIALS_PROVIDER_IP: &str = "169.254.169.254";
+ const AWS_CREDENTIALS_PROVIDER_PATH: &str =
+ "latest/meta-data/iam/security-credentials";
+ const AWS_EC2_METADATA_TOKEN_HEADER: &str = "X-aws-ec2-metadata-token";
+
+ let token_url = format!("http://{}/latest/api/token", AWS_CREDENTIALS_PROVIDER_IP);
Review Comment:
This is the fix for #2350
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1209876570
I would like to request holding off merging this PR until we have done the official 0.4 object store release from this repo, tracked by https://github.com/apache/arrow-rs/issues/2338
My rationale is that there are enough moving parts and we are (effectively) still testing some of this code in production so I don't want to accidentally introduce a regression that would require a new release quickly after the previous one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] ursabot commented on pull request #2352: Replace rusoto with custom implementation for AWS (#2176)
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1214911293
Benchmark runs are scheduled for baseline = 47a2c21efd09cbe08c213ef2d2c11201e43b98f4 and contender = 3f0e12d8d362752181c75836d25d424862acc424. 3f0e12d8d362752181c75836d25d424862acc424 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/1515564dca084742a2c8d507b990e004...da116332cb16435b81b9c257e153c559/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/2e7526c6931e421bba1683996888ef55...bac0ec89194646b59933b5d260aac5fd/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/c7cd9f6cc4fc40dcb4e4082be347f039...364ea01195a84a57b90b690e189e9aca/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/bb426b4ba5bd4bd1afe50225e9c752e6...74cb2097d4c14b0898ee9ce961b40839/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on pull request #2352: Replace rusoto with custom implementation for AWS (#2176)
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1214883194
I don't have any objections, though I haven't reviewed this PR in detail either. I will confirm we deployed these changes in our IOx environment and they seem to be working great
cc @roeap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] crepererum commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
crepererum commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r940003767
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,493 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &'static str = "x-amz-date";
+const HASH_HEADER: &'static str = "x-amz-content-sha256";
+const TOKEN_HEADER: &'static str = "x-amz-security-token";
+const AUTH_HEADER: &'static str = "authorization";
+
+const ALL_HEADERS: &[&'static str; 4] =
+ &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
Review Comment:
Big cooperation firewall/VPN stuff with TLS stripping :see_no_evil:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1207427498
Looks like I found a bug in the azure implementation :sweat_smile:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r941048585
##########
.github/workflows/object_store.yml:
##########
@@ -59,6 +59,13 @@ jobs:
image: localstack/localstack:0.14.4
ports:
- 4566:4566
+ ec2-metadata:
Review Comment:
Yup - this is the instance metadata mechanism for getting credentials
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] crepererum commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
crepererum commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r941013890
##########
object_store/src/client/pagination.rs:
##########
@@ -0,0 +1,55 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::Result;
+use futures::Stream;
+use std::future::Future;
+
+/// Performs a paginated operation
Review Comment:
This docstring says close to nothing :grin:
Maybe add a few words on how a stream is constructed out of a state and an async function and what this provides over `unfold` and `poll_fn` (which also exists for streams). Esp. that the output is paginated as well, it is NOT flattened.
Furthermore, I would suggest renaming `list` to `list_fn` or something, because initially I thought `list` is A list -- not a function to list -- that somehow gets unrolled here.
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &str = "x-amz-date";
+const HASH_HEADER: &str = "x-amz-content-sha256";
+const TOKEN_HEADER: &str = "x-amz-security-token";
+const AUTH_HEADER: &str = "authorization";
+
+const ALL_HEADERS: &[&str; 4] = &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
Review Comment:
I'm always a bit torn when it comes to extension traits for foreign types. Sure the make the code more elegant (I can clearly see that that in `client.rs`, which now feels like rather normal `reqwest` client stuff with very few parameters) but looking at code that uses is often makes me wonder "what's that new/fancy method on this type I've worked with before?!". So you can leave it as it is, just don't overdo it :wink:
Also: they are technically prone to break under semantic versioning, because upstream may be free to add new methods (not a breaking change) which then collide with this method name. Since your "use side" doesn't use a fully qualified name, the code will stop compiling due to ambiguity.
One thing to avoid both (my confusion and the potential breakage) would be to choose a more specific method name like `aws_sign` or `s3_sign`.
##########
object_store/src/aws/client.rs:
##########
@@ -0,0 +1,486 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::aws::credential::{AwsCredential, CredentialExt, CredentialProvider};
+use crate::client::pagination::paginated;
+use crate::client::retry::RetryExt;
+use crate::multipart::UploadPart;
+use crate::path::DELIMITER;
+use crate::util::{format_http_range, format_prefix};
+use crate::{
+ BoxStream, ListResult, MultipartId, ObjectMeta, Path, Result, RetryConfig, StreamExt,
+};
+use bytes::{Buf, Bytes};
+use chrono::{DateTime, Utc};
+use percent_encoding::{utf8_percent_encode, AsciiSet, PercentEncode, NON_ALPHANUMERIC};
+use reqwest::{Client, Method, Response, StatusCode};
Review Comment:
```suggestion
use reqwest::{Client as ReqwestClient, Method, Response, StatusCode};
```
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &str = "x-amz-date";
+const HASH_HEADER: &str = "x-amz-content-sha256";
+const TOKEN_HEADER: &str = "x-amz-security-token";
+const AUTH_HEADER: &str = "authorization";
+
+const ALL_HEADERS: &[&str; 4] = &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+ /// Sign a request <https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html>
Review Comment:
This doc should go the the trait, not the impl (or both if the the impl is special).
##########
object_store/src/aws/mod.rs:
##########
@@ -0,0 +1,631 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for S3
+//!
+//! ## Multi-part uploads
+//!
+//! Multi-part uploads can be initiated with the [ObjectStore::put_multipart] method.
+//! Data passed to the writer is automatically buffered to meet the minimum size
+//! requirements for a part. Multiple parts are uploaded concurrently.
+//!
+//! If the writer fails for any reason, you may have parts uploaded to AWS but not
+//! used that you may be charged for. Use the [ObjectStore::abort_multipart] method
+//! to abort the upload and drop those unneeded parts. In addition, you may wish to
+//! consider implementing [automatic cleanup] of unused parts that are older than one
+//! week.
+//!
+//! [automatic cleanup]: https://aws.amazon.com/blogs/aws/s3-lifecycle-management-update-support-for-multipart-uploads-and-delete-markers/
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{DateTime, Utc};
+use futures::stream::BoxStream;
+use futures::TryStreamExt;
+use snafu::{OptionExt, ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use tracing::info;
+
+use crate::aws::client::{S3Client, S3Config};
+use crate::aws::credential::{AwsCredential, CredentialProvider};
+use crate::multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart};
+use crate::{
+ GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Path, Result,
+ RetryConfig, StreamExt,
+};
+
+mod client;
+mod credential;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+ #[snafu(display("Last-Modified Header missing from response"))]
+ MissingLastModified,
+
+ #[snafu(display("Content-Length Header missing from response"))]
+ MissingContentLength,
+
+ #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+ InvalidLastModified {
+ last_modified: String,
+ source: chrono::ParseError,
+ },
+
+ #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+ InvalidContentLength {
+ content_length: String,
+ source: std::num::ParseIntError,
+ },
+
+ #[snafu(display("Missing region"))]
+ MissingRegion,
+
+ #[snafu(display("Missing bucket name"))]
+ MissingBucketName,
+
+ #[snafu(display("Missing AccessKeyId"))]
+ MissingAccessKeyId,
+
+ #[snafu(display("Missing SecretAccessKey"))]
+ MissingSecretAccessKey,
+
+ #[snafu(display("ETag Header missing from response"))]
+ MissingEtag,
+
+ #[snafu(display("Received header containing non-ASCII data"))]
+ BadHeader { source: reqwest::header::ToStrError },
+
+ #[snafu(display("Error reading token file: {}", source))]
+ ReadTokenFile { source: std::io::Error },
+}
+
+impl From<Error> for super::Error {
+ fn from(err: Error) -> Self {
+ Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ }
+ }
+}
+
+/// Interface for [Amazon S3](https://aws.amazon.com/s3/).
+#[derive(Debug)]
+pub struct AmazonS3 {
+ client: Arc<S3Client>,
Review Comment:
What's the reason that the `S3Client` is not just inlined into the store object? Feels a bit like a layer that isn't really necessary.
##########
object_store/src/aws/client.rs:
##########
@@ -0,0 +1,486 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::aws::credential::{AwsCredential, CredentialExt, CredentialProvider};
+use crate::client::pagination::paginated;
+use crate::client::retry::RetryExt;
+use crate::multipart::UploadPart;
+use crate::path::DELIMITER;
+use crate::util::{format_http_range, format_prefix};
+use crate::{
+ BoxStream, ListResult, MultipartId, ObjectMeta, Path, Result, RetryConfig, StreamExt,
+};
+use bytes::{Buf, Bytes};
+use chrono::{DateTime, Utc};
+use percent_encoding::{utf8_percent_encode, AsciiSet, PercentEncode, NON_ALPHANUMERIC};
+use reqwest::{Client, Method, Response, StatusCode};
+use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
+use std::ops::Range;
+use std::sync::Arc;
+
+// http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+//
+// Do not URI-encode any of the unreserved characters that RFC 3986 defines:
+// A-Z, a-z, 0-9, hyphen ( - ), underscore ( _ ), period ( . ), and tilde ( ~ ).
+const STRICT_ENCODE_SET: AsciiSet = NON_ALPHANUMERIC
+ .remove(b'-')
+ .remove(b'.')
+ .remove(b'_')
+ .remove(b'~');
+
+/// This struct is used to maintain the URI path encoding
+const STRICT_PATH_ENCODE_SET: AsciiSet = STRICT_ENCODE_SET.remove(b'/');
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+pub(crate) enum Error {
+ #[snafu(display("Error performing get request {}: {}", path, source))]
+ GetRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing put request {}: {}", path, source))]
+ PutRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing delete request {}: {}", path, source))]
+ DeleteRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing copy request {}: {}", path, source))]
+ CopyRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing list request: {}", source))]
+ ListRequest { source: reqwest::Error },
+
+ #[snafu(display("Error performing create multipart request: {}", source))]
+ CreateMultipartRequest { source: reqwest::Error },
+
+ #[snafu(display("Error performing complete multipart request: {}", source))]
+ CompleteMultipartRequest { source: reqwest::Error },
+
+ #[snafu(display("Got invalid list response: {}", source))]
+ InvalidListResponse { source: quick_xml::de::DeError },
+
+ #[snafu(display("Got invalid multipart response: {}", source))]
+ InvalidMultipartResponse { source: quick_xml::de::DeError },
+}
+
+impl From<Error> for crate::Error {
+ fn from(err: Error) -> Self {
+ match err {
+ Error::GetRequest { source, path }
+ | Error::DeleteRequest { source, path }
+ | Error::CopyRequest { source, path }
+ | Error::PutRequest { source, path }
+ if matches!(source.status(), Some(StatusCode::NOT_FOUND)) =>
+ {
+ Self::NotFound {
+ path,
+ source: Box::new(source),
+ }
+ }
+ _ => Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ },
+ }
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListResponse {
+ #[serde(default)]
+ pub contents: Vec<ListContents>,
+ #[serde(default)]
+ pub common_prefixes: Vec<ListPrefix>,
+ #[serde(default)]
+ pub next_continuation_token: Option<String>,
+}
+
+impl TryFrom<ListResponse> for ListResult {
+ type Error = crate::Error;
+
+ fn try_from(value: ListResponse) -> Result<Self> {
+ let common_prefixes = value
+ .common_prefixes
+ .into_iter()
+ .map(|x| Ok(Path::parse(&x.prefix)?))
+ .collect::<Result<_>>()?;
+
+ let objects = value
+ .contents
+ .into_iter()
+ .map(TryFrom::try_from)
+ .collect::<Result<_>>()?;
+
+ Ok(Self {
+ common_prefixes,
+ objects,
+ })
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListPrefix {
+ pub prefix: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListContents {
+ pub key: String,
+ pub size: usize,
+ pub last_modified: DateTime<Utc>,
+}
+
+impl TryFrom<ListContents> for ObjectMeta {
+ type Error = crate::Error;
+
+ fn try_from(value: ListContents) -> Result<Self> {
+ Ok(Self {
+ location: Path::parse(value.key)?,
+ last_modified: value.last_modified,
+ size: value.size,
+ })
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct InitiateMultipart {
+ upload_id: String,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase", rename = "CompleteMultipartUpload")]
+struct CompleteMultipart {
+ part: Vec<MultipartPart>,
+}
+
+#[derive(Debug, Serialize)]
+struct MultipartPart {
+ #[serde(rename = "$unflatten=ETag")]
+ e_tag: String,
+ #[serde(rename = "$unflatten=PartNumber")]
+ part_number: usize,
+}
+
+#[derive(Debug)]
+pub struct S3Config {
+ pub region: String,
+ pub endpoint: String,
+ pub bucket: String,
+ pub credentials: CredentialProvider,
+ pub retry_config: RetryConfig,
+ pub allow_http: bool,
+}
+
+impl S3Config {
+ fn path_url(&self, path: &Path) -> String {
+ format!("{}/{}/{}", self.endpoint, self.bucket, encode_path(path))
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct S3Client {
+ config: S3Config,
+ client: Client,
Review Comment:
```suggestion
reqwest_client: ReqwestClient,
```
I was wondering what the client in a client in a client was.
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
This module makes me a bit unhappy. It's such a PITA that AWS requires all that stuff. However I see you clearly did your homework and it's documented and the tests work, so I just assume it does the right thing.
If you want me to double-check the code with the official docs, I can do that but I think it's not really required.
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &str = "x-amz-date";
+const HASH_HEADER: &str = "x-amz-content-sha256";
+const TOKEN_HEADER: &str = "x-amz-security-token";
+const AUTH_HEADER: &str = "authorization";
+
+const ALL_HEADERS: &[&str; 4] = &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+ /// Sign a request <https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html>
+ fn sign(mut self, credential: &AwsCredential, region: &str, service: &str) -> Self {
+ // Hack around lack of access to underlying request
+ // https://github.com/seanmonstar/reqwest/issues/1212
+ let mut request = self
+ .try_clone()
+ .expect("not stream")
+ .build()
+ .expect("request valid");
+
+ let date = Utc::now();
+ let signer = RequestSigner {
+ date,
+ credential,
+ service,
+ region,
+ };
+
+ signer.sign(&mut request);
+
+ for header in ALL_HEADERS {
+ if let Some(val) = request.headers_mut().remove(*header) {
+ self = self.header(*header, val)
+ }
+ }
+ self
+ }
+}
+
+fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) -> ring::hmac::Tag {
+ let key = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, secret.as_ref());
+ ring::hmac::sign(&key, bytes.as_ref())
+}
+
+/// Computes the SHA256 digest of `body` returned as a hex encoded string
+fn hex_digest(bytes: &[u8]) -> String {
+ let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
+ hex_encode(digest.as_ref())
+}
+
+/// Returns `bytes` as a lower-case hex encoded string
+fn hex_encode(bytes: &[u8]) -> String {
+ use std::fmt::Write;
+ let mut out = String::with_capacity(bytes.len() * 2);
+ for byte in bytes {
+ // String writing is infallible
+ let _ = write!(out, "{:02x}", byte);
+ }
+ out
+}
+
+/// Canonicalizes headers into the AWS Canonical Form.
+///
+/// <http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
Review Comment:
```suggestion
/// <https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
```
##########
object_store/src/aws/mod.rs:
##########
@@ -0,0 +1,631 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for S3
+//!
+//! ## Multi-part uploads
+//!
+//! Multi-part uploads can be initiated with the [ObjectStore::put_multipart] method.
+//! Data passed to the writer is automatically buffered to meet the minimum size
+//! requirements for a part. Multiple parts are uploaded concurrently.
+//!
+//! If the writer fails for any reason, you may have parts uploaded to AWS but not
+//! used that you may be charged for. Use the [ObjectStore::abort_multipart] method
+//! to abort the upload and drop those unneeded parts. In addition, you may wish to
+//! consider implementing [automatic cleanup] of unused parts that are older than one
+//! week.
+//!
+//! [automatic cleanup]: https://aws.amazon.com/blogs/aws/s3-lifecycle-management-update-support-for-multipart-uploads-and-delete-markers/
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{DateTime, Utc};
+use futures::stream::BoxStream;
+use futures::TryStreamExt;
+use snafu::{OptionExt, ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use tracing::info;
+
+use crate::aws::client::{S3Client, S3Config};
+use crate::aws::credential::{AwsCredential, CredentialProvider};
+use crate::multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart};
+use crate::{
+ GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Path, Result,
+ RetryConfig, StreamExt,
+};
+
+mod client;
+mod credential;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+ #[snafu(display("Last-Modified Header missing from response"))]
+ MissingLastModified,
+
+ #[snafu(display("Content-Length Header missing from response"))]
+ MissingContentLength,
+
+ #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+ InvalidLastModified {
+ last_modified: String,
+ source: chrono::ParseError,
+ },
+
+ #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+ InvalidContentLength {
+ content_length: String,
+ source: std::num::ParseIntError,
+ },
+
+ #[snafu(display("Missing region"))]
+ MissingRegion,
+
+ #[snafu(display("Missing bucket name"))]
+ MissingBucketName,
+
+ #[snafu(display("Missing AccessKeyId"))]
+ MissingAccessKeyId,
+
+ #[snafu(display("Missing SecretAccessKey"))]
+ MissingSecretAccessKey,
+
+ #[snafu(display("ETag Header missing from response"))]
+ MissingEtag,
+
+ #[snafu(display("Received header containing non-ASCII data"))]
+ BadHeader { source: reqwest::header::ToStrError },
+
+ #[snafu(display("Error reading token file: {}", source))]
+ ReadTokenFile { source: std::io::Error },
+}
+
+impl From<Error> for super::Error {
+ fn from(err: Error) -> Self {
+ Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ }
Review Comment:
Shouldn't there be at least a `NotFound` error (and the generic test suite should probably assert that fetching an unknown object actually results in an "not found" error, not some generic one).
##########
.github/workflows/object_store.yml:
##########
@@ -59,6 +59,13 @@ jobs:
image: localstack/localstack:0.14.4
ports:
- 4566:4566
+ ec2-metadata:
Review Comment:
Why do we need EC2 now? Is this to test that EC2 VMs could fetch S3 credentials using yet-another-mechanism[^1]?
[^1]: That's AWS blaming, it's not against this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold merged pull request #2352: Replace rusoto with custom implementation for AWS (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold merged PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2352: Replace rusoto with custom implementation for AWS (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1214859793
I intend to get this in this afternoon unless anybody would like longer to review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r943385887
##########
object_store/src/aws/client.rs:
##########
@@ -0,0 +1,486 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::aws::credential::{AwsCredential, CredentialExt, CredentialProvider};
+use crate::client::pagination::paginated;
+use crate::client::retry::RetryExt;
+use crate::multipart::UploadPart;
+use crate::path::DELIMITER;
+use crate::util::{format_http_range, format_prefix};
+use crate::{
+ BoxStream, ListResult, MultipartId, ObjectMeta, Path, Result, RetryConfig, StreamExt,
+};
+use bytes::{Buf, Bytes};
+use chrono::{DateTime, Utc};
+use percent_encoding::{utf8_percent_encode, AsciiSet, PercentEncode, NON_ALPHANUMERIC};
+use reqwest::{Client, Method, Response, StatusCode};
+use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
+use std::ops::Range;
+use std::sync::Arc;
+
+// http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+//
+// Do not URI-encode any of the unreserved characters that RFC 3986 defines:
+// A-Z, a-z, 0-9, hyphen ( - ), underscore ( _ ), period ( . ), and tilde ( ~ ).
+const STRICT_ENCODE_SET: AsciiSet = NON_ALPHANUMERIC
+ .remove(b'-')
+ .remove(b'.')
+ .remove(b'_')
+ .remove(b'~');
+
+/// This struct is used to maintain the URI path encoding
+const STRICT_PATH_ENCODE_SET: AsciiSet = STRICT_ENCODE_SET.remove(b'/');
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+pub(crate) enum Error {
+ #[snafu(display("Error performing get request {}: {}", path, source))]
+ GetRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing put request {}: {}", path, source))]
+ PutRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing delete request {}: {}", path, source))]
+ DeleteRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing copy request {}: {}", path, source))]
+ CopyRequest {
+ source: reqwest::Error,
+ path: String,
+ },
+
+ #[snafu(display("Error performing list request: {}", source))]
+ ListRequest { source: reqwest::Error },
+
+ #[snafu(display("Error performing create multipart request: {}", source))]
+ CreateMultipartRequest { source: reqwest::Error },
+
+ #[snafu(display("Error performing complete multipart request: {}", source))]
+ CompleteMultipartRequest { source: reqwest::Error },
+
+ #[snafu(display("Got invalid list response: {}", source))]
+ InvalidListResponse { source: quick_xml::de::DeError },
+
+ #[snafu(display("Got invalid multipart response: {}", source))]
+ InvalidMultipartResponse { source: quick_xml::de::DeError },
+}
+
+impl From<Error> for crate::Error {
+ fn from(err: Error) -> Self {
+ match err {
+ Error::GetRequest { source, path }
+ | Error::DeleteRequest { source, path }
+ | Error::CopyRequest { source, path }
+ | Error::PutRequest { source, path }
+ if matches!(source.status(), Some(StatusCode::NOT_FOUND)) =>
+ {
+ Self::NotFound {
+ path,
+ source: Box::new(source),
+ }
+ }
+ _ => Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ },
+ }
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListResponse {
+ #[serde(default)]
+ pub contents: Vec<ListContents>,
+ #[serde(default)]
+ pub common_prefixes: Vec<ListPrefix>,
+ #[serde(default)]
+ pub next_continuation_token: Option<String>,
+}
+
+impl TryFrom<ListResponse> for ListResult {
+ type Error = crate::Error;
+
+ fn try_from(value: ListResponse) -> Result<Self> {
+ let common_prefixes = value
+ .common_prefixes
+ .into_iter()
+ .map(|x| Ok(Path::parse(&x.prefix)?))
+ .collect::<Result<_>>()?;
+
+ let objects = value
+ .contents
+ .into_iter()
+ .map(TryFrom::try_from)
+ .collect::<Result<_>>()?;
+
+ Ok(Self {
+ common_prefixes,
+ objects,
+ })
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListPrefix {
+ pub prefix: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListContents {
+ pub key: String,
+ pub size: usize,
+ pub last_modified: DateTime<Utc>,
+}
+
+impl TryFrom<ListContents> for ObjectMeta {
+ type Error = crate::Error;
+
+ fn try_from(value: ListContents) -> Result<Self> {
+ Ok(Self {
+ location: Path::parse(value.key)?,
+ last_modified: value.last_modified,
+ size: value.size,
+ })
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct InitiateMultipart {
+ upload_id: String,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase", rename = "CompleteMultipartUpload")]
+struct CompleteMultipart {
+ part: Vec<MultipartPart>,
+}
+
+#[derive(Debug, Serialize)]
+struct MultipartPart {
+ #[serde(rename = "$unflatten=ETag")]
+ e_tag: String,
+ #[serde(rename = "$unflatten=PartNumber")]
+ part_number: usize,
+}
+
+#[derive(Debug)]
+pub struct S3Config {
+ pub region: String,
+ pub endpoint: String,
+ pub bucket: String,
+ pub credentials: CredentialProvider,
+ pub retry_config: RetryConfig,
+ pub allow_http: bool,
+}
+
+impl S3Config {
+ fn path_url(&self, path: &Path) -> String {
+ format!("{}/{}/{}", self.endpoint, self.bucket, encode_path(path))
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct S3Client {
+ config: S3Config,
+ client: Client,
Review Comment:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
alamb commented on PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#issuecomment-1211876997
Sounds like the IOx 'testing in production' has gone very well ❤️
So to move this PR forward:
1. It seems like like there are a few more comments from @crepererum to address.
2. The 0.4.0 release is going along nicely,
3. I think we are on track to merge this over the weekend or early next week
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r939673201
##########
object_store/src/multipart.rs:
##########
@@ -185,10 +186,15 @@ where
// If shutdown task is not set, set it
let parts = std::mem::take(&mut self.completed_parts);
+ let parts = parts.into_iter().map(Option::unwrap).collect();
Review Comment:
I need to double-check if this needs to propagate an error, I'm not sure under what circumstances a part might be missing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r940312171
##########
object_store/src/aws/credential.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::retry::RetryExt;
+use crate::client::token::{TemporaryToken, TokenCache};
+use crate::{Result, RetryConfig};
+use bytes::Buf;
+use chrono::{DateTime, Utc};
+use futures::TryFutureExt;
+use reqwest::header::{HeaderMap, HeaderValue};
+use reqwest::{Client, Method, Request, RequestBuilder};
+use serde::Deserialize;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::time::Instant;
+
+type StdError = Box<dyn std::error::Error + Send + Sync>;
+
+/// SHA256 hash of empty string
+static EMPTY_SHA256_HASH: &str =
+ "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+#[derive(Debug)]
+pub struct AwsCredential {
+ pub key_id: String,
+ pub secret_key: String,
+ pub token: Option<String>,
+}
+
+impl AwsCredential {
+ /// Signs a string
+ ///
+ /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html>
+ fn sign(
+ &self,
+ to_sign: &str,
+ date: DateTime<Utc>,
+ region: &str,
+ service: &str,
+ ) -> String {
+ let date_string = date.format("%Y%m%d").to_string();
+ let date_hmac = hmac_sha256(format!("AWS4{}", self.secret_key), date_string);
+ let region_hmac = hmac_sha256(date_hmac, region);
+ let service_hmac = hmac_sha256(region_hmac, service);
+ let signing_hmac = hmac_sha256(service_hmac, b"aws4_request");
+ hex_encode(hmac_sha256(signing_hmac, to_sign).as_ref())
+ }
+}
+
+struct RequestSigner<'a> {
+ date: DateTime<Utc>,
+ credential: &'a AwsCredential,
+ service: &'a str,
+ region: &'a str,
+}
+
+const DATE_HEADER: &str = "x-amz-date";
+const HASH_HEADER: &str = "x-amz-content-sha256";
+const TOKEN_HEADER: &str = "x-amz-security-token";
+const AUTH_HEADER: &str = "authorization";
+
+const ALL_HEADERS: &[&str; 4] = &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
+
+impl<'a> RequestSigner<'a> {
+ fn sign(&self, request: &mut Request) {
+ if let Some(ref token) = self.credential.token {
+ let token_val = HeaderValue::from_str(token).unwrap();
+ request.headers_mut().insert(TOKEN_HEADER, token_val);
+ }
+
+ let host_val = HeaderValue::from_str(request.url().host_str().unwrap()).unwrap();
+ request.headers_mut().insert("host", host_val);
+
+ let date_str = self.date.format("%Y%m%dT%H%M%SZ").to_string();
+ let date_val = HeaderValue::from_str(&date_str).unwrap();
+ request.headers_mut().insert(DATE_HEADER, date_val);
+
+ let digest = match request.body() {
+ None => EMPTY_SHA256_HASH.to_string(),
+ Some(body) => hex_digest(body.as_bytes().unwrap()),
+ };
+
+ let header_digest = HeaderValue::from_str(&digest).unwrap();
+ request.headers_mut().insert(HASH_HEADER, header_digest);
+
+ let (signed_headers, canonical_headers) = canonicalize_headers(request.headers());
+
+ // https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+ let canonical_request = format!(
+ "{}\n{}\n{}\n{}\n{}\n{}",
+ request.method().as_str(),
+ request.url().path(), // S3 doesn't percent encode this like other services
+ request.url().query().unwrap_or(""), // This assumes the query pairs are in order
+ canonical_headers,
+ signed_headers,
+ digest
+ );
+
+ let hashed_canonical_request = hex_digest(canonical_request.as_bytes());
+ let scope = format!(
+ "{}/{}/{}/aws4_request",
+ self.date.format("%Y%m%d"),
+ self.region,
+ self.service
+ );
+
+ let string_to_sign = format!(
+ "AWS4-HMAC-SHA256\n{}\n{}\n{}",
+ self.date.format("%Y%m%dT%H%M%SZ"),
+ scope,
+ hashed_canonical_request
+ );
+
+ // sign the string
+ let signature =
+ self.credential
+ .sign(&string_to_sign, self.date, self.region, self.service);
+
+ // build the actual auth header
+ let authorisation = format!(
+ "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
+ self.credential.key_id, scope, signed_headers, signature
+ );
+
+ let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ request.headers_mut().insert(AUTH_HEADER, authorization_val);
+ }
+}
+
+pub trait CredentialExt {
+ fn sign(self, credential: &AwsCredential, region: &str, service: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+ /// Sign a request <https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html>
+ fn sign(mut self, credential: &AwsCredential, region: &str, service: &str) -> Self {
+ // Hack around lack of access to underlying request
+ // https://github.com/seanmonstar/reqwest/issues/1212
+ let mut request = self
+ .try_clone()
+ .expect("not stream")
+ .build()
+ .expect("request valid");
+
+ let date = Utc::now();
+ let signer = RequestSigner {
+ date,
+ credential,
+ service,
+ region,
+ };
+
+ signer.sign(&mut request);
+
+ for header in ALL_HEADERS {
+ if let Some(val) = request.headers_mut().remove(*header) {
+ self = self.header(*header, val)
+ }
+ }
+ self
+ }
+}
+
+fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) -> ring::hmac::Tag {
+ let key = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, secret.as_ref());
+ ring::hmac::sign(&key, bytes.as_ref())
+}
+
+/// Computes the SHA256 digest of `body` returned as a hex encoded string
+fn hex_digest(bytes: &[u8]) -> String {
+ let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
+ hex_encode(digest.as_ref())
+}
+
+/// Returns `bytes` as a lower-case hex encoded string
+fn hex_encode(bytes: &[u8]) -> String {
+ use std::fmt::Write;
+ let mut out = String::with_capacity(bytes.len() * 2);
+ for byte in bytes {
+ // String writing is infallible
+ let _ = write!(out, "{:02x}", byte);
+ }
+ out
+}
+
+/// Canonicalizes headers into the AWS Canonical Form.
+///
+/// <http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
+fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
+ let mut headers = BTreeMap::<&str, Vec<&str>>::new();
+ let mut value_count = 0;
+ let mut value_bytes = 0;
+ let mut key_bytes = 0;
+
+ for (key, value) in header_map {
+ let key = key.as_str();
+ if ["authorization", "content-length", "user-agent"].contains(&key) {
+ continue;
+ }
+
+ let value = std::str::from_utf8(value.as_bytes()).unwrap();
+ key_bytes += key.len();
+ value_bytes += value.len();
+ value_count += 1;
+ headers.entry(key).or_default().push(value);
+ }
+
+ let mut signed_headers = String::with_capacity(key_bytes + headers.len());
+ let mut canonical_headers =
+ String::with_capacity(key_bytes + value_bytes + headers.len() + value_count);
+
+ for (header_idx, (name, values)) in headers.into_iter().enumerate() {
+ if header_idx != 0 {
+ signed_headers.push(';');
+ }
+
+ signed_headers.push_str(name);
+ canonical_headers.push_str(name);
+ canonical_headers.push(':');
+ for (value_idx, value) in values.into_iter().enumerate() {
+ if value_idx != 0 {
+ canonical_headers.push(',');
+ }
+ canonical_headers.push_str(value.trim());
+ }
+ canonical_headers.push('\n');
+ }
+
+ (signed_headers, canonical_headers)
+}
+
+#[derive(Debug)]
+pub enum CredentialProvider {
+ Static {
+ credential: Arc<AwsCredential>,
+ },
+ /// <https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html>
+ Instance {
+ cache: TokenCache<Arc<AwsCredential>>,
+ },
+ WebIdentity {
+ cache: TokenCache<Arc<AwsCredential>>,
+ token: String,
+ role_arn: String,
+ session_name: String,
+ endpoint: String,
+ },
+}
+
+impl CredentialProvider {
+ pub async fn get_credential(
+ &self,
+ client: &Client,
+ retry_config: &RetryConfig,
+ ) -> Result<Arc<AwsCredential>> {
+ match self {
+ CredentialProvider::Static { credential } => Ok(Arc::clone(credential)),
+ CredentialProvider::Instance { cache } => {
+ cache
+ .get_or_insert_with(|| {
+ const METADATA_ENDPOINT: &str = "http://169.254.169.254";
+ instance_creds(client, retry_config, METADATA_ENDPOINT).map_err(
+ |source| crate::Error::Generic {
+ store: "S3",
+ source,
+ },
+ )
+ })
+ .await
+ }
+ CredentialProvider::WebIdentity {
+ cache,
+ token,
+ role_arn,
+ session_name,
+ endpoint,
+ } => {
+ cache
+ .get_or_insert_with(|| {
+ web_identity(
+ client,
+ retry_config,
+ token,
+ role_arn,
+ session_name,
+ endpoint,
+ )
+ .map_err(|source| {
+ crate::Error::Generic {
+ store: "S3",
+ source,
+ }
+ })
+ })
+ .await
+ }
+ }
+ }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct InstanceCredentials {
+ access_key_id: String,
+ secret_access_key: String,
+ token: String,
+ expiration: DateTime<Utc>,
+}
+
+impl From<InstanceCredentials> for AwsCredential {
+ fn from(s: InstanceCredentials) -> Self {
+ Self {
+ key_id: s.access_key_id,
+ secret_key: s.secret_access_key,
+ token: Some(s.token),
+ }
+ }
+}
+
+/// <https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html#instance-metadata-security-credentials>
+async fn instance_creds(
+ client: &Client,
+ retry_config: &RetryConfig,
+ endpoint: &str,
+) -> Result<TemporaryToken<Arc<AwsCredential>>, StdError> {
+ const CREDENTIALS_PATH: &str = "latest/meta-data/iam/security-credentials";
+ const AWS_EC2_METADATA_TOKEN_HEADER: &str = "X-aws-ec2-metadata-token";
+
+ let token_url = format!("{}/latest/api/token", endpoint);
+ let token = client
+ .request(Method::PUT, token_url)
+ .header("X-aws-ec2-metadata-token-ttl-seconds", "600") // 10 minute TTL
+ .send_retry(retry_config)
+ .await?
+ .text()
+ .await?;
+
+ let role_url = format!("{}/{}/", endpoint, CREDENTIALS_PATH);
+ let role = client
+ .request(Method::GET, role_url)
+ .header(AWS_EC2_METADATA_TOKEN_HEADER, &token)
+ .send_retry(retry_config)
+ .await?
+ .text()
+ .await?;
+
+ let creds_url = format!("{}/{}/{}", endpoint, CREDENTIALS_PATH, role);
+ let creds: InstanceCredentials = client
+ .request(Method::GET, creds_url)
+ .header(AWS_EC2_METADATA_TOKEN_HEADER, &token)
+ .send_retry(retry_config)
+ .await?
+ .json()
+ .await?;
+
+ let now = Utc::now();
+ let ttl = (creds.expiration - now).to_std().unwrap_or_default();
+ Ok(TemporaryToken {
+ token: Arc::new(creds.into()),
+ expiry: Instant::now() + ttl,
+ })
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct AssumeRoleResponse {
+ assume_role_with_web_identity_result: AssumeRoleResult,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct AssumeRoleResult {
+ credentials: AssumeRoleCredentials,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct AssumeRoleCredentials {
+ session_token: String,
+ secret_access_key: String,
+ access_key_id: String,
+ expiration: DateTime<Utc>,
+}
+
+impl From<AssumeRoleCredentials> for AwsCredential {
+ fn from(s: AssumeRoleCredentials) -> Self {
+ Self {
+ key_id: s.access_key_id,
+ secret_key: s.secret_access_key,
+ token: Some(s.session_token),
+ }
+ }
+}
+
+/// <https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts-technical-overview.html>
+async fn web_identity(
Review Comment:
I don't have a good way to integration test this, but I have tested it with some real EKS tokens and it seems to work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2352: Replace rusoto with custom implementation (#2176)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2352:
URL: https://github.com/apache/arrow-rs/pull/2352#discussion_r941049514
##########
object_store/src/aws/mod.rs:
##########
@@ -0,0 +1,631 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for S3
+//!
+//! ## Multi-part uploads
+//!
+//! Multi-part uploads can be initiated with the [ObjectStore::put_multipart] method.
+//! Data passed to the writer is automatically buffered to meet the minimum size
+//! requirements for a part. Multiple parts are uploaded concurrently.
+//!
+//! If the writer fails for any reason, you may have parts uploaded to AWS but not
+//! used that you may be charged for. Use the [ObjectStore::abort_multipart] method
+//! to abort the upload and drop those unneeded parts. In addition, you may wish to
+//! consider implementing [automatic cleanup] of unused parts that are older than one
+//! week.
+//!
+//! [automatic cleanup]: https://aws.amazon.com/blogs/aws/s3-lifecycle-management-update-support-for-multipart-uploads-and-delete-markers/
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{DateTime, Utc};
+use futures::stream::BoxStream;
+use futures::TryStreamExt;
+use snafu::{OptionExt, ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use tracing::info;
+
+use crate::aws::client::{S3Client, S3Config};
+use crate::aws::credential::{AwsCredential, CredentialProvider};
+use crate::multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart};
+use crate::{
+ GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Path, Result,
+ RetryConfig, StreamExt,
+};
+
+mod client;
+mod credential;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+ #[snafu(display("Last-Modified Header missing from response"))]
+ MissingLastModified,
+
+ #[snafu(display("Content-Length Header missing from response"))]
+ MissingContentLength,
+
+ #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+ InvalidLastModified {
+ last_modified: String,
+ source: chrono::ParseError,
+ },
+
+ #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+ InvalidContentLength {
+ content_length: String,
+ source: std::num::ParseIntError,
+ },
+
+ #[snafu(display("Missing region"))]
+ MissingRegion,
+
+ #[snafu(display("Missing bucket name"))]
+ MissingBucketName,
+
+ #[snafu(display("Missing AccessKeyId"))]
+ MissingAccessKeyId,
+
+ #[snafu(display("Missing SecretAccessKey"))]
+ MissingSecretAccessKey,
+
+ #[snafu(display("ETag Header missing from response"))]
+ MissingEtag,
+
+ #[snafu(display("Received header containing non-ASCII data"))]
+ BadHeader { source: reqwest::header::ToStrError },
+
+ #[snafu(display("Error reading token file: {}", source))]
+ ReadTokenFile { source: std::io::Error },
+}
+
+impl From<Error> for super::Error {
+ fn from(err: Error) -> Self {
+ Self::Generic {
+ store: "S3",
+ source: Box::new(err),
+ }
Review Comment:
The NotFound variant is produced by the client
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org