This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs-object-store.git
The following commit(s) were added to refs/heads/main by this push:
new c216ede fix: return errors instead of panicking on invalid requests
(#894)
c216ede is described below
commit c216edef0ac3d0d39da5bcdfeed3dcd034e26597
Author: Xin Sun <[email protected]>
AuthorDate: Sun Nov 1 20:56:52 2026 +0800
fix: return errors instead of panicking on invalid requests (#894)
Change-Id: I76f22524f0691db92df4886f43920b763396468f
---
src/aws/credential.rs | 31 +++--
src/azure/builder.rs | 11 ++
src/azure/credential.rs | 42 +++---
src/client/builder.rs | 86 ++++++++----
src/client/http/connection.rs | 16 ++-
src/client/retry.rs | 86 ++++++++++--
src/http/mod.rs | 7 +
tests/request_errors.rs | 316 ++++++++++++++++++++++++++++++++++++++++++
8 files changed, 520 insertions(+), 75 deletions(-)
diff --git a/src/aws/credential.rs b/src/aws/credential.rs
index 72017fd..8084496 100644
--- a/src/aws/credential.rs
+++ b/src/aws/credential.rs
@@ -20,8 +20,8 @@ use crate::client::builder::HttpRequestBuilder;
use crate::client::retry::RetryExt;
use crate::client::token::{TemporaryToken, TokenCache};
use crate::client::{
- CryptoProvider, DigestAlgorithm, HttpClient, HttpError, HttpRequest,
TokenProvider,
- crypto_provider,
+ CryptoProvider, DigestAlgorithm, HttpClient, HttpError, HttpErrorKind,
HttpRequest,
+ TokenProvider, crypto_provider,
};
use crate::util::{append_strict_query_pairs, hex_digest, hex_encode};
use crate::{CredentialProvider, Result, RetryConfig};
@@ -218,11 +218,11 @@ impl<'a> AwsAuthorizer<'a> {
request: &mut HttpRequest,
pre_calculated_digest: Option<&[u8]>,
) -> Result<()> {
+ let url =
Url::parse(&request.uri().to_string()).map_err(invalid_request)?;
let crypto = crypto_provider(self.crypto)?;
- let url = Url::parse(&request.uri().to_string()).unwrap();
if let Some(ref token) = self.credential.token {
- let token_val = HeaderValue::from_str(token).unwrap();
+ let token_val =
HeaderValue::from_str(token).map_err(invalid_request)?;
let header = self.token_header.as_ref().unwrap_or(&TOKEN_HEADER);
request.headers_mut().insert(header, token_val);
}
@@ -265,7 +265,7 @@ impl<'a> AwsAuthorizer<'a> {
request
.headers_mut()
.insert(&HOST, HeaderValue::from_str(host).unwrap());
- let (signed_headers, canonical_headers) =
canonicalize_headers(request.headers());
+ let (signed_headers, canonical_headers) =
canonicalize_headers(request.headers())?;
request.headers_mut().remove(&HOST);
let scope = self.scope(date);
@@ -292,7 +292,7 @@ impl<'a> AwsAuthorizer<'a> {
ALGORITHM, self.credential.key_id, scope, signed_headers, signature
);
- let authorization_val = HeaderValue::from_str(&authorisation).unwrap();
+ let authorization_val =
HeaderValue::from_str(&authorisation).map_err(invalid_request)?;
request
.headers_mut()
.insert(&AUTHORIZATION, authorization_val);
@@ -336,7 +336,7 @@ impl<'a> AwsAuthorizer<'a> {
headers.append(name.clone(), value.clone());
}
- let (signed_headers, canonical_headers) =
canonicalize_headers(&headers);
+ let (signed_headers, canonical_headers) =
canonicalize_headers(&headers)?;
//
https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html
url.query_pairs_mut()
@@ -458,7 +458,7 @@ impl CredentialExt for HttpRequestBuilder {
match authorizer {
Some(authorizer) => {
let (client, request) = self.into_parts();
- let mut request = request.expect("request valid");
+ let mut request = request.map_err(invalid_request)?;
authorizer.try_authorize(&mut request, payload_sha256)?;
Ok(Self::from_parts(client, request))
@@ -468,6 +468,13 @@ impl CredentialExt for HttpRequestBuilder {
}
}
+fn invalid_request(error: impl std::error::Error + Send + Sync + 'static) ->
crate::Error {
+ crate::Error::Generic {
+ store: STORE,
+ source: Box::new(HttpError::new(HttpErrorKind::InvalidRequest, error)),
+ }
+}
+
/// Canonicalizes query parameters into the AWS canonical form
///
/// Parameters are sorted by encoded name, with ties broken by encoded value,
as required by the
@@ -519,7 +526,7 @@ fn append_normalized_whitespace_value(headers: &'_ mut
String, input: &str) {
/// Canonicalizes headers into the AWS Canonical Form.
///
///
<https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
-fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
+fn canonicalize_headers(header_map: &HeaderMap) -> Result<(String, String)> {
let mut headers = BTreeMap::<&str, Vec<&str>>::new();
let mut value_count = 0;
let mut value_bytes = 0;
@@ -531,7 +538,7 @@ fn canonicalize_headers(header_map: &HeaderMap) -> (String,
String) {
continue;
}
- let value = std::str::from_utf8(value.as_bytes()).unwrap();
+ let value =
std::str::from_utf8(value.as_bytes()).map_err(invalid_request)?;
key_bytes += key.len();
value_bytes += value.len();
value_count += 1;
@@ -559,7 +566,7 @@ fn canonicalize_headers(header_map: &HeaderMap) -> (String,
String) {
canonical_headers.push('\n');
}
- (signed_headers, canonical_headers)
+ Ok((signed_headers, canonical_headers))
}
/// Credentials sourced from the instance metadata service
@@ -1772,7 +1779,7 @@ mod tests {
headers.insert("authorization", "SHOULD_BE_IGNORED".parse().unwrap());
headers.insert("content-length", "1337".parse().unwrap());
- let (signed_headers, canonical_headers) =
super::canonicalize_headers(&headers);
+ let (signed_headers, canonical_headers) =
super::canonicalize_headers(&headers).unwrap();
assert_eq!(
signed_headers,
diff --git a/src/azure/builder.rs b/src/azure/builder.rs
index d1d68fe..2d1734f 100644
--- a/src/azure/builder.rs
+++ b/src/azure/builder.rs
@@ -64,6 +64,9 @@ enum Error {
source: url::ParseError,
},
+ #[error("Endpoint URL cannot be used as a base")]
+ InvalidEndpoint,
+
#[error("Account must be specified")]
MissingAccount {},
@@ -1246,6 +1249,10 @@ impl MicrosoftAzureBuilder {
// Allow overriding defaults. Values taken from
// from
https://docs.rs/azure_storage/0.2.0/src/azure_storage/core/clients/storage_account_client.rs.html#129-141
let url = url_from_env("AZURITE_BLOB_STORAGE_URL",
"http://127.0.0.1:10000";)?;
+ if url.cannot_be_a_base() {
+ return Err(Error::InvalidEndpoint.into());
+ }
+
let credential = if let Some(k) = self.access_key {
AzureCredential::AccessKey(AzureAccessKey::try_new(&k)?)
} else if let Some(bearer_token) = self.bearer_token {
@@ -1277,6 +1284,10 @@ impl MicrosoftAzureBuilder {
Error::UnableToParseUrl { url, source }
})?;
+ if url.cannot_be_a_base() {
+ return Err(Error::InvalidEndpoint.into());
+ }
+
let credential = if let Some(credential) = self.credentials {
credential
} else {
diff --git a/src/azure/credential.rs b/src/azure/credential.rs
index 5537e34..3746c30 100644
--- a/src/azure/credential.rs
+++ b/src/azure/credential.rs
@@ -22,8 +22,8 @@ use crate::client::builder::{HttpRequestBuilder,
add_query_pairs};
use crate::client::retry::RetryExt;
use crate::client::token::{TemporaryToken, TokenCache};
use crate::client::{
- CredentialProvider, CryptoProvider, DigestAlgorithm, HttpClient,
HttpError, HttpRequest,
- TokenProvider, crypto_provider,
+ CredentialProvider, CryptoProvider, DigestAlgorithm, HttpClient,
HttpError, HttpErrorKind,
+ HttpRequest, TokenProvider, crypto_provider,
};
use async_trait::async_trait;
use base64::Engine;
@@ -277,8 +277,8 @@ impl<'a> AzureAuthorizer<'a> {
match self.credential {
AzureCredential::AccessKey(key) => {
+ let url =
Url::parse(&request.uri().to_string()).map_err(invalid_request)?;
let crypto = self.crypto()?;
- let url = Url::parse(&request.uri().to_string()).unwrap();
let signature = generate_authorization(
crypto,
request.headers(),
@@ -288,21 +288,19 @@ impl<'a> AzureAuthorizer<'a> {
key,
)?;
- // "signature" is a base 64 encoded string so it should never
- // contain illegal characters
request.headers_mut().append(
AUTHORIZATION,
- HeaderValue::from_str(signature.as_str()).unwrap(),
+
HeaderValue::from_str(&signature).map_err(invalid_request)?,
);
}
AzureCredential::BearerToken(token) => {
request.headers_mut().append(
AUTHORIZATION,
- HeaderValue::from_str(format!("Bearer
{token}").as_str()).unwrap(),
+ HeaderValue::from_str(&format!("Bearer
{token}")).map_err(invalid_request)?,
);
}
AzureCredential::SASToken(query_pairs) => {
- add_query_pairs(request.uri_mut(), query_pairs);
+ add_query_pairs(request.uri_mut(),
query_pairs).map_err(invalid_request)?;
}
}
Ok(())
@@ -328,7 +326,7 @@ impl CredentialExt for HttpRequestBuilder {
account: &str,
) -> crate::Result<Self> {
let (client, request) = self.into_parts();
- let mut request = request.expect("request valid");
+ let mut request = request.map_err(invalid_request)?;
match credential.as_deref() {
Some(credential) => {
@@ -347,6 +345,13 @@ impl CredentialExt for HttpRequestBuilder {
}
}
+fn invalid_request(error: impl std::error::Error + Send + Sync + 'static) ->
crate::Error {
+ crate::Error::Generic {
+ store: STORE,
+ source: Box::new(HttpError::new(HttpErrorKind::InvalidRequest, error)),
+ }
+}
+
/// Generate signed key for authorization via access keys
///
<https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key>
fn generate_authorization(
@@ -357,7 +362,7 @@ fn generate_authorization(
account: &str,
key: &AzureAccessKey,
) -> crate::Result<String> {
- let str_to_sign = string_to_sign(h, u, method, account);
+ let str_to_sign = string_to_sign(h, u, method, account)?;
let mut ctx = crypto.hmac(DigestAlgorithm::Sha256, &key.0)?;
ctx.update(str_to_sign.as_bytes());
let auth = ctx.finish()?;
@@ -524,7 +529,7 @@ fn string_to_sign_user_delegation_sas(
}
///
<https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-signature-string>
-fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) ->
String {
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) ->
crate::Result<String> {
// content length must only be specified if != 0
// this is valid from 2015-02-21
let content_length = h
@@ -535,7 +540,7 @@ fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method,
account: &str) -> Str
.flatten()
.filter(|&v| v != "0")
.unwrap_or_default();
- format!(
+ Ok(format!(
"{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
method.as_ref(),
add_if_exists(h, &CONTENT_ENCODING),
@@ -549,19 +554,18 @@ fn string_to_sign(h: &HeaderMap, u: &Url, method:
&Method, account: &str) -> Str
add_if_exists(h, &IF_NONE_MATCH),
add_if_exists(h, &IF_UNMODIFIED_SINCE),
add_if_exists(h, &RANGE),
- canonicalize_header(h),
+ canonicalize_header(h)?,
canonicalize_resource(account, u)
- )
+ ))
}
///
<https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-headers-string>
-fn canonicalize_header(headers: &HeaderMap) -> String {
+fn canonicalize_header(headers: &HeaderMap) -> crate::Result<String> {
let mut names = headers
.iter()
.filter(|&(k, _)| k.as_str().starts_with("x-ms"))
- // TODO remove unwraps
- .map(|(k, _)| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
- .collect::<Vec<_>>();
+ .map(|(k, v)| Ok((k.as_str(), v.to_str().map_err(invalid_request)?)))
+ .collect::<crate::Result<Vec<_>>>()?;
names.sort_unstable();
let mut result = String::new();
@@ -571,7 +575,7 @@ fn canonicalize_header(headers: &HeaderMap) -> String {
result.push_str(value);
result.push('\n');
}
- result
+ Ok(result)
}
///
<https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-resource-string>
diff --git a/src/client/builder.rs b/src/client/builder.rs
index cca6f93..5f267b4 100644
--- a/src/client/builder.rs
+++ b/src/client/builder.rs
@@ -17,7 +17,7 @@
use crate::client::{HttpClient, HttpError, HttpErrorKind, HttpRequest,
HttpRequestBody};
use http::header::{InvalidHeaderName, InvalidHeaderValue};
-use http::uri::InvalidUri;
+use http::uri::{InvalidUri, InvalidUriParts};
use http::{HeaderName, HeaderValue, Method, Uri};
#[derive(Debug, thiserror::Error)]
@@ -25,6 +25,9 @@ pub(crate) enum RequestBuilderError {
#[error("Invalid URI")]
InvalidUri(#[from] InvalidUri),
+ #[error("Invalid URI parts: {0}")]
+ InvalidUriParts(#[from] InvalidUriParts),
+
#[error("Invalid Header Value")]
InvalidHeaderValue(#[from] InvalidHeaderValue),
@@ -40,7 +43,7 @@ pub(crate) enum RequestBuilderError {
impl From<RequestBuilderError> for HttpError {
fn from(value: RequestBuilderError) -> Self {
- Self::new(HttpErrorKind::Request, value)
+ Self::new(HttpErrorKind::InvalidRequest, value)
}
}
@@ -203,29 +206,18 @@ impl HttpRequestBuilder {
feature = "azure-base"
))]
pub(crate) fn query<T: serde::Serialize + ?Sized>(mut self, query: &T) ->
Self {
- let mut error = None;
- if let Ok(ref mut req) = self.request {
+ self.request = self.request.and_then(|mut req| {
let mut out = format!("{}?", req.uri().path());
let start_position = out.len();
let mut encoder = form_urlencoded::Serializer::for_suffix(&mut
out, start_position);
let serializer = serde_urlencoded::Serializer::new(&mut encoder);
+ query.serialize(serializer)?;
- if let Err(err) = query.serialize(serializer) {
- error = Some(err.into());
- }
-
- match http::uri::PathAndQuery::from_maybe_shared(out) {
- Ok(p) => {
- let mut parts = req.uri().clone().into_parts();
- parts.path_and_query = Some(p);
- *req.uri_mut() = Uri::from_parts(parts).unwrap();
- }
- Err(err) => error = Some(err.into()),
- }
- }
- if let Some(err) = error {
- self.request = Err(err);
- }
+ let mut parts = req.uri().clone().into_parts();
+ parts.path_and_query =
Some(http::uri::PathAndQuery::from_maybe_shared(out)?);
+ *req.uri_mut() = Uri::from_parts(parts)?;
+ Ok(req)
+ });
self
}
@@ -264,7 +256,10 @@ impl HttpRequestBuilder {
}
#[cfg(any(test, feature = "azure-base"))]
-pub(crate) fn add_query_pairs<I, K, V>(uri: &mut Uri, query_pairs: I)
+pub(crate) fn add_query_pairs<I, K, V>(
+ uri: &mut Uri,
+ query_pairs: I,
+) -> Result<(), RequestBuilderError>
where
I: IntoIterator,
I::Item: std::borrow::Borrow<(K, V)>,
@@ -289,8 +284,9 @@ where
serializer.extend_pairs(query_pairs);
- parts.path_and_query = Some(out.try_into().unwrap());
- *uri = Uri::from_parts(parts).unwrap();
+ parts.path_and_query = Some(out.try_into()?);
+ *uri = Uri::from_parts(parts)?;
+ Ok(())
}
#[cfg(test)]
@@ -301,16 +297,16 @@ mod tests {
fn test_add_query_pairs() {
let mut uri = Uri::from_static("https://[email protected]/bananas";);
- add_query_pairs(&mut uri, [("foo", "1")]);
+ add_query_pairs(&mut uri, [("foo", "1")]).unwrap();
assert_eq!(uri.to_string(), "https://[email protected]/bananas?foo=1";);
- add_query_pairs(&mut uri, [("bingo", "foo"), ("auth", "test")]);
+ add_query_pairs(&mut uri, [("bingo", "foo"), ("auth",
"test")]).unwrap();
assert_eq!(
uri.to_string(),
"https://[email protected]/bananas?foo=1&bingo=foo&auth=test";
);
- add_query_pairs(&mut uri, [("t1", "funky shenanigans"), ("a", "π")]);
+ add_query_pairs(&mut uri, [("t1", "funky shenanigans"), ("a",
"π")]).unwrap();
assert_eq!(
uri.to_string(),
"https://[email protected]/bananas?foo=1&bingo=foo&auth=test&t1=funky+shenanigans&a=%F0%9F%98%80";
@@ -320,7 +316,7 @@ mod tests {
#[test]
fn test_add_query_pairs_no_path() {
let mut uri = Uri::from_static("https://[email protected]";);
- add_query_pairs(&mut uri, [("foo", "1")]);
+ add_query_pairs(&mut uri, [("foo", "1")]).unwrap();
assert_eq!(uri.to_string(), "https://[email protected]/?foo=1";);
}
@@ -348,6 +344,42 @@ mod tests {
);
}
+ #[cfg(feature = "reqwest")]
+ #[test]
+ fn test_query_invalid_uri() {
+ let client = HttpClient::new(reqwest::Client::new());
+ let (_, result) = HttpRequestBuilder::new(client)
+ .uri("example.invalid")
+ .query(&[("list-type", "2")])
+ .header("x-test", "valid")
+ .into_parts();
+ assert!(matches!(
+ result,
+ Err(RequestBuilderError::InvalidUriParts(_))
+ ));
+ }
+
+ #[test]
+ fn test_add_query_pairs_invalid_uri() {
+ let mut uri = Uri::from_static("example.invalid");
+ let original = uri.clone();
+ assert!(add_query_pairs(&mut uri, [("foo", "bar")]).is_err());
+ assert_eq!(uri, original);
+ }
+
+ #[cfg(feature = "reqwest")]
+ #[test]
+ fn test_query_serialization_error() {
+ let client = HttpClient::new(reqwest::Client::new());
+ let (_, result) = HttpRequestBuilder::new(client)
+ .uri("example.invalid")
+ .query(&42)
+ .uri("https://example.invalid";)
+ .query(&[("foo", "bar")])
+ .into_parts();
+ assert!(matches!(result, Err(RequestBuilderError::SerdeUrl(_))));
+ }
+
fn assert_request_uri(builder: HttpRequestBuilder, expected: &str) {
assert_eq!(builder.into_parts().1.unwrap().uri().to_string(), expected)
}
diff --git a/src/client/http/connection.rs b/src/client/http/connection.rs
index 0558a7e..a088d0f 100644
--- a/src/client/http/connection.rs
+++ b/src/client/http/connection.rs
@@ -28,6 +28,8 @@ use std::error::Error;
use std::sync::Arc;
#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
use tokio::runtime::Handle;
+#[cfg(feature = "reqwest")]
+use url::Url;
/// An HTTP protocol error
///
@@ -49,6 +51,10 @@ pub struct HttpError {
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum HttpErrorKind {
+ /// Invalid input prevented the request from being built or signed
+ ///
+ /// Retrying the same input cannot succeed. Will not be automatically
retried
+ InvalidRequest,
/// An error occurred whilst connecting to the remote
///
/// Will be automatically retried
@@ -97,7 +103,9 @@ impl HttpError {
#[cfg(target_arch = "wasm32")]
let is_connect = || false;
- let mut kind = if e.is_timeout() {
+ let mut kind = if e.is_builder() {
+ HttpErrorKind::InvalidRequest
+ } else if e.is_timeout() {
HttpErrorKind::Timeout
} else if is_connect() {
HttpErrorKind::Connect
@@ -221,7 +229,8 @@ impl HttpService for reqwest::Client {
async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
let (parts, body) = req.into_parts();
- let url = parts.uri.to_string().parse().unwrap();
+ let url = Url::parse(&parts.uri.to_string())
+ .map_err(|e| HttpError::new(HttpErrorKind::InvalidRequest, e))?;
let mut req = reqwest::Request::new(parts.method, url);
*req.headers_mut() = parts.headers;
*req.body_mut() = Some(body.into_reqwest());
@@ -245,7 +254,8 @@ impl HttpService for reqwest::Client {
use wasm_bindgen_futures::spawn_local;
let (parts, body) = req.into_parts();
- let url = parts.uri.to_string().parse().unwrap();
+ let url = Url::parse(&parts.uri.to_string())
+ .map_err(|e| HttpError::new(HttpErrorKind::InvalidRequest, e))?;
let mut req = reqwest::Request::new(parts.method, url);
*req.headers_mut() = parts.headers;
*req.body_mut() = Some(body.into_reqwest());
diff --git a/src/client/retry.rs b/src/client/retry.rs
index c79012a..8c6dd8f 100644
--- a/src/client/retry.rs
+++ b/src/client/retry.rs
@@ -19,7 +19,7 @@
use crate::PutPayload;
use crate::client::backoff::{Backoff, BackoffConfig};
-use crate::client::builder::HttpRequestBuilder;
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
use crate::client::{HttpClient, HttpError, HttpErrorKind, HttpRequest,
HttpResponse};
use futures_util::future::BoxFuture;
use http::StatusCode;
@@ -38,7 +38,7 @@ pub struct RetryError(Box<RetryErrorImpl>);
/// Box error to avoid large error variant
#[derive(Debug)]
struct RetryErrorImpl {
- method: Method,
+ method: Option<Method>,
uri: Option<Uri>,
retries: usize,
max_retries: usize,
@@ -49,10 +49,14 @@ struct RetryErrorImpl {
impl std::fmt::Display for RetryError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(f, "Error performing {} ", self.0.method)?;
+ match &self.0.method {
+ Some(method) => write!(f, "Error performing {method} ")?,
+ None => write!(f, "Error performing request ")?,
+ }
match &self.0.uri {
Some(uri) => write!(f, "{uri} ")?,
- None => write!(f, "REDACTED ")?,
+ None if self.0.method.is_some() => write!(f, "REDACTED ")?,
+ None => {}
}
write!(f, "in {:?}", self.0.elapsed)?;
if self.0.retries != 0 {
@@ -199,6 +203,7 @@ impl From<RetryError> for std::io::Error {
}
_ => match err.inner() {
RequestError::Http(h) => match h.kind() {
+ HttpErrorKind::InvalidRequest => ErrorKind::InvalidInput,
HttpErrorKind::Timeout => ErrorKind::TimedOut,
HttpErrorKind::Connect => ErrorKind::NotConnected,
_ => ErrorKind::Other,
@@ -315,7 +320,7 @@ impl RetryableRequestBuilder {
/// A retryable request
pub(crate) struct RetryableRequest {
client: HttpClient,
- http: HttpRequest,
+ http: Result<HttpRequest, RequestBuilderError>,
sensitive: bool,
idempotent: Option<bool>,
@@ -332,9 +337,10 @@ impl RetryableRequest {
}
fn err(&self, error: RequestError, ctx: &RetryContext) -> RetryError {
+ let request = self.http.as_ref().ok();
RetryError(Box::new(RetryErrorImpl {
- uri: (!self.sensitive).then(|| self.http.uri().clone()),
- method: self.http.method().clone(),
+ uri: request.filter(|_| !self.sensitive).map(|r| r.uri().clone()),
+ method: request.map(|r| r.method().clone()),
retries: ctx.retries,
max_retries: ctx.max_retries,
elapsed: ctx.start.elapsed(),
@@ -344,8 +350,22 @@ impl RetryableRequest {
}
pub(crate) async fn send(self, ctx: &mut RetryContext) ->
Result<HttpResponse> {
+ let http = match self.http {
+ Ok(ref http) => http,
+ Err(error) => {
+ return Err(RetryError(Box::new(RetryErrorImpl {
+ method: None,
+ uri: None,
+ retries: ctx.retries,
+ max_retries: ctx.max_retries,
+ elapsed: ctx.start.elapsed(),
+ retry_timeout: ctx.retry_timeout,
+ inner: RequestError::Http(error.into()),
+ })));
+ }
+ };
loop {
- let mut request = self.http.clone();
+ let mut request = http.clone();
if let Some(payload) = &self.payload {
*request.body_mut() = payload.clone().into();
@@ -432,14 +452,14 @@ impl RetryableRequest {
}
}
Err(e) => {
- let is_idempotent = self
- .idempotent
- .unwrap_or_else(|| self.http.method().is_safe());
+ let is_idempotent = self.idempotent.unwrap_or_else(||
http.method().is_safe());
let do_retry = match e.kind() {
HttpErrorKind::Connect | HttpErrorKind::Request =>
true, // Request not sent, can retry
HttpErrorKind::Timeout | HttpErrorKind::Interrupted =>
is_idempotent,
- HttpErrorKind::Unknown | HttpErrorKind::Decode =>
false,
+ HttpErrorKind::InvalidRequest
+ | HttpErrorKind::Unknown
+ | HttpErrorKind::Decode => false,
};
if ctx.exhausted() || !do_retry {
@@ -486,7 +506,6 @@ impl RetryExt for HttpRequestBuilder {
fn retryable_request(self) -> RetryableRequest {
let (client, request) = self.into_parts();
- let request = request.expect("request must be valid");
RetryableRequest {
client,
@@ -511,7 +530,10 @@ mod tests {
use crate::RetryConfig;
use crate::client::mock_server::MockServer;
use crate::client::retry::{RequestError, RetryContext, RetryExt,
body_contains_error};
- use crate::client::{HttpClient, HttpError, HttpErrorKind, HttpResponse};
+ use crate::client::{
+ HttpClient, HttpError, HttpErrorKind, HttpRequest, HttpResponse,
HttpService,
+ };
+ use async_trait::async_trait;
use http::Method;
use http::StatusCode;
use hyper::Response;
@@ -523,10 +545,46 @@ mod tests {
use reqwest::Client;
use std::convert::Infallible;
use std::error::Error;
+ use std::sync::Arc;
+ use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::net::TcpListener;
use tokio::time::timeout;
+ #[tokio::test]
+ async fn test_invalid_request_no_retry() {
+ #[derive(Debug)]
+ struct InvalidRequestService(Arc<AtomicUsize>);
+
+ #[async_trait]
+ impl HttpService for InvalidRequestService {
+ async fn call(&self, _: HttpRequest) -> Result<HttpResponse,
HttpError> {
+ self.0.fetch_add(1, Ordering::SeqCst);
+ Err(HttpError::new(
+ HttpErrorKind::InvalidRequest,
+ std::io::Error::new(std::io::ErrorKind::InvalidInput,
"invalid request"),
+ ))
+ }
+ }
+
+ let calls = Arc::new(AtomicUsize::new(0));
+ let client =
HttpClient::new(InvalidRequestService(Arc::clone(&calls)));
+ let config = RetryConfig {
+ max_retries: 2,
+ ..Default::default()
+ };
+ let error = client
+ .request(Method::GET, "https://example.invalid/object";)
+ .send_retry(&config)
+ .await
+ .unwrap_err();
+ assert_eq!(calls.load(Ordering::SeqCst), 1);
+ assert!(matches!(
+ error.inner(),
+ RequestError::Http(error) if error.kind() ==
HttpErrorKind::InvalidRequest
+ ));
+ }
+
#[test]
fn test_body_contains_error() {
// Example error message provided by
https://repost.aws/knowledge-center/s3-resolve-200-internalerror
diff --git a/src/http/mod.rs b/src/http/mod.rs
index 1c83d45..43b18d0 100644
--- a/src/http/mod.rs
+++ b/src/http/mod.rs
@@ -68,6 +68,9 @@ enum Error {
url: String,
},
+ #[error("Endpoint URL cannot be used as a base")]
+ InvalidEndpoint,
+
#[error("Unable to extract metadata from headers: {}", source)]
Metadata {
source: crate::client::header::Error,
@@ -283,6 +286,10 @@ impl HttpBuilder {
let url = self.url.ok_or(Error::MissingUrl)?;
let parsed = Url::parse(&url).map_err(|source| Error::UnableToParseUrl
{ url, source })?;
+ if parsed.cannot_be_a_base() {
+ return Err(Error::InvalidEndpoint.into());
+ }
+
let client =
http_connector(self.http_connector)?.connect(&self.client_options)?;
Ok(HttpStore {
diff --git a/tests/request_errors.rs b/tests/request_errors.rs
new file mode 100644
index 0000000..e620e99
--- /dev/null
+++ b/tests/request_errors.rs
@@ -0,0 +1,316 @@
+// 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.
+
+#![cfg(all(feature = "aws", feature = "azure", feature = "gcp", feature =
"http"))]
+
+use async_trait::async_trait;
+use object_store::ClientOptions;
+use object_store::client::{
+ HttpClient, HttpConnector, HttpError, HttpErrorKind, HttpRequest,
HttpResponse, HttpService,
+};
+use std::error::Error;
+use std::sync::{
+ Arc,
+ atomic::{AtomicUsize, Ordering},
+};
+
+fn find_error<'a, T: Error + 'static>(mut error: &'a (dyn Error + 'static)) ->
Option<&'a T> {
+ loop {
+ if let Some(error) = error.downcast_ref::<T>() {
+ return Some(error);
+ }
+ error = error.source()?;
+ }
+}
+
+fn assert_invalid_request(error: &object_store::Error) {
+ let http_error = find_error::<HttpError>(error).expect("HttpError missing
from error chain");
+ assert_eq!(http_error.kind(), HttpErrorKind::InvalidRequest, "{error}");
+}
+
+#[derive(Debug, Clone)]
+struct NoNetwork(Arc<AtomicUsize>);
+
+#[async_trait]
+impl HttpService for NoNetwork {
+ async fn call(&self, _: HttpRequest) -> Result<HttpResponse, HttpError> {
+ self.0.fetch_add(1, Ordering::SeqCst);
+ Ok(http::Response::builder()
+ .status(400)
+ .body(String::from("test transport").into())
+ .unwrap())
+ }
+}
+
+impl HttpConnector for NoNetwork {
+ fn connect(&self, _: &ClientOptions) -> object_store::Result<HttpClient> {
+ Ok(HttpClient::new(self.clone()))
+ }
+}
+
+mod requests {
+ use super::*;
+ use futures_util::TryStreamExt;
+ use object_store::{Attribute, Attributes, ObjectStore, ObjectStoreExt,
PutOptions};
+
+ fn build_store(
+ provider: &str,
+ endpoint: &str,
+ signed: bool,
+ bad_token: bool,
+ transport: NoNetwork,
+ ) -> object_store::Result<Box<dyn ObjectStore>> {
+ let token = if bad_token { "test\ninvalid" } else { "test" };
+ match provider {
+ "s3" => object_store::aws::AmazonS3Builder::new()
+ .with_bucket_name("bucket")
+ .with_region("us-east-1")
+ .with_access_key_id("test")
+ .with_secret_access_key("test")
+ .with_token(token)
+ .with_endpoint(endpoint)
+ .with_allow_http(true)
+ .with_virtual_hosted_style_request(true)
+ .with_skip_signature(!signed)
+ .with_http_connector(transport)
+ .build()
+ .map(|s| Box::new(s) as _),
+ "gcs" => object_store::gcp::GoogleCloudStorageBuilder::new()
+ .with_bucket_name("bucket")
+ .with_base_url(endpoint)
+ .with_bearer_token(token)
+ .with_skip_signature(!signed)
+
.with_client_options(ClientOptions::new().with_allow_http(true))
+ .with_http_connector(transport)
+ .build()
+ .map(|s| Box::new(s) as _),
+ "azure" => object_store::azure::MicrosoftAzureBuilder::new()
+ .with_account("account")
+ .with_container_name("container")
+ .with_endpoint(endpoint.into())
+ .with_bearer_token_authorization(token)
+ .with_allow_http(true)
+ .with_skip_signature(!signed)
+ .with_http_connector(transport)
+ .build()
+ .map(|s| Box::new(s) as _),
+ "http" => object_store::http::HttpBuilder::new()
+ .with_url(endpoint)
+
.with_client_options(ClientOptions::new().with_allow_http(true))
+ .with_http_connector(transport)
+ .build()
+ .map(|s| Box::new(s) as _),
+ _ => unreachable!(),
+ }
+ }
+
+ async fn check_request(provider: &str, endpoint: &str, signed: bool, op:
&str) {
+ let calls = Arc::new(AtomicUsize::new(0));
+ let mut stage = "build";
+ let ret = (async {
+ let store = build_store(
+ provider,
+ endpoint,
+ signed,
+ op == "bad_token",
+ NoNetwork(calls.clone()),
+ )?;
+ stage = "operation";
+ match op {
+ "list" =>
store.list(None).try_collect::<Vec<_>>().await.map(|_| ()),
+ "bad_header" => {
+ let mut attributes = Attributes::new();
+ attributes.insert(Attribute::ContentType,
"test\ninvalid".into());
+ store
+ .put_opts(
+ &"key".into(),
+ "data".into(),
+ PutOptions {
+ attributes,
+ ..Default::default()
+ },
+ )
+ .await
+ .map(|_| ())
+ }
+ "get" | "bad_token" => store.get(&"key".into()).await.map(|_|
()),
+ _ => unreachable!("unknown operation: {op}"),
+ }
+ })
+ .await;
+ let error = ret.expect_err("expected request failure");
+ // The custom transport accepts relative URIs. Signed S3 requests
reject
+ // them earlier because the authorizer requires an absolute URL.
+ let expected_calls = usize::from(
+ endpoint == "http://bucket.example.invalid"; &&
!op.starts_with("bad_")
+ || provider == "gcs" && endpoint == "/relative"
+ || provider == "s3" && endpoint == "/relative" && !signed,
+ );
+ assert_eq!(
+ calls.load(Ordering::SeqCst),
+ expected_calls,
+ "{provider} {endpoint} {op} {stage}"
+ );
+ if expected_calls == 1 {
+ assert!(
+ error
+ .to_string()
+ .contains("400 Bad Request: test transport"),
+ "{error}"
+ );
+ assert!(find_error::<HttpError>(&error).is_none());
+ } else if stage == "operation" {
+ assert_invalid_request(&error);
+ } else {
+ // Invalid endpoints rejected at build time have provider-specific
errors.
+ assert!(find_error::<HttpError>(&error).is_none());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_invalid_requests() {
+ for provider in ["s3", "gcs", "azure", "http"] {
+ for endpoint in [
+ "bucket.example.invalid",
+ "127.0.0.1:18764",
+ "http://[";,
+ "/relative",
+ "data:text/plain,test",
+ "mailto:[email protected]";,
+ "http://bucket.example.invalid";,
+ ] {
+ for signed in [true, false] {
+ if provider == "http" && signed {
+ continue;
+ }
+ for op in ["list", "get"] {
+ check_request(provider, endpoint, signed, op).await;
+ }
+ }
+ }
+ }
+ for provider in ["s3", "gcs", "azure", "http"] {
+ for signed in [true, false] {
+ if provider == "http" && signed {
+ continue;
+ }
+ check_request(
+ provider,
+ "http://bucket.example.invalid";,
+ signed,
+ "bad_header",
+ )
+ .await;
+ }
+ if provider != "http" {
+ check_request(provider, "http://bucket.example.invalid";, true,
"bad_token").await;
+ }
+ }
+ }
+}
+
+mod signing {
+ use super::*;
+ use object_store::{Attribute, Attributes, ObjectStore, ObjectStoreExt,
PutOptions};
+
+ async fn error_without_transport<T>(
+ name: &str,
+ calls: &AtomicUsize,
+ future: impl std::future::Future<Output = object_store::Result<T>>,
+ ) -> object_store::Error {
+ let error = future.await.err().expect("expected signing failure");
+ assert_eq!(calls.load(Ordering::SeqCst), 0, "{name}");
+ error
+ }
+
+ #[tokio::test]
+ async fn test_invalid_signing() {
+ let calls = Arc::new(AtomicUsize::new(0));
+ let mut headers = http::HeaderMap::new();
+ headers.insert("x-test",
http::HeaderValue::from_bytes(&[0xff]).unwrap());
+ let aws = object_store::aws::AmazonS3Builder::new()
+ .with_bucket_name("bucket")
+ .with_region("us-east-1")
+ .with_access_key_id("test")
+ .with_secret_access_key("test")
+ .with_endpoint("http://bucket.example.invalid";)
+ .with_virtual_hosted_style_request(true)
+ .with_client_options(
+ ClientOptions::new()
+ .with_allow_http(true)
+ .with_default_headers(headers),
+ )
+ .with_http_connector(NoNetwork(calls.clone()))
+ .build()
+ .unwrap();
+ let error =
+ error_without_transport("s3_default_header_non_utf8", &calls,
aws.get(&"key".into()))
+ .await;
+ assert_invalid_request(&error);
+ let azure = object_store::azure::MicrosoftAzureBuilder::new()
+ .with_account("account")
+ .with_container_name("container")
+ .with_endpoint("http://bucket.example.invalid".into())
+ .with_access_key("dGVzdA==")
+ .with_allow_http(true)
+ .with_http_connector(NoNetwork(calls.clone()))
+ .build()
+ .unwrap();
+ let mut attributes = Attributes::new();
+ attributes.insert(Attribute::Metadata("label".into()), "δΈζ".into());
+ let error = error_without_transport(
+ "azure_shared_key_unicode_metadata",
+ &calls,
+ azure.put_opts(
+ &"key".into(),
+ "data".into(),
+ PutOptions {
+ attributes,
+ ..Default::default()
+ },
+ ),
+ )
+ .await;
+ assert_invalid_request(&error);
+ }
+}
+
+mod transport {
+ use super::*;
+ use futures_util::TryStreamExt;
+ use object_store::{ObjectStore, ObjectStoreExt};
+
+ #[tokio::test]
+ async fn test_reqwest_invalid_uri() {
+ let gcs_relative = object_store::gcp::GoogleCloudStorageBuilder::new()
+ .with_bucket_name("bucket")
+ .with_base_url("/relative")
+ .with_bearer_token("test")
+ .with_skip_signature(true)
+ .with_client_options(ClientOptions::new().with_allow_http(true))
+ .build()
+ .unwrap();
+ let error = gcs_relative.get(&"key".into()).await.unwrap_err();
+ assert_invalid_request(&error);
+ let error = gcs_relative
+ .list(None)
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap_err();
+ assert_invalid_request(&error);
+ }
+}
