Compare commits

...

15 Commits

Author SHA1 Message Date
Alex Auvolat c99f55c420 Add restriction on part ordering in CompleteMultipartUpload 2022-01-25 12:45:00 +01:00
Alex Auvolat acdf893362 Fix partnumber 2022-01-25 12:25:23 +01:00
Alex Auvolat 338b1b83ee Implement part_number for GetObject 2022-01-24 21:04:42 +01:00
Alex Auvolat 6dab836f3a Multipart improvements
- support part_number for HeadObject
- add checks in complete_multipart_upload
2022-01-24 21:04:40 +01:00
Alex Auvolat 513a6b15f9 Handle OPTIONS on website endpoint 2022-01-24 12:32:28 +01:00
Alex Auvolat ea7fb901eb Implement {Put,Get,Delete}BucketCors and CORS in general
- OPTIONS request against API endpoint
- Returning corresponding CORS headers on API calls
- Returning corresponding CORS headers on website GET's
2022-01-24 11:58:00 +01:00
Trinity Pointard 820924534a use clamp instead of min(max()) 2022-01-24 11:56:59 +01:00
Quentin Dufour 94f0e7c135 Test ListParts endpoint with awscli 2022-01-21 10:42:35 +01:00
Quentin Dufour 440374524b Implement ListParts 2022-01-21 10:42:30 +01:00
Quentin Dufour fe003d6fbc Add ListPartsResult structure 2022-01-20 16:38:55 +01:00
trinity-1686a e55fa38c99 Add date verification to presigned urls (#196)
fix #96
fix #162 by returning Forbidden instead Bad Request

Co-authored-by: Trinity Pointard <trinity.pointard@gmail.com>
Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/196
Co-authored-by: trinity-1686a <trinity.pointard@gmail.com>
Co-committed-by: trinity-1686a <trinity.pointard@gmail.com>
2022-01-18 12:22:31 +01:00
trinity-1686a 178e35f868 refactor s3_router and api_server to make unused Endpoint parameters more obvious 2022-01-17 15:50:24 +01:00
Alex Auvolat 7c049f1c94 Fix extreme value to be less extreme so that integration test works on 32bits 2022-01-17 12:56:29 +01:00
Jill fdcddbe168 Fix Multipart Upload with WinSCP (#164) (#193)
Closes #164.

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/193
Co-authored-by: Jill <kokakiwi@deuxfleurs.fr>
Co-committed-by: Jill <kokakiwi@deuxfleurs.fr>
2022-01-17 11:18:40 +01:00
Jill b45dcc1925 Support STREAMING-AWS4-HMAC-SHA256-PAYLOAD (#64) (#156)
Closes #64.

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/156
Co-authored-by: Jill <kokakiwi@deuxfleurs.fr>
Co-committed-by: Jill <kokakiwi@deuxfleurs.fr>
2022-01-17 10:55:31 +01:00
29 changed files with 2239 additions and 576 deletions
Generated
+39
View File
@@ -433,7 +433,9 @@ dependencies = [
"idna",
"log",
"md-5",
"nom",
"percent-encoding",
"pin-project",
"quick-xml",
"roxmltree",
"serde",
@@ -967,6 +969,12 @@ dependencies = [
"autocfg",
]
[[package]]
name = "minimal-lexical"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
[[package]]
name = "mio"
version = "0.7.13"
@@ -1011,6 +1019,17 @@ dependencies = [
"tokio-util",
]
[[package]]
name = "nom"
version = "7.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b1d11e1ef389c76fe5b81bcaf2ea32cf88b62bc494e19f493d0b30e7a930109"
dependencies = [
"memchr",
"minimal-lexical",
"version_check",
]
[[package]]
name = "ntapi"
version = "0.3.6"
@@ -1092,6 +1111,26 @@ version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
[[package]]
name = "pin-project"
version = "1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "576bc800220cc65dac09e99e97b08b358cfab6e17078de8dc5fee223bd2d0c08"
dependencies = [
"pin-project-internal",
]
[[package]]
name = "pin-project-internal"
version = "1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e8fe8163d14ce7f0cdac2e040116f22eac817edabff0be91e8aff7e9accf389"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "pin-project-lite"
version = "0.2.7"
+53
View File
@@ -671,7 +671,9 @@ in
idna = rustPackages."registry+https://github.com/rust-lang/crates.io-index".idna."0.2.3" { inherit profileName; };
log = rustPackages."registry+https://github.com/rust-lang/crates.io-index".log."0.4.14" { inherit profileName; };
md5 = rustPackages."registry+https://github.com/rust-lang/crates.io-index".md-5."0.9.1" { inherit profileName; };
nom = rustPackages."registry+https://github.com/rust-lang/crates.io-index".nom."7.1.0" { inherit profileName; };
percent_encoding = rustPackages."registry+https://github.com/rust-lang/crates.io-index".percent-encoding."2.1.0" { inherit profileName; };
pin_project = rustPackages."registry+https://github.com/rust-lang/crates.io-index".pin-project."1.0.8" { inherit profileName; };
quick_xml = rustPackages."registry+https://github.com/rust-lang/crates.io-index".quick-xml."0.21.0" { inherit profileName; };
roxmltree = rustPackages."registry+https://github.com/rust-lang/crates.io-index".roxmltree."0.14.1" { inherit profileName; };
serde = rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde."1.0.130" { inherit profileName; };
@@ -1321,6 +1323,16 @@ in
};
});
"registry+https://github.com/rust-lang/crates.io-index".minimal-lexical."0.2.1" = overridableMkRustCrate (profileName: rec {
name = "minimal-lexical";
version = "0.2.1";
registry = "registry+https://github.com/rust-lang/crates.io-index";
src = fetchCratesIo { inherit name version; sha256 = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"; };
features = builtins.concatLists [
[ "std" ]
];
});
"registry+https://github.com/rust-lang/crates.io-index".mio."0.7.13" = overridableMkRustCrate (profileName: rec {
name = "mio";
version = "0.7.13";
@@ -1381,6 +1393,25 @@ in
};
});
"registry+https://github.com/rust-lang/crates.io-index".nom."7.1.0" = overridableMkRustCrate (profileName: rec {
name = "nom";
version = "7.1.0";
registry = "registry+https://github.com/rust-lang/crates.io-index";
src = fetchCratesIo { inherit name version; sha256 = "1b1d11e1ef389c76fe5b81bcaf2ea32cf88b62bc494e19f493d0b30e7a930109"; };
features = builtins.concatLists [
[ "alloc" ]
[ "default" ]
[ "std" ]
];
dependencies = {
memchr = rustPackages."registry+https://github.com/rust-lang/crates.io-index".memchr."2.4.1" { inherit profileName; };
minimal_lexical = rustPackages."registry+https://github.com/rust-lang/crates.io-index".minimal-lexical."0.2.1" { inherit profileName; };
};
buildDependencies = {
version_check = buildRustPackages."registry+https://github.com/rust-lang/crates.io-index".version_check."0.9.3" { profileName = "__noProfile"; };
};
});
"registry+https://github.com/rust-lang/crates.io-index".ntapi."0.3.6" = overridableMkRustCrate (profileName: rec {
name = "ntapi";
version = "0.3.6";
@@ -1490,6 +1521,28 @@ in
src = fetchCratesIo { inherit name version; sha256 = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"; };
});
"registry+https://github.com/rust-lang/crates.io-index".pin-project."1.0.8" = overridableMkRustCrate (profileName: rec {
name = "pin-project";
version = "1.0.8";
registry = "registry+https://github.com/rust-lang/crates.io-index";
src = fetchCratesIo { inherit name version; sha256 = "576bc800220cc65dac09e99e97b08b358cfab6e17078de8dc5fee223bd2d0c08"; };
dependencies = {
pin_project_internal = buildRustPackages."registry+https://github.com/rust-lang/crates.io-index".pin-project-internal."1.0.8" { profileName = "__noProfile"; };
};
});
"registry+https://github.com/rust-lang/crates.io-index".pin-project-internal."1.0.8" = overridableMkRustCrate (profileName: rec {
name = "pin-project-internal";
version = "1.0.8";
registry = "registry+https://github.com/rust-lang/crates.io-index";
src = fetchCratesIo { inherit name version; sha256 = "6e8fe8163d14ce7f0cdac2e040116f22eac817edabff0be91e8aff7e9accf389"; };
dependencies = {
proc_macro2 = rustPackages."registry+https://github.com/rust-lang/crates.io-index".proc-macro2."1.0.30" { inherit profileName; };
quote = rustPackages."registry+https://github.com/rust-lang/crates.io-index".quote."1.0.10" { inherit profileName; };
syn = rustPackages."registry+https://github.com/rust-lang/crates.io-index".syn."1.0.80" { inherit profileName; };
};
});
"registry+https://github.com/rust-lang/crates.io-index".pin-project-lite."0.2.7" = overridableMkRustCrate (profileName: rec {
name = "pin-project-lite";
version = "0.2.7";
+1 -1
View File
@@ -76,7 +76,7 @@ in let
*/
''^(src|tests)'' # fixed default
''.*\.(rs|toml)$'' # fixed default
''^(crdt|replication|cli|helper)'' # our crate submodules
''^(crdt|replication|cli|helper|signature)'' # our crate submodules
];
};
@@ -9,7 +9,8 @@ Implemented:
- putting and getting objects in buckets
- multipart uploads
- listing objects
- access control on a per-key-per-bucket basis
- access control on a per-access-key-per-bucket basis
- CORS headers on web endpoint
Not implemented:
@@ -31,9 +32,11 @@ All APIs that are not mentionned are not implemented and will return a 501 Not I
| CreateBucket | Implemented |
| CreateMultipartUpload | Implemented |
| DeleteBucket | Implemented |
| DeleteBucketCors | Implemented |
| DeleteBucketWebsite | Implemented |
| DeleteObject | Implemented |
| DeleteObjects | Implemented |
| GetBucketCors | Implemented |
| GetBucketLocation | Implemented |
| GetBucketVersioning | Stub (see below) |
| GetBucketWebsite | Implemented |
@@ -44,8 +47,9 @@ All APIs that are not mentionned are not implemented and will return a 501 Not I
| ListObjects | Implemented, bugs? (see below) |
| ListObjectsV2 | Implemented |
| ListMultipartUpload | Implemented |
| ListParts | Missing |
| ListParts | Implemented |
| PutObject | Implemented |
| PutBucketCors | Implemented |
| PutBucketWebsite | Partially implemented (see below)|
| UploadPart | Implemented |
| UploadPartCopy | Implemented |
@@ -26,10 +26,10 @@ your motivations for doing so in the PR message.
| | UploadPart |
| | [*ListMultipartUploads*](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/103) |
| | [*ListParts*](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/103) |
| **A-tier** (will implement) | |
| | [*GetBucketCors*](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/138) |
| | [*PutBucketCors*](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/138) |
| | [*DeleteBucketCors*](https://git.deuxfleurs.fr/Deuxfleurs/garage/issues/138) |
| **A-tier** | |
| | GetBucketCors |
| | PutBucketCors |
| | DeleteBucketCors |
| | UploadPartCopy |
| | GetBucketWebsite |
| | PutBucketWebsite |
+1 -1
View File
@@ -8,7 +8,7 @@ cat > /tmp/garage.mc/config.json <<EOF
"version": "10",
"aliases": {
"garage": {
"url": "https://localhost:4443",
"url": "http://127.0.0.1:3911",
"accessKey": "$ACCESS_KEY",
"secretKey": "$SECRET_KEY",
"api": "S3v4",
+53 -2
View File
@@ -137,7 +137,7 @@ if [ -z "$SKIP_AWS" ]; then
aws s3api list-objects-v2 --bucket eprouvette --page-size 0 >$CMDOUT
[ $(jq '.Contents | length' $CMDOUT) == 8 ]
[ $(jq '.CommonPrefixes | length' $CMDOUT) == 0 ]
aws s3api list-objects-v2 --bucket eprouvette --page-size 999999999999999 >$CMDOUT
aws s3api list-objects-v2 --bucket eprouvette --page-size 999999999 >$CMDOUT
[ $(jq '.Contents | length' $CMDOUT) == 8 ]
[ $(jq '.CommonPrefixes | length' $CMDOUT) == 0 ]
aws s3api list-objects-v2 --bucket eprouvette --page-size 1 >$CMDOUT
@@ -252,7 +252,58 @@ if [ -z "$SKIP_AWS" ]; then
echo "Deleted ${key}:${uid}"
done
# Test for UploadPartCopy
echo "Test for ListParts"
UPLOAD_ID=$(aws s3api create-multipart-upload --bucket eprouvette --key list-parts | jq -r .UploadId)
aws s3api list-parts --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID >$CMDOUT
[ $(jq '.Parts | length' $CMDOUT) == 0 ]
[ $(jq -r '.StorageClass' $CMDOUT) == 'STANDARD' ] # check that the result is not empty
ETAG1=$(aws s3api upload-part --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID --part-number 1 --body /tmp/garage.2.rnd | jq .ETag)
aws s3api list-parts --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID >$CMDOUT
[ $(jq '.Parts | length' $CMDOUT) == 1 ]
[ $(jq '.Parts[0].PartNumber' $CMDOUT) == 1 ]
[ $(jq '.Parts[0].Size' $CMDOUT) == 5242880 ]
[ $(jq '.Parts[0].ETag' $CMDOUT) == $ETAG1 ]
ETAG2=$(aws s3api upload-part --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID --part-number 3 --body /tmp/garage.3.rnd | jq .ETag)
ETAG3=$(aws s3api upload-part --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID --part-number 2 --body /tmp/garage.2.rnd | jq .ETag)
aws s3api list-parts --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID >$CMDOUT
[ $(jq '.Parts | length' $CMDOUT) == 3 ]
[ $(jq '.Parts[1].ETag' $CMDOUT) == $ETAG3 ]
aws s3api list-parts --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID --page-size 1 >$CMDOUT
[ $(jq '.Parts | length' $CMDOUT) == 3 ]
[ $(jq '.Parts[1].ETag' $CMDOUT) == $ETAG3 ]
cat >/tmp/garage.multipart_struct <<EOF
{
"Parts": [
{
"ETag": $ETAG1,
"PartNumber": 1
},
{
"ETag": $ETAG3,
"PartNumber": 2
},
{
"ETag": $ETAG2,
"PartNumber": 3
}
]
}
EOF
aws s3api complete-multipart-upload \
--bucket eprouvette --key list-parts --upload-id $UPLOAD_ID \
--multipart-upload file:///tmp/garage.multipart_struct
! aws s3api list-parts --bucket eprouvette --key list-parts --upload-id $UPLOAD_ID >$CMDOUT
aws s3 rm "s3://eprouvette/list-parts"
# @FIXME We do not write tests with --starting-token due to a bug with awscli
# See here: https://github.com/aws/aws-cli/issues/6666
echo "Test for UploadPartCopy"
aws s3 cp "/tmp/garage.3.rnd" "s3://eprouvette/copy_part_source"
UPLOAD_ID=$(aws s3api create-multipart-upload --bucket eprouvette --key test_multipart | jq -r .UploadId)
PART1=$(aws s3api upload-part \
+2
View File
@@ -28,10 +28,12 @@ hmac = "0.10"
idna = "0.2"
log = "0.4"
md-5 = "0.9"
nom = "7.1"
sha2 = "0.9"
futures = "0.3"
futures-util = "0.3"
pin-project = "1.0"
tokio = { version = "1.0", default-features = false, features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros", "sync", "signal", "fs"] }
http = "0.2"
+98 -52
View File
@@ -1,4 +1,3 @@
use std::cmp::{max, min};
use std::net::SocketAddr;
use std::sync::Arc;
@@ -6,7 +5,7 @@ use futures::future::Future;
use hyper::header;
use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Request, Response, Server};
use hyper::{Body, Method, Request, Response, Server};
use garage_util::data::*;
use garage_util::error::Error as GarageError;
@@ -14,12 +13,15 @@ use garage_util::error::Error as GarageError;
use garage_model::garage::Garage;
use garage_model::key_table::Key;
use garage_table::util::*;
use crate::error::*;
use crate::signature::check_signature;
use crate::signature::payload::check_payload_signature;
use crate::helpers::*;
use crate::s3_bucket::*;
use crate::s3_copy::*;
use crate::s3_cors::*;
use crate::s3_delete::*;
use crate::s3_get::*;
use crate::s3_list::*;
@@ -90,7 +92,10 @@ async fn handler(
}
async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Response<Body>, Error> {
let (api_key, content_sha256) = check_signature(&garage, &req).await?;
let (api_key, content_sha256) = check_payload_signature(&garage, &req).await?;
let api_key = api_key.ok_or_else(|| {
Error::Forbidden("Garage does not support anonymous access yet".to_string())
})?;
let authority = req
.headers()
@@ -100,32 +105,38 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
let host = authority_to_host(authority)?;
let bucket = garage
let bucket_name = garage
.config
.s3_api
.root_domain
.as_ref()
.and_then(|root_domain| host_to_bucket(&host, root_domain));
let endpoint = Endpoint::from_request(&req, bucket.map(ToOwned::to_owned))?;
let (endpoint, bucket_name) = Endpoint::from_request(&req, bucket_name.map(ToOwned::to_owned))?;
debug!("Endpoint: {:?}", endpoint);
// Special code path for CreateBucket API endpoint
if let Endpoint::CreateBucket { bucket } = endpoint {
return handle_create_bucket(&garage, req, content_sha256, api_key, bucket).await;
}
let bucket_name = match endpoint.get_bucket() {
let bucket_name = match bucket_name {
None => return handle_request_without_bucket(garage, req, api_key, endpoint).await,
Some(bucket) => bucket.to_string(),
};
// Special code path for CreateBucket API endpoint
if let Endpoint::CreateBucket {} = endpoint {
return handle_create_bucket(&garage, req, content_sha256, api_key, bucket_name).await;
}
let bucket_id = resolve_bucket(&garage, &bucket_name, &api_key).await?;
let bucket = garage
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.filter(|b| !b.state.is_deleted())
.ok_or(Error::NoSuchBucket)?;
let allowed = match endpoint.authorization_type() {
Authorization::Read(_) => api_key.allow_read(&bucket_id),
Authorization::Write(_) => api_key.allow_write(&bucket_id),
Authorization::Owner(_) => api_key.allow_owner(&bucket_id),
Authorization::Read => api_key.allow_read(&bucket_id),
Authorization::Write => api_key.allow_write(&bucket_id),
Authorization::Owner => api_key.allow_owner(&bucket_id),
_ => unreachable!(),
};
@@ -135,14 +146,27 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
));
}
match endpoint {
Endpoint::HeadObject { key, .. } => handle_head(garage, &req, bucket_id, &key).await,
Endpoint::GetObject { key, .. } => handle_get(garage, &req, bucket_id, &key).await,
// Look up what CORS rule might apply to response.
// Requests for methods different than GET, HEAD or POST
// are always preflighted, i.e. the browser should make
// an OPTIONS call before to check it is allowed
let matching_cors_rule = match *req.method() {
Method::GET | Method::HEAD | Method::POST => find_matching_cors_rule(&bucket, &req)?,
_ => None,
};
let resp = match endpoint {
Endpoint::Options => handle_options(&req, &bucket).await,
Endpoint::HeadObject {
key, part_number, ..
} => handle_head(garage, &req, bucket_id, &key, part_number).await,
Endpoint::GetObject {
key, part_number, ..
} => handle_get(garage, &req, bucket_id, &key, part_number).await,
Endpoint::UploadPart {
key,
part_number,
upload_id,
..
} => {
handle_put_part(
garage,
@@ -155,14 +179,11 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
)
.await
}
Endpoint::CopyObject { key, .. } => {
handle_copy(garage, &api_key, &req, bucket_id, &key).await
}
Endpoint::CopyObject { key } => handle_copy(garage, &api_key, &req, bucket_id, &key).await,
Endpoint::UploadPartCopy {
key,
part_number,
upload_id,
..
} => {
handle_upload_part_copy(
garage,
@@ -175,25 +196,21 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
)
.await
}
Endpoint::PutObject { key, .. } => {
handle_put(garage, req, bucket_id, &key, content_sha256).await
Endpoint::PutObject { key } => {
handle_put(garage, req, bucket_id, &key, &api_key, content_sha256).await
}
Endpoint::AbortMultipartUpload { key, upload_id, .. } => {
Endpoint::AbortMultipartUpload { key, upload_id } => {
handle_abort_multipart_upload(garage, bucket_id, &key, &upload_id).await
}
Endpoint::DeleteObject { key, .. } => handle_delete(garage, bucket_id, &key).await,
Endpoint::CreateMultipartUpload { bucket, key } => {
handle_create_multipart_upload(garage, &req, &bucket, bucket_id, &key).await
Endpoint::CreateMultipartUpload { key } => {
handle_create_multipart_upload(garage, &req, &bucket_name, bucket_id, &key).await
}
Endpoint::CompleteMultipartUpload {
bucket,
key,
upload_id,
} => {
Endpoint::CompleteMultipartUpload { key, upload_id } => {
handle_complete_multipart_upload(
garage,
req,
&bucket,
&bucket_name,
bucket_id,
&key,
&upload_id,
@@ -201,19 +218,18 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
)
.await
}
Endpoint::CreateBucket { .. } => unreachable!(),
Endpoint::HeadBucket { .. } => {
Endpoint::CreateBucket {} => unreachable!(),
Endpoint::HeadBucket {} => {
let empty_body: Body = Body::from(vec![]);
let response = Response::builder().body(empty_body).unwrap();
Ok(response)
}
Endpoint::DeleteBucket { .. } => {
Endpoint::DeleteBucket {} => {
handle_delete_bucket(&garage, bucket_id, bucket_name, api_key).await
}
Endpoint::GetBucketLocation { .. } => handle_get_bucket_location(garage),
Endpoint::GetBucketVersioning { .. } => handle_get_bucket_versioning(),
Endpoint::GetBucketLocation {} => handle_get_bucket_location(garage),
Endpoint::GetBucketVersioning {} => handle_get_bucket_versioning(),
Endpoint::ListObjects {
bucket,
delimiter,
encoding_type,
marker,
@@ -224,10 +240,10 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
garage,
&ListObjectsQuery {
common: ListQueryCommon {
bucket_name: bucket,
bucket_name,
bucket_id,
delimiter: delimiter.map(|d| d.to_string()),
page_size: max_keys.map(|p| min(1000, max(1, p))).unwrap_or(1000),
page_size: max_keys.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
prefix: prefix.unwrap_or_default(),
urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
},
@@ -240,7 +256,6 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
.await
}
Endpoint::ListObjectsV2 {
bucket,
delimiter,
encoding_type,
max_keys,
@@ -255,10 +270,10 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
garage,
&ListObjectsQuery {
common: ListQueryCommon {
bucket_name: bucket,
bucket_name,
bucket_id,
delimiter: delimiter.map(|d| d.to_string()),
page_size: max_keys.map(|p| min(1000, max(1, p))).unwrap_or(1000),
page_size: max_keys.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
prefix: prefix.unwrap_or_default(),
},
@@ -277,7 +292,6 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
}
}
Endpoint::ListMultipartUploads {
bucket,
delimiter,
encoding_type,
key_marker,
@@ -289,10 +303,10 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
garage,
&ListMultipartUploadsQuery {
common: ListQueryCommon {
bucket_name: bucket,
bucket_name,
bucket_id,
delimiter: delimiter.map(|d| d.to_string()),
page_size: max_uploads.map(|p| min(1000, max(1, p))).unwrap_or(1000),
page_size: max_uploads.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
prefix: prefix.unwrap_or_default(),
urlencode_resp: encoding_type.map(|e| e == "url").unwrap_or(false),
},
@@ -302,16 +316,48 @@ async fn handler_inner(garage: Arc<Garage>, req: Request<Body>) -> Result<Respon
)
.await
}
Endpoint::DeleteObjects { .. } => {
Endpoint::ListParts {
key,
max_parts,
part_number_marker,
upload_id,
} => {
handle_list_parts(
garage,
&ListPartsQuery {
bucket_name,
bucket_id,
key,
upload_id,
part_number_marker: part_number_marker.map(|p| p.clamp(1, 10000)),
max_parts: max_parts.map(|p| p.clamp(1, 1000)).unwrap_or(1000),
},
)
.await
}
Endpoint::DeleteObjects {} => {
handle_delete_objects(garage, bucket_id, req, content_sha256).await
}
Endpoint::GetBucketWebsite { .. } => handle_get_website(garage, bucket_id).await,
Endpoint::PutBucketWebsite { .. } => {
Endpoint::GetBucketWebsite {} => handle_get_website(&bucket).await,
Endpoint::PutBucketWebsite {} => {
handle_put_website(garage, bucket_id, req, content_sha256).await
}
Endpoint::DeleteBucketWebsite { .. } => handle_delete_website(garage, bucket_id).await,
Endpoint::DeleteBucketWebsite {} => handle_delete_website(garage, bucket_id).await,
Endpoint::GetBucketCors {} => handle_get_cors(&bucket).await,
Endpoint::PutBucketCors {} => handle_put_cors(garage, bucket_id, req, content_sha256).await,
Endpoint::DeleteBucketCors {} => handle_delete_cors(garage, bucket_id).await,
endpoint => Err(Error::NotImplemented(endpoint.name().to_owned())),
};
// If request was a success and we have a CORS rule that applies to it,
// add the corresponding CORS headers to the response
let mut resp_ok = resp?;
if let Some(rule) = matching_cors_rule {
add_cors_headers(&mut resp_ok, rule)
.ok_or_internal_error("Invalid bucket CORS configuration")?;
}
Ok(resp_ok)
}
async fn handle_request_without_bucket(
+17 -5
View File
@@ -58,6 +58,19 @@ pub enum Error {
#[error(display = "At least one of the preconditions you specified did not hold")]
PreconditionFailed,
/// Parts specified in CMU request do not match parts actually uploaded
#[error(display = "Parts given to CompleteMultipartUpload do not match uploaded parts")]
InvalidPart,
/// Parts given to CompleteMultipartUpload were not in ascending order
#[error(display = "Parts given to CompleteMultipartUpload were not in ascending order")]
InvalidPartOrder,
/// In CompleteMultipartUpload: not enough data
/// (here we are more lenient than AWS S3)
#[error(display = "Proposed upload is smaller than the minimum allowed object size")]
EntityTooSmall,
// Category: bad request
/// The request contained an invalid UTF-8 sequence in its path or in other parameters
#[error(display = "Invalid UTF-8: {}", _0)]
@@ -143,6 +156,9 @@ impl Error {
Error::BucketAlreadyExists => "BucketAlreadyExists",
Error::BucketNotEmpty => "BucketNotEmpty",
Error::PreconditionFailed => "PreconditionFailed",
Error::InvalidPart => "InvalidPart",
Error::InvalidPartOrder => "InvalidPartOrder",
Error::EntityTooSmall => "EntityTooSmall",
Error::Forbidden(_) => "AccessDenied",
Error::AuthorizationHeaderMalformed(_) => "AuthorizationHeaderMalformed",
Error::NotImplemented(_) => "NotImplemented",
@@ -206,11 +222,7 @@ where
fn ok_or_bad_request<M: AsRef<str>>(self, reason: M) -> Result<T, Error> {
match self {
Ok(x) => Ok(x),
Err(e) => Err(Error::BadRequest(format!(
"{}: {}",
reason.as_ref(),
e.to_string()
))),
Err(e) => Err(Error::BadRequest(format!("{}: {}", reason.as_ref(), e))),
}
}
}
+1
View File
@@ -15,6 +15,7 @@ mod signature;
pub mod helpers;
mod s3_bucket;
mod s3_copy;
pub mod s3_cors;
mod s3_delete;
pub mod s3_get;
mod s3_list;
+8 -5
View File
@@ -120,7 +120,10 @@ pub async fn handle_create_bucket(
bucket_name: String,
) -> Result<Response<Body>, Error> {
let body = hyper::body::to_bytes(req.into_body()).await?;
verify_signed_content(content_sha256, &body[..])?;
if let Some(content_sha256) = content_sha256 {
verify_signed_content(content_sha256, &body[..])?;
}
let cmd =
parse_create_bucket_xml(&body[..]).ok_or_bad_request("Invalid create bucket XML query")?;
@@ -320,7 +323,7 @@ mod tests {
assert_eq!(
parse_create_bucket_xml(
br#"
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
</CreateBucketConfiguration >
"#
),
@@ -329,8 +332,8 @@ mod tests {
assert_eq!(
parse_create_bucket_xml(
br#"
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<LocationConstraint>Europe</LocationConstraint>
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<LocationConstraint>Europe</LocationConstraint>
</CreateBucketConfiguration >
"#
),
@@ -339,7 +342,7 @@ mod tests {
assert_eq!(
parse_create_bucket_xml(
br#"
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<CreateBucketConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
</Crea >
"#
),
+2 -2
View File
@@ -487,7 +487,7 @@ impl CopyPreconditionHeaders {
.get("x-amz-copy-source-if-modified-since")
.map(|x| x.to_str())
.transpose()?
.map(|x| httpdate::parse_http_date(x))
.map(httpdate::parse_http_date)
.transpose()
.ok_or_bad_request("Invalid date in x-amz-copy-source-if-modified-since")?,
copy_source_if_none_match: req
@@ -505,7 +505,7 @@ impl CopyPreconditionHeaders {
.get("x-amz-copy-source-if-unmodified-since")
.map(|x| x.to_str())
.transpose()?
.map(|x| httpdate::parse_http_date(x))
.map(httpdate::parse_http_date)
.transpose()
.ok_or_bad_request("Invalid date in x-amz-copy-source-if-unmodified-since")?,
})
+391
View File
@@ -0,0 +1,391 @@
use quick_xml::de::from_reader;
use std::sync::Arc;
use http::header::{
ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN,
ACCESS_CONTROL_EXPOSE_HEADERS, ACCESS_CONTROL_REQUEST_HEADERS, ACCESS_CONTROL_REQUEST_METHOD,
};
use hyper::{header::HeaderName, Body, Method, Request, Response, StatusCode};
use serde::{Deserialize, Serialize};
use crate::error::*;
use crate::s3_xml::{to_xml_with_header, xmlns_tag, IntValue, Value};
use crate::signature::verify_signed_content;
use garage_model::bucket_table::{Bucket, CorsRule as GarageCorsRule};
use garage_model::garage::Garage;
use garage_table::*;
use garage_util::data::*;
pub async fn handle_get_cors(bucket: &Bucket) -> Result<Response<Body>, Error> {
let param = bucket
.params()
.ok_or_internal_error("Bucket should not be deleted at this point")?;
if let Some(cors) = param.cors_config.get() {
let wc = CorsConfiguration {
xmlns: (),
cors_rules: cors
.iter()
.map(CorsRule::from_garage_cors_rule)
.collect::<Vec<_>>(),
};
let xml = to_xml_with_header(&wc)?;
Ok(Response::builder()
.status(StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/xml")
.body(Body::from(xml))?)
} else {
Ok(Response::builder()
.status(StatusCode::NO_CONTENT)
.body(Body::empty())?)
}
}
pub async fn handle_delete_cors(
garage: Arc<Garage>,
bucket_id: Uuid,
) -> Result<Response<Body>, Error> {
let mut bucket = garage
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.ok_or(Error::NoSuchBucket)?;
let param = bucket
.params_mut()
.ok_or_internal_error("Bucket should not be deleted at this point")?;
param.cors_config.update(None);
garage.bucket_table.insert(&bucket).await?;
Ok(Response::builder()
.status(StatusCode::NO_CONTENT)
.body(Body::empty())?)
}
pub async fn handle_put_cors(
garage: Arc<Garage>,
bucket_id: Uuid,
req: Request<Body>,
content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> {
let body = hyper::body::to_bytes(req.into_body()).await?;
if let Some(content_sha256) = content_sha256 {
verify_signed_content(content_sha256, &body[..])?;
}
let mut bucket = garage
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.ok_or(Error::NoSuchBucket)?;
let param = bucket
.params_mut()
.ok_or_internal_error("Bucket should not be deleted at this point")?;
let conf: CorsConfiguration = from_reader(&body as &[u8])?;
conf.validate()?;
param
.cors_config
.update(Some(conf.into_garage_cors_config()?));
garage.bucket_table.insert(&bucket).await?;
Ok(Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?)
}
pub async fn handle_options(req: &Request<Body>, bucket: &Bucket) -> Result<Response<Body>, Error> {
let origin = req
.headers()
.get("Origin")
.ok_or_bad_request("Missing Origin header")?
.to_str()?;
let request_method = req
.headers()
.get(ACCESS_CONTROL_REQUEST_METHOD)
.ok_or_bad_request("Missing Access-Control-Request-Method header")?
.to_str()?;
let request_headers = match req.headers().get(ACCESS_CONTROL_REQUEST_HEADERS) {
Some(h) => h.to_str()?.split(',').map(|h| h.trim()).collect::<Vec<_>>(),
None => vec![],
};
if let Some(cors_config) = bucket.params().unwrap().cors_config.get() {
let matching_rule = cors_config
.iter()
.find(|rule| cors_rule_matches(rule, origin, request_method, request_headers.iter()));
if let Some(rule) = matching_rule {
let mut resp = Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?;
add_cors_headers(&mut resp, rule).ok_or_internal_error("Invalid CORS configuration")?;
return Ok(resp);
}
}
Err(Error::Forbidden("This CORS request is not allowed.".into()))
}
pub fn find_matching_cors_rule<'a>(
bucket: &'a Bucket,
req: &Request<Body>,
) -> Result<Option<&'a GarageCorsRule>, Error> {
if let Some(cors_config) = bucket.params().unwrap().cors_config.get() {
if let Some(origin) = req.headers().get("Origin") {
let origin = origin.to_str()?;
let request_headers = match req.headers().get(ACCESS_CONTROL_REQUEST_HEADERS) {
Some(h) => h.to_str()?.split(',').map(|h| h.trim()).collect::<Vec<_>>(),
None => vec![],
};
return Ok(cors_config.iter().find(|rule| {
cors_rule_matches(
rule,
origin,
&req.method().to_string(),
request_headers.iter(),
)
}));
}
}
Ok(None)
}
fn cors_rule_matches<'a, HI, S>(
rule: &GarageCorsRule,
origin: &'a str,
method: &'a str,
mut request_headers: HI,
) -> bool
where
HI: Iterator<Item = S>,
S: AsRef<str>,
{
rule.allow_origins.iter().any(|x| x == "*" || x == origin)
&& rule.allow_methods.iter().any(|x| x == "*" || x == method)
&& request_headers.all(|h| {
rule.allow_headers
.iter()
.any(|x| x == "*" || x == h.as_ref())
})
}
pub fn add_cors_headers(
resp: &mut Response<Body>,
rule: &GarageCorsRule,
) -> Result<(), http::header::InvalidHeaderValue> {
let h = resp.headers_mut();
h.insert(
ACCESS_CONTROL_ALLOW_ORIGIN,
rule.allow_origins.join(", ").parse()?,
);
h.insert(
ACCESS_CONTROL_ALLOW_METHODS,
rule.allow_methods.join(", ").parse()?,
);
h.insert(
ACCESS_CONTROL_ALLOW_HEADERS,
rule.allow_headers.join(", ").parse()?,
);
h.insert(
ACCESS_CONTROL_EXPOSE_HEADERS,
rule.expose_headers.join(", ").parse()?,
);
Ok(())
}
// ---- SERIALIZATION AND DESERIALIZATION TO/FROM S3 XML ----
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename = "CORSConfiguration")]
pub struct CorsConfiguration {
#[serde(serialize_with = "xmlns_tag", skip_deserializing)]
pub xmlns: (),
#[serde(rename = "CORSRule")]
pub cors_rules: Vec<CorsRule>,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct CorsRule {
#[serde(rename = "ID")]
pub id: Option<Value>,
#[serde(rename = "MaxAgeSeconds")]
pub max_age_seconds: Option<IntValue>,
#[serde(rename = "AllowedOrigin")]
pub allowed_origins: Vec<Value>,
#[serde(rename = "AllowedMethod")]
pub allowed_methods: Vec<Value>,
#[serde(rename = "AllowedHeader", default)]
pub allowed_headers: Vec<Value>,
#[serde(rename = "ExposeHeader", default)]
pub expose_headers: Vec<Value>,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct AllowedMethod {
#[serde(rename = "AllowedMethod")]
pub allowed_method: Value,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct AllowedHeader {
#[serde(rename = "AllowedHeader")]
pub allowed_header: Value,
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct ExposeHeader {
#[serde(rename = "ExposeHeader")]
pub expose_header: Value,
}
impl CorsConfiguration {
pub fn validate(&self) -> Result<(), Error> {
for r in self.cors_rules.iter() {
r.validate()?;
}
Ok(())
}
pub fn into_garage_cors_config(self) -> Result<Vec<GarageCorsRule>, Error> {
Ok(self
.cors_rules
.iter()
.map(CorsRule::to_garage_cors_rule)
.collect())
}
}
impl CorsRule {
pub fn validate(&self) -> Result<(), Error> {
for method in self.allowed_methods.iter() {
method
.0
.parse::<Method>()
.ok_or_bad_request("Invalid CORSRule method")?;
}
for header in self
.allowed_headers
.iter()
.chain(self.expose_headers.iter())
{
header
.0
.parse::<HeaderName>()
.ok_or_bad_request("Invalid HTTP header name")?;
}
Ok(())
}
pub fn to_garage_cors_rule(&self) -> GarageCorsRule {
let convert_vec =
|vval: &[Value]| vval.iter().map(|x| x.0.to_owned()).collect::<Vec<String>>();
GarageCorsRule {
id: self.id.as_ref().map(|x| x.0.to_owned()),
max_age_seconds: self.max_age_seconds.as_ref().map(|x| x.0 as u64),
allow_origins: convert_vec(&self.allowed_origins),
allow_methods: convert_vec(&self.allowed_methods),
allow_headers: convert_vec(&self.allowed_headers),
expose_headers: convert_vec(&self.expose_headers),
}
}
pub fn from_garage_cors_rule(rule: &GarageCorsRule) -> Self {
let convert_vec = |vval: &[String]| {
vval.iter()
.map(|x| Value(x.clone()))
.collect::<Vec<Value>>()
};
Self {
id: rule.id.as_ref().map(|x| Value(x.clone())),
max_age_seconds: rule.max_age_seconds.map(|x| IntValue(x as i64)),
allowed_origins: convert_vec(&rule.allow_origins),
allowed_methods: convert_vec(&rule.allow_methods),
allowed_headers: convert_vec(&rule.allow_headers),
expose_headers: convert_vec(&rule.expose_headers),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use quick_xml::de::from_str;
#[test]
fn test_deserialize() -> Result<(), Error> {
let message = r#"<?xml version="1.0" encoding="UTF-8"?>
<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<CORSRule>
<AllowedOrigin>http://www.example.com</AllowedOrigin>
<AllowedMethod>PUT</AllowedMethod>
<AllowedMethod>POST</AllowedMethod>
<AllowedMethod>DELETE</AllowedMethod>
<AllowedHeader>*</AllowedHeader>
</CORSRule>
<CORSRule>
<AllowedOrigin>*</AllowedOrigin>
<AllowedMethod>GET</AllowedMethod>
</CORSRule>
<CORSRule>
<ID>qsdfjklm</ID>
<MaxAgeSeconds>12345</MaxAgeSeconds>
<AllowedOrigin>https://perdu.com</AllowedOrigin>
<AllowedMethod>GET</AllowedMethod>
<AllowedMethod>DELETE</AllowedMethod>
<AllowedHeader>*</AllowedHeader>
<ExposeHeader>*</ExposeHeader>
</CORSRule>
</CORSConfiguration>"#;
let conf: CorsConfiguration = from_str(message).unwrap();
let ref_value = CorsConfiguration {
xmlns: (),
cors_rules: vec![
CorsRule {
id: None,
max_age_seconds: None,
allowed_origins: vec!["http://www.example.com".into()],
allowed_methods: vec!["PUT".into(), "POST".into(), "DELETE".into()],
allowed_headers: vec!["*".into()],
expose_headers: vec![],
},
CorsRule {
id: None,
max_age_seconds: None,
allowed_origins: vec!["*".into()],
allowed_methods: vec!["GET".into()],
allowed_headers: vec![],
expose_headers: vec![],
},
CorsRule {
id: Some("qsdfjklm".into()),
max_age_seconds: Some(IntValue(12345)),
allowed_origins: vec!["https://perdu.com".into()],
allowed_methods: vec!["GET".into(), "DELETE".into()],
allowed_headers: vec!["*".into()],
expose_headers: vec!["*".into()],
},
],
};
assert_eq! {
ref_value,
conf
};
let message2 = to_xml_with_header(&ref_value)?;
let cleanup = |c: &str| c.replace(char::is_whitespace, "");
assert_eq!(cleanup(message), cleanup(&message2));
Ok(())
}
}
+4 -1
View File
@@ -81,7 +81,10 @@ pub async fn handle_delete_objects(
content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> {
let body = hyper::body::to_bytes(req.into_body()).await?;
verify_signed_content(content_sha256, &body[..])?;
if let Some(content_sha256) = content_sha256 {
verify_signed_content(content_sha256, &body[..])?;
}
let cmd_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
let cmd = parse_delete_objects_xml(&cmd_xml).ok_or_bad_request("Invalid delete XML query")?;
+244 -98
View File
@@ -3,6 +3,10 @@ use std::sync::Arc;
use std::time::{Duration, UNIX_EPOCH};
use futures::stream::*;
use http::header::{
ACCEPT_RANGES, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE,
IF_NONE_MATCH, LAST_MODIFIED, RANGE,
};
use hyper::body::Bytes;
use hyper::{Body, Request, Response, StatusCode};
@@ -11,9 +15,12 @@ use garage_util::data::*;
use garage_model::garage::Garage;
use garage_model::object_table::*;
use garage_model::version_table::*;
use crate::error::*;
const X_AMZ_MP_PARTS_COUNT: &str = "x-amz-mp-parts-count";
fn object_headers(
version: &ObjectVersion,
version_meta: &ObjectVersionMeta,
@@ -24,15 +31,12 @@ fn object_headers(
let date_str = httpdate::fmt_http_date(date);
let mut resp = Response::builder()
.header(
"Content-Type",
version_meta.headers.content_type.to_string(),
)
.header("Last-Modified", date_str)
.header("Accept-Ranges", "bytes".to_string());
.header(CONTENT_TYPE, version_meta.headers.content_type.to_string())
.header(LAST_MODIFIED, date_str)
.header(ACCEPT_RANGES, "bytes".to_string());
if !version_meta.etag.is_empty() {
resp = resp.header("ETag", format!("\"{}\"", version_meta.etag));
resp = resp.header(ETAG, format!("\"{}\"", version_meta.etag));
}
for (k, v) in version_meta.headers.other.iter() {
@@ -52,7 +56,7 @@ fn try_answer_cached(
// precedence and If-Modified-Since is ignored (as per 6.Precedence from rfc7232). The rational
// being that etag based matching is more accurate, it has no issue with sub-second precision
// for instance (in case of very fast updates)
let cached = if let Some(none_match) = req.headers().get(http::header::IF_NONE_MATCH) {
let cached = if let Some(none_match) = req.headers().get(IF_NONE_MATCH) {
let none_match = none_match.to_str().ok()?;
let expected = format!("\"{}\"", version_meta.etag);
let found = none_match
@@ -60,7 +64,7 @@ fn try_answer_cached(
.map(str::trim)
.any(|etag| etag == expected || etag == "\"*\"");
found
} else if let Some(modified_since) = req.headers().get(http::header::IF_MODIFIED_SINCE) {
} else if let Some(modified_since) = req.headers().get(IF_MODIFIED_SINCE) {
let modified_since = modified_since.to_str().ok()?;
let client_date = httpdate::parse_http_date(modified_since).ok()?;
let server_date = UNIX_EPOCH + Duration::from_millis(version.timestamp);
@@ -87,6 +91,7 @@ pub async fn handle_head(
req: &Request<Body>,
bucket_id: Uuid,
key: &str,
part_number: Option<u64>,
) -> Result<Response<Body>, Error> {
let object = garage
.object_table
@@ -94,30 +99,78 @@ pub async fn handle_head(
.await?
.ok_or(Error::NoSuchKey)?;
let version = object
let object_version = object
.versions()
.iter()
.rev()
.find(|v| v.is_data())
.ok_or(Error::NoSuchKey)?;
let version_meta = match &version.state {
ObjectVersionState::Complete(ObjectVersionData::Inline(meta, _)) => meta,
ObjectVersionState::Complete(ObjectVersionData::FirstBlock(meta, _)) => meta,
let version_data = match &object_version.state {
ObjectVersionState::Complete(c) => c,
_ => unreachable!(),
};
if let Some(cached) = try_answer_cached(version, version_meta, req) {
let version_meta = match version_data {
ObjectVersionData::Inline(meta, _) => meta,
ObjectVersionData::FirstBlock(meta, _) => meta,
_ => unreachable!(),
};
if let Some(cached) = try_answer_cached(object_version, version_meta, req) {
return Ok(cached);
}
let body: Body = Body::empty();
let response = object_headers(version, version_meta)
.header("Content-Length", format!("{}", version_meta.size))
.status(StatusCode::OK)
.body(body)
.unwrap();
Ok(response)
if let Some(pn) = part_number {
match version_data {
ObjectVersionData::Inline(_, bytes) => {
if pn != 1 {
return Err(Error::InvalidPart);
}
Ok(object_headers(object_version, version_meta)
.header(CONTENT_LENGTH, format!("{}", bytes.len()))
.header(
CONTENT_RANGE,
format!("bytes 0-{}/{}", bytes.len() - 1, bytes.len()),
)
.header(X_AMZ_MP_PARTS_COUNT, "1")
.status(StatusCode::PARTIAL_CONTENT)
.body(Body::empty())?)
}
ObjectVersionData::FirstBlock(_, _) => {
let version = garage
.version_table
.get(&object_version.uuid, &EmptyKey)
.await?
.ok_or(Error::NoSuchKey)?;
let (part_offset, part_end) =
calculate_part_bounds(&version, pn).ok_or(Error::InvalidPart)?;
let n_parts = version.parts_etags.items().len();
Ok(object_headers(object_version, version_meta)
.header(CONTENT_LENGTH, format!("{}", part_end - part_offset))
.header(
CONTENT_RANGE,
format!(
"bytes {}-{}/{}",
part_offset,
part_end - 1,
version_meta.size
),
)
.header(X_AMZ_MP_PARTS_COUNT, format!("{}", n_parts))
.status(StatusCode::PARTIAL_CONTENT)
.body(Body::empty())?)
}
_ => unreachable!(),
}
} else {
Ok(object_headers(object_version, version_meta)
.header(CONTENT_LENGTH, format!("{}", version_meta.size))
.status(StatusCode::OK)
.body(Body::empty())?)
}
}
/// Handle GET request
@@ -126,6 +179,7 @@ pub async fn handle_get(
req: &Request<Body>,
bucket_id: Uuid,
key: &str,
part_number: Option<u64>,
) -> Result<Response<Body>, Error> {
let object = garage
.object_table
@@ -154,35 +208,31 @@ pub async fn handle_get(
return Ok(cached);
}
let range = match req.headers().get("range") {
Some(range) => {
let range_str = range.to_str()?;
let mut ranges = http_range::HttpRange::parse(range_str, last_v_meta.size)
.map_err(|e| (e, last_v_meta.size))?;
if ranges.len() > 1 {
// garage does not support multi-range requests yet, so we respond with the entire
// object when multiple ranges are requested
None
} else {
ranges.pop()
}
match (part_number, parse_range_header(req, last_v_meta.size)?) {
(Some(_), Some(_)) => {
return Err(Error::BadRequest(
"Cannot specify both partNumber and Range header".into(),
));
}
None => None,
};
if let Some(range) = range {
return handle_get_range(
garage,
last_v,
last_v_data,
last_v_meta,
range.start,
range.start + range.length,
)
.await;
(Some(pn), None) => {
return handle_get_part(garage, last_v, last_v_data, last_v_meta, pn).await;
}
(None, Some(range)) => {
return handle_get_range(
garage,
last_v,
last_v_data,
last_v_meta,
range.start,
range.start + range.length,
)
.await;
}
(None, None) => (),
}
let resp_builder = object_headers(last_v, last_v_meta)
.header("Content-Length", format!("{}", last_v_meta.size))
.header(CONTENT_LENGTH, format!("{}", last_v_meta.size))
.status(StatusCode::OK);
match &last_v_data {
@@ -238,9 +288,9 @@ async fn handle_get_range(
end: u64,
) -> Result<Response<Body>, Error> {
let resp_builder = object_headers(version, version_meta)
.header("Content-Length", format!("{}", end - begin))
.header(CONTENT_LENGTH, format!("{}", end - begin))
.header(
"Content-Range",
CONTENT_RANGE,
format!("bytes {}-{}/{}", begin, end - 1, version_meta.size),
)
.status(StatusCode::PARTIAL_CONTENT);
@@ -258,58 +308,154 @@ async fn handle_get_range(
}
}
ObjectVersionData::FirstBlock(_meta, _first_block_hash) => {
let version = garage.version_table.get(&version.uuid, &EmptyKey).await?;
let version = match version {
Some(v) => v,
None => return Err(Error::NoSuchKey),
};
let version = garage
.version_table
.get(&version.uuid, &EmptyKey)
.await?
.ok_or(Error::NoSuchKey)?;
// We will store here the list of blocks that have an intersection with the requested
// range, as well as their "true offset", which is their actual offset in the complete
// file (whereas block.offset designates the offset of the block WITHIN THE PART
// block.part_number, which is not the same in the case of a multipart upload)
let mut blocks = Vec::with_capacity(std::cmp::min(
version.blocks.len(),
4 + ((end - begin) / std::cmp::max(version.blocks.items()[0].1.size as u64, 1024))
as usize,
));
let mut true_offset = 0;
for (_, b) in version.blocks.items().iter() {
if true_offset >= end {
break;
}
// Keep only blocks that have an intersection with the requested range
if true_offset < end && true_offset + b.size > begin {
blocks.push((*b, true_offset));
}
true_offset += b.size;
}
let body_stream = futures::stream::iter(blocks)
.map(move |(block, true_offset)| {
let garage = garage.clone();
async move {
let data = garage.block_manager.rpc_get_block(&block.hash).await?;
let data = Bytes::from(data);
let start_in_block = if true_offset > begin {
0
} else {
begin - true_offset
};
let end_in_block = if true_offset + block.size < end {
block.size
} else {
end - true_offset
};
Result::<Bytes, Error>::Ok(
data.slice(start_in_block as usize..end_in_block as usize),
)
}
})
.buffered(2);
let body = hyper::body::Body::wrap_stream(body_stream);
let body = body_from_blocks_range(garage, version.blocks.items(), begin, end);
Ok(resp_builder.body(body)?)
}
}
}
async fn handle_get_part(
garage: Arc<Garage>,
object_version: &ObjectVersion,
version_data: &ObjectVersionData,
version_meta: &ObjectVersionMeta,
part_number: u64,
) -> Result<Response<Body>, Error> {
let resp_builder =
object_headers(object_version, version_meta).status(StatusCode::PARTIAL_CONTENT);
match version_data {
ObjectVersionData::Inline(_, bytes) => {
if part_number != 1 {
return Err(Error::InvalidPart);
}
Ok(resp_builder
.header(CONTENT_LENGTH, format!("{}", bytes.len()))
.header(
CONTENT_RANGE,
format!("bytes {}-{}/{}", 0, bytes.len() - 1, bytes.len()),
)
.header(X_AMZ_MP_PARTS_COUNT, "1")
.body(Body::from(bytes.to_vec()))?)
}
ObjectVersionData::FirstBlock(_, _) => {
let version = garage
.version_table
.get(&object_version.uuid, &EmptyKey)
.await?
.ok_or(Error::NoSuchKey)?;
let (begin, end) =
calculate_part_bounds(&version, part_number).ok_or(Error::InvalidPart)?;
let n_parts = version.parts_etags.items().len();
let body = body_from_blocks_range(garage, version.blocks.items(), begin, end);
Ok(resp_builder
.header(CONTENT_LENGTH, format!("{}", end - begin))
.header(
CONTENT_RANGE,
format!("bytes {}-{}/{}", begin, end - 1, version_meta.size),
)
.header(X_AMZ_MP_PARTS_COUNT, format!("{}", n_parts))
.body(body)?)
}
_ => unreachable!(),
}
}
fn parse_range_header(
req: &Request<Body>,
total_size: u64,
) -> Result<Option<http_range::HttpRange>, Error> {
let range = match req.headers().get(RANGE) {
Some(range) => {
let range_str = range.to_str()?;
let mut ranges =
http_range::HttpRange::parse(range_str, total_size).map_err(|e| (e, total_size))?;
if ranges.len() > 1 {
// garage does not support multi-range requests yet, so we respond with the entire
// object when multiple ranges are requested
None
} else {
ranges.pop()
}
}
None => None,
};
Ok(range)
}
fn calculate_part_bounds(v: &Version, part_number: u64) -> Option<(u64, u64)> {
let mut offset = 0;
for (i, (bk, bv)) in v.blocks.items().iter().enumerate() {
if bk.part_number == part_number {
let size: u64 = v.blocks.items()[i..]
.iter()
.take_while(|(k, _)| k.part_number == part_number)
.map(|(_, v)| v.size)
.sum();
return Some((offset, offset + size));
}
offset += bv.size;
}
None
}
fn body_from_blocks_range(
garage: Arc<Garage>,
all_blocks: &[(VersionBlockKey, VersionBlock)],
begin: u64,
end: u64,
) -> Body {
// We will store here the list of blocks that have an intersection with the requested
// range, as well as their "true offset", which is their actual offset in the complete
// file (whereas block.offset designates the offset of the block WITHIN THE PART
// block.part_number, which is not the same in the case of a multipart upload)
let mut blocks: Vec<(VersionBlock, u64)> = Vec::with_capacity(std::cmp::min(
all_blocks.len(),
4 + ((end - begin) / std::cmp::max(all_blocks[0].1.size as u64, 1024)) as usize,
));
let mut true_offset = 0;
for (_, b) in all_blocks.iter() {
if true_offset >= end {
break;
}
// Keep only blocks that have an intersection with the requested range
if true_offset < end && true_offset + b.size > begin {
blocks.push((*b, true_offset));
}
true_offset += b.size;
}
let body_stream = futures::stream::iter(blocks)
.map(move |(block, true_offset)| {
let garage = garage.clone();
async move {
let data = garage.block_manager.rpc_get_block(&block.hash).await?;
let data = Bytes::from(data);
let start_in_block = if true_offset > begin {
0
} else {
begin - true_offset
};
let end_in_block = if true_offset + block.size < end {
block.size
} else {
end - true_offset
};
Result::<Bytes, Error>::Ok(
data.slice(start_in_block as usize..end_in_block as usize),
)
}
})
.buffered(2);
hyper::body::Body::wrap_stream(body_stream)
}
+371 -5
View File
@@ -1,3 +1,4 @@
use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet};
use std::iter::{Iterator, Peekable};
use std::sync::Arc;
@@ -10,12 +11,18 @@ use garage_util::time::*;
use garage_model::garage::Garage;
use garage_model::object_table::*;
use garage_model::version_table::Version;
use garage_table::EmptyKey;
use crate::encoding::*;
use crate::error::*;
use crate::s3_put;
use crate::s3_xml;
const DUMMY_NAME: &str = "Dummy Key";
const DUMMY_KEY: &str = "GKDummyKey";
#[derive(Debug)]
pub struct ListQueryCommon {
pub bucket_name: String,
@@ -42,6 +49,16 @@ pub struct ListMultipartUploadsQuery {
pub common: ListQueryCommon,
}
#[derive(Debug)]
pub struct ListPartsQuery {
pub bucket_name: String,
pub bucket_id: Uuid,
pub key: String,
pub upload_id: String,
pub part_number_marker: Option<u64>,
pub max_parts: u64,
}
pub async fn handle_list(
garage: Arc<Garage>,
query: &ListObjectsQuery,
@@ -54,6 +71,7 @@ pub async fn handle_list(
}
};
debug!("ListObjects {:?}", query);
let mut acc = query.build_accumulator();
let pagination = fetch_list_entries(&query.common, query.begin()?, &mut acc, &io).await?;
@@ -121,7 +139,7 @@ pub async fn handle_list(
key: uriencode_maybe(key, query.common.urlencode_resp),
last_modified: s3_xml::Value(msec_to_rfc3339(info.last_modified)),
size: s3_xml::IntValue(info.size as i64),
etag: s3_xml::Value(format!("\"{}\"", info.etag.to_string())),
etag: s3_xml::Value(format!("\"{}\"", info.etag)),
storage_class: s3_xml::Value("STANDARD".to_string()),
})
.collect(),
@@ -152,6 +170,7 @@ pub async fn handle_list_multipart_upload(
}
};
debug!("ListMultipartUploads {:?}", query);
let mut acc = query.build_accumulator();
let pagination = fetch_list_entries(&query.common, query.begin()?, &mut acc, &io).await?;
@@ -208,12 +227,12 @@ pub async fn handle_list_multipart_upload(
upload_id: s3_xml::Value(hex::encode(uuid)),
storage_class: s3_xml::Value("STANDARD".to_string()),
initiator: s3_xml::Initiator {
display_name: s3_xml::Value("Dummy Key".to_string()),
id: s3_xml::Value("GKDummyKey".to_string()),
display_name: s3_xml::Value(DUMMY_NAME.to_string()),
id: s3_xml::Value(DUMMY_KEY.to_string()),
},
owner: s3_xml::Owner {
display_name: s3_xml::Value("Dummy Key".to_string()),
id: s3_xml::Value("GKDummyKey".to_string()),
display_name: s3_xml::Value(DUMMY_NAME.to_string()),
id: s3_xml::Value(DUMMY_KEY.to_string()),
},
})
.collect(),
@@ -233,6 +252,57 @@ pub async fn handle_list_multipart_upload(
.body(Body::from(xml.into_bytes()))?)
}
pub async fn handle_list_parts(
garage: Arc<Garage>,
query: &ListPartsQuery,
) -> Result<Response<Body>, Error> {
debug!("ListParts {:?}", query);
let upload_id = s3_put::decode_upload_id(&query.upload_id)?;
let (object, version) = futures::try_join!(
garage.object_table.get(&query.bucket_id, &query.key),
garage.version_table.get(&upload_id, &EmptyKey),
)?;
let (info, next) = fetch_part_info(query, object, version, upload_id)?;
let result = s3_xml::ListPartsResult {
xmlns: (),
bucket: s3_xml::Value(query.bucket_name.to_string()),
key: s3_xml::Value(query.key.to_string()),
upload_id: s3_xml::Value(query.upload_id.to_string()),
part_number_marker: query.part_number_marker.map(|e| s3_xml::IntValue(e as i64)),
next_part_number_marker: next.map(|e| s3_xml::IntValue(e as i64)),
max_parts: s3_xml::IntValue(query.max_parts as i64),
is_truncated: s3_xml::Value(next.map(|_| "true").unwrap_or("false").to_string()),
parts: info
.iter()
.map(|part| s3_xml::PartItem {
etag: s3_xml::Value(format!("\"{}\"", part.etag)),
last_modified: s3_xml::Value(msec_to_rfc3339(part.timestamp)),
part_number: s3_xml::IntValue(part.part_number as i64),
size: s3_xml::IntValue(part.size as i64),
})
.collect(),
initiator: s3_xml::Initiator {
display_name: s3_xml::Value(DUMMY_NAME.to_string()),
id: s3_xml::Value(DUMMY_KEY.to_string()),
},
owner: s3_xml::Owner {
display_name: s3_xml::Value(DUMMY_NAME.to_string()),
id: s3_xml::Value(DUMMY_KEY.to_string()),
},
storage_class: s3_xml::Value("STANDARD".to_string()),
};
let xml = s3_xml::to_xml_with_header(&result)?;
Ok(Response::builder()
.header("Content-Type", "application/xml")
.body(Body::from(xml.into_bytes()))?)
}
/*
* Private enums and structs
*/
@@ -250,6 +320,14 @@ struct UploadInfo {
timestamp: u64,
}
#[derive(Debug, PartialEq)]
struct PartInfo {
etag: String,
timestamp: u64,
part_number: u64,
size: u64,
}
enum ExtractionResult {
NoMore,
Filled,
@@ -364,6 +442,109 @@ where
}
}
fn fetch_part_info(
query: &ListPartsQuery,
object: Option<Object>,
version: Option<Version>,
upload_id: Uuid,
) -> Result<(Vec<PartInfo>, Option<u64>), Error> {
// Check results
let object = object.ok_or(Error::NoSuchKey)?;
let obj_version = object
.versions()
.iter()
.find(|v| v.uuid == upload_id && v.is_uploading())
.ok_or(Error::NoSuchUpload)?;
let version = version.ok_or(Error::NoSuchKey)?;
// Cut the beginning of our 2 vectors if required
let (etags, blocks) = match &query.part_number_marker {
Some(marker) => {
let next = marker + 1;
let part_idx = into_ok_or_err(
version
.parts_etags
.items()
.binary_search_by(|(part_num, _)| part_num.cmp(&next)),
);
let parts = &version.parts_etags.items()[part_idx..];
let block_idx = into_ok_or_err(
version
.blocks
.items()
.binary_search_by(|(vkey, _)| vkey.part_number.cmp(&next)),
);
let blocks = &version.blocks.items()[block_idx..];
(parts, blocks)
}
None => (version.parts_etags.items(), version.blocks.items()),
};
// Use the block vector to compute a (part_number, size) vector
let mut size = Vec::<(u64, u64)>::new();
blocks.iter().for_each(|(key, val)| {
let mut new_size = val.size;
match size.pop() {
Some((part_number, size)) if part_number == key.part_number => new_size += size,
Some(v) => size.push(v),
None => (),
}
size.push((key.part_number, new_size))
});
// Merge the etag vector and size vector to build a PartInfo vector
let max_parts = query.max_parts as usize;
let (mut etag_iter, mut size_iter) = (etags.iter().peekable(), size.iter().peekable());
let mut info = Vec::<PartInfo>::with_capacity(max_parts);
while info.len() < max_parts {
match (etag_iter.peek(), size_iter.peek()) {
(Some((ep, etag)), Some((sp, size))) => match ep.cmp(sp) {
Ordering::Less => {
debug!("ETag information ignored due to missing corresponding block information. Query: {:?}", query);
etag_iter.next();
}
Ordering::Equal => {
info.push(PartInfo {
etag: etag.to_string(),
timestamp: obj_version.timestamp,
part_number: *ep,
size: *size,
});
etag_iter.next();
size_iter.next();
}
Ordering::Greater => {
debug!("Block information ignored due to missing corresponding ETag information. Query: {:?}", query);
size_iter.next();
}
},
(None, None) => return Ok((info, None)),
_ => {
debug!(
"Additional block or ETag information ignored. Query: {:?}",
query
);
return Ok((info, None));
}
}
}
match info.last() {
Some(part_info) => {
let pagination = Some(part_info.part_number);
Ok((info, pagination))
}
None => Ok((info, None)),
}
}
/*
* ListQuery logic
*/
@@ -715,6 +896,14 @@ impl ExtractAccumulator for UploadAccumulator {
* Utility functions
*/
/// This is a stub for Result::into_ok_or_err that is not yet in Rust stable
fn into_ok_or_err<T>(r: Result<T, T>) -> T {
match r {
Ok(r) => r,
Err(r) => r,
}
}
/// Returns the common prefix of the object given the query prefix and delimiter
fn common_prefix<'a>(object: &'a Object, query: &ListQueryCommon) -> Option<&'a str> {
match &query.delimiter {
@@ -766,6 +955,8 @@ fn key_after_prefix(pfx: &str) -> Option<String> {
#[cfg(test)]
mod tests {
use super::*;
use garage_model::version_table::*;
use garage_util::*;
use std::iter::FromIterator;
const TS: u64 = 1641394898314;
@@ -1014,4 +1205,179 @@ mod tests {
Ok(())
}
fn version() -> Version {
let uuid = Uuid::from([0x08; 32]);
let blocks = vec![
(
VersionBlockKey {
part_number: 1,
offset: 1,
},
VersionBlock {
hash: uuid,
size: 3,
},
),
(
VersionBlockKey {
part_number: 1,
offset: 2,
},
VersionBlock {
hash: uuid,
size: 2,
},
),
(
VersionBlockKey {
part_number: 2,
offset: 1,
},
VersionBlock {
hash: uuid,
size: 8,
},
),
(
VersionBlockKey {
part_number: 5,
offset: 1,
},
VersionBlock {
hash: uuid,
size: 7,
},
),
(
VersionBlockKey {
part_number: 8,
offset: 1,
},
VersionBlock {
hash: uuid,
size: 5,
},
),
];
let etags = vec![
(1, "etag1".to_string()),
(3, "etag2".to_string()),
(5, "etag3".to_string()),
(8, "etag4".to_string()),
(9, "etag5".to_string()),
];
Version {
bucket_id: uuid,
key: "a".to_string(),
uuid: uuid,
deleted: false.into(),
blocks: crdt::Map::<VersionBlockKey, VersionBlock>::from_iter(blocks),
parts_etags: crdt::Map::<u64, String>::from_iter(etags),
}
}
fn obj() -> Object {
Object::new(bucket(), "d".to_string(), vec![objup_version([0x08; 32])])
}
#[test]
fn test_fetch_part_info() -> Result<(), Error> {
let uuid = Uuid::from([0x08; 32]);
let mut query = ListPartsQuery {
bucket_name: "a".to_string(),
bucket_id: uuid,
key: "a".to_string(),
upload_id: "xx".to_string(),
part_number_marker: None,
max_parts: 2,
};
assert!(
fetch_part_info(&query, None, None, uuid).is_err(),
"No object and version should fail"
);
assert!(
fetch_part_info(&query, Some(obj()), None, uuid).is_err(),
"No version should faild"
);
assert!(
fetch_part_info(&query, None, Some(version()), uuid).is_err(),
"No object should fail"
);
// Start from the beginning but with limited size to trigger pagination
let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
assert_eq!(pagination.unwrap(), 5);
assert_eq!(
info,
vec![
PartInfo {
etag: "etag1".to_string(),
timestamp: TS,
part_number: 1,
size: 5
},
PartInfo {
etag: "etag3".to_string(),
timestamp: TS,
part_number: 5,
size: 7
},
]
);
// Use previous pagination to make a new request
query.part_number_marker = Some(pagination.unwrap());
let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
assert!(pagination.is_none());
assert_eq!(
info,
vec![PartInfo {
etag: "etag4".to_string(),
timestamp: TS,
part_number: 8,
size: 5
},]
);
// Trying to access a part that is way larger than registered ones
query.part_number_marker = Some(9999);
let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
assert!(pagination.is_none());
assert_eq!(info, vec![]);
// Try without any limitation
query.max_parts = 1000;
query.part_number_marker = None;
let (info, pagination) = fetch_part_info(&query, Some(obj()), Some(version()), uuid)?;
assert!(pagination.is_none());
assert_eq!(
info,
vec![
PartInfo {
etag: "etag1".to_string(),
timestamp: TS,
part_number: 1,
size: 5
},
PartInfo {
etag: "etag3".to_string(),
timestamp: TS,
part_number: 5,
size: 7
},
PartInfo {
etag: "etag4".to_string(),
timestamp: TS,
part_number: 8,
size: 5
},
]
);
Ok(())
}
}
+138 -30
View File
@@ -1,8 +1,10 @@
use std::collections::{BTreeMap, VecDeque};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::sync::Arc;
use futures::stream::*;
use hyper::{Body, Request, Response};
use chrono::{DateTime, NaiveDateTime, Utc};
use futures::{prelude::*, TryFutureExt};
use hyper::body::{Body, Bytes};
use hyper::{Request, Response};
use md5::{digest::generic_array::*, Digest as Md5Digest, Md5};
use sha2::Sha256;
@@ -14,19 +16,23 @@ use garage_util::time::*;
use garage_model::block::INLINE_THRESHOLD;
use garage_model::block_ref_table::*;
use garage_model::garage::Garage;
use garage_model::key_table::Key;
use garage_model::object_table::*;
use garage_model::version_table::*;
use crate::error::*;
use crate::s3_xml;
use crate::signature::verify_signed_content;
use crate::signature::streaming::SignedPayloadStream;
use crate::signature::LONG_DATETIME;
use crate::signature::{compute_scope, verify_signed_content};
pub async fn handle_put(
garage: Arc<Garage>,
req: Request<Body>,
bucket_id: Uuid,
key: &str,
content_sha256: Option<Hash>,
api_key: &Key,
mut content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> {
// Generate identity of new version
let version_uuid = gen_uuid();
@@ -40,11 +46,53 @@ pub async fn handle_put(
Some(x) => Some(x.to_str()?.to_string()),
None => None,
};
let payload_seed_signature = match req.headers().get("x-amz-content-sha256") {
Some(header) if header == "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" => {
let content_sha256 = content_sha256
.take()
.ok_or_bad_request("No signature provided")?;
Some(content_sha256)
}
_ => None,
};
// Parse body of uploaded file
let body = req.into_body();
let (head, body) = req.into_parts();
let body = body.map_err(Error::from);
let mut chunker = BodyChunker::new(body, garage.config.block_size);
let body = if let Some(signature) = payload_seed_signature {
let secret_key = &api_key
.state
.as_option()
.ok_or_internal_error("Deleted key state")?
.secret_key;
let date = head
.headers
.get("x-amz-date")
.ok_or_bad_request("Missing X-Amz-Date field")?
.to_str()?;
let date: NaiveDateTime =
NaiveDateTime::parse_from_str(date, LONG_DATETIME).ok_or_bad_request("Invalid date")?;
let date: DateTime<Utc> = DateTime::from_utc(date, Utc);
let scope = compute_scope(&date, &garage.config.s3_api.s3_region);
let signing_hmac = crate::signature::signing_hmac(
&date,
secret_key,
&garage.config.s3_api.s3_region,
"s3",
)
.ok_or_internal_error("Unable to build signing HMAC")?;
SignedPayloadStream::new(body, signing_hmac, date, &scope, signature)?
.map_err(Error::from)
.boxed()
} else {
body.boxed()
};
let mut chunker = StreamChunker::new(body, garage.config.block_size);
let first_block = chunker.next().await?.unwrap_or_default();
// If body is small enough, store it directly in the object table
@@ -178,13 +226,13 @@ fn ensure_checksum_matches(
Ok(())
}
async fn read_and_put_blocks(
async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
garage: &Garage,
version: &Version,
part_number: u64,
first_block: Vec<u8>,
first_block_hash: Hash,
chunker: &mut BodyChunker,
chunker: &mut StreamChunker<S>,
) -> Result<(u64, GenericArray<u8, typenum::U16>, Hash), Error> {
let mut md5hasher = Md5::new();
let mut sha256hasher = Sha256::new();
@@ -205,8 +253,11 @@ async fn read_and_put_blocks(
.rpc_put_block(first_block_hash, first_block);
loop {
let (_, _, next_block) =
futures::try_join!(put_curr_block, put_curr_version_block, chunker.next())?;
let (_, _, next_block) = futures::try_join!(
put_curr_block.map_err(Error::from),
put_curr_version_block.map_err(Error::from),
chunker.next(),
)?;
if let Some(block) = next_block {
md5hasher.update(&block[..]);
sha256hasher.update(&block[..]);
@@ -266,32 +317,34 @@ async fn put_block_meta(
Ok(())
}
struct BodyChunker {
body: Body,
struct StreamChunker<S: Stream<Item = Result<Bytes, Error>>> {
stream: S,
read_all: bool,
block_size: usize,
buf: VecDeque<u8>,
}
impl BodyChunker {
fn new(body: Body, block_size: usize) -> Self {
impl<S: Stream<Item = Result<Bytes, Error>> + Unpin> StreamChunker<S> {
fn new(stream: S, block_size: usize) -> Self {
Self {
body,
stream,
read_all: false,
block_size,
buf: VecDeque::with_capacity(2 * block_size),
}
}
async fn next(&mut self) -> Result<Option<Vec<u8>>, GarageError> {
async fn next(&mut self) -> Result<Option<Vec<u8>>, Error> {
while !self.read_all && self.buf.len() < self.block_size {
if let Some(block) = self.body.next().await {
if let Some(block) = self.stream.next().await {
let bytes = block?;
trace!("Body next: {} bytes", bytes.len());
self.buf.extend(&bytes[..]);
self.buf.extend(bytes);
} else {
self.read_all = true;
}
}
if self.buf.is_empty() {
Ok(None)
} else if self.buf.len() <= self.block_size {
@@ -368,12 +421,20 @@ pub async fn handle_put_part(
// Read first chuck, and at the same time try to get object to see if it exists
let key = key.to_string();
let mut chunker = BodyChunker::new(req.into_body(), garage.config.block_size);
let body = req.into_body().map_err(Error::from);
let mut chunker = StreamChunker::new(body, garage.config.block_size);
let (object, version, first_block) = futures::try_join!(
garage.object_table.get(&bucket_id, &key),
garage.version_table.get(&version_uuid, &EmptyKey),
chunker.next()
garage
.object_table
.get(&bucket_id, &key)
.map_err(Error::from),
garage
.version_table
.get(&version_uuid, &EmptyKey)
.map_err(Error::from),
chunker.next(),
)?;
// Check object is valid and multipart block can be accepted
@@ -444,10 +505,13 @@ pub async fn handle_complete_multipart_upload(
content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> {
let body = hyper::body::to_bytes(req.into_body()).await?;
verify_signed_content(content_sha256, &body[..])?;
if let Some(content_sha256) = content_sha256 {
verify_signed_content(content_sha256, &body[..])?;
}
let body_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
let body_list_of_parts = parse_complete_multpart_upload_body(&body_xml)
let body_list_of_parts = parse_complete_multipart_upload_body(&body_xml)
.ok_or_bad_request("Invalid CompleteMultipartUpload XML")?;
debug!(
"CompleteMultipartUpload list of parts: {:?}",
@@ -456,6 +520,7 @@ pub async fn handle_complete_multipart_upload(
let version_uuid = decode_upload_id(upload_id)?;
// Get object and version
let key = key.to_string();
let (object, version) = futures::try_join!(
garage.object_table.get(&bucket_id, &key),
@@ -480,6 +545,31 @@ pub async fn handle_complete_multipart_upload(
_ => unreachable!(),
};
// Check that part numbers are an increasing sequence.
// (it doesn't need to start at 1 nor to be a continuous sequence,
// see discussion in #192)
if body_list_of_parts.is_empty() {
return Err(Error::EntityTooSmall);
}
if !body_list_of_parts
.iter()
.zip(body_list_of_parts.iter().skip(1))
.all(|(p1, p2)| p1.part_number < p2.part_number)
{
return Err(Error::InvalidPartOrder);
}
// Garage-specific restriction, see #204: part numbers must be
// consecutive starting at 1
if body_list_of_parts[0].part_number != 1
|| !body_list_of_parts
.iter()
.zip(body_list_of_parts.iter().skip(1))
.all(|(p1, p2)| p1.part_number + 1 == p2.part_number)
{
return Err(Error::NotImplemented("Garage does not support completing a Multipart upload with non-consecutive part numbers. This is a restriction of Garage's data model, which might be fixed in a future release. See issue #204 for more information on this topic.".into()));
}
// Check that the list of parts they gave us corresponds to the parts we have here
debug!("Expected parts from request: {:?}", body_list_of_parts);
debug!("Parts stored in version: {:?}", version.parts_etags.items());
@@ -492,18 +582,31 @@ pub async fn handle_complete_multipart_upload(
.iter()
.map(|x| (&x.part_number, &x.etag))
.eq(parts);
if !same_parts {
return Err(Error::InvalidPart);
}
// Check that all blocks belong to one of the parts
let block_parts = version
.blocks
.items()
.iter()
.map(|(bk, _)| bk.part_number)
.collect::<BTreeSet<_>>();
let same_parts = body_list_of_parts
.iter()
.map(|x| x.part_number)
.eq(block_parts.into_iter());
if !same_parts {
return Err(Error::BadRequest(
"We don't have the same parts".to_string(),
"Part numbers in block list and part list do not match. This can happen if a part was partially uploaded. Please abort the multipart upload and try again.".into(),
));
}
// Calculate etag of final object
// To understand how etags are calculated, read more here:
// https://teppen.io/2018/06/23/aws_s3_etags/
let num_parts = version.blocks.items().last().unwrap().0.part_number
- version.blocks.items().first().unwrap().0.part_number
+ 1;
let num_parts = body_list_of_parts.len();
let mut etag_md5_hasher = Md5::new();
for (_, etag) in version.parts_etags.items().iter() {
etag_md5_hasher.update(etag.as_bytes());
@@ -639,7 +742,7 @@ struct CompleteMultipartUploadPart {
part_number: u64,
}
fn parse_complete_multpart_upload_body(
fn parse_complete_multipart_upload_body(
xml: &roxmltree::Document,
) -> Option<Vec<CompleteMultipartUploadPart>> {
let mut parts = vec![];
@@ -651,6 +754,11 @@ fn parse_complete_multpart_upload_body(
}
for item in cmu.children() {
// Only parse <Part> nodes
if !item.is_element() {
continue;
}
if item.has_tag_name("Part") {
let etag = item.children().find(|e| e.has_tag_name("ETag"))?.text()?;
let part_number = item
+127 -219
View File
@@ -8,6 +8,17 @@ use hyper::{HeaderMap, Method, Request};
/// This macro is used to generate very repetitive match {} blocks in this module
/// It is _not_ made to be used anywhere else
macro_rules! s3_match {
(@match $enum:expr , [ $($endpoint:ident,)* ]) => {{
// usage: s3_match {@match my_enum, [ VariantWithField1, VariantWithField2 ..] }
// returns true if the variant was one of the listed variants, false otherwise.
use Endpoint::*;
match $enum {
$(
$endpoint { .. } => true,
)*
_ => false
}
}};
(@extract $enum:expr , $param:ident, [ $($endpoint:ident,)* ]) => {{
// usage: s3_match {@extract my_enum, field_name, [ VariantWithField1, VariantWithField2 ..] }
// returns Some(field_value), or None if the variant was not one of the listed variants.
@@ -19,10 +30,10 @@ macro_rules! s3_match {
_ => None
}
}};
(@gen_parser ($keyword:expr, $key:expr, $bucket:expr, $query:expr, $header:expr),
(@gen_parser ($keyword:expr, $key:expr, $query:expr, $header:expr),
key: [$($kw_k:ident $(if $required_k:ident)? $(header $header_k:expr)? => $api_k:ident $(($($conv_k:ident :: $param_k:ident),*))?,)*],
no_key: [$($kw_nk:ident $(if $required_nk:ident)? $(if_header $header_nk:expr)? => $api_nk:ident $(($($conv_nk:ident :: $param_nk:ident),*))?,)*]) => {{
// usage: s3_match {@gen_parser (keyword, key, bucket, query, header),
// usage: s3_match {@gen_parser (keyword, key, query, header),
// key: [
// SOME_KEYWORD => VariantWithKey,
// ...
@@ -38,7 +49,6 @@ macro_rules! s3_match {
match ($keyword, !$key.is_empty()){
$(
($kw_k, true) if true $(&& $query.$required_k.is_some())? $(&& $header.contains_key($header_k))? => Ok($api_k {
bucket: $bucket,
key: $key,
$($(
$param_k: s3_match!(@@parse_param $query, $conv_k, $param_k),
@@ -47,7 +57,6 @@ macro_rules! s3_match {
)*
$(
($kw_nk, false) $(if $query.$required_nk.is_some())? $(if $header.contains($header_nk))? => Ok($api_nk {
bucket: $bucket,
$($(
$param_nk: s3_match!(@@parse_param $query, $conv_nk, $param_nk),
)*)?
@@ -87,7 +96,6 @@ macro_rules! s3_match {
$(
$(#[$outer:meta])*
$variant:ident $({
bucket: String,
$($name:ident: $ty:ty,)*
})?,
)*
@@ -97,7 +105,6 @@ macro_rules! s3_match {
$(
$(#[$outer])*
$variant $({
bucket: String,
$($name: $ty, )*
})?,
)*
@@ -108,15 +115,6 @@ macro_rules! s3_match {
$(Endpoint::$variant $({ $($name: _,)* .. })? => stringify!($variant),)*
}
}
/// Get the bucket the request target. Returns None for requests not related to a bucket.
pub fn get_bucket(&self) -> Option<&str> {
match self {
$(
Endpoint::$variant $({ bucket, $($name: _,)* .. })? => s3_match!{@if ($(bucket $($name)*)?) then (Some(bucket)) else (None)},
)*
}
}
}
};
(@if ($($cond:tt)+) then ($($then:tt)*) else ($($else:tt)*)) => {
@@ -138,215 +136,158 @@ s3_match! {@func
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Endpoint {
AbortMultipartUpload {
bucket: String,
key: String,
upload_id: String,
},
CompleteMultipartUpload {
bucket: String,
key: String,
upload_id: String,
},
CopyObject {
bucket: String,
key: String,
},
CreateBucket {
bucket: String,
},
CreateMultipartUpload {
bucket: String,
key: String,
},
DeleteBucket {
bucket: String,
},
DeleteBucketAnalyticsConfiguration {
bucket: String,
id: String,
},
DeleteBucketCors {
bucket: String,
},
DeleteBucketEncryption {
bucket: String,
},
DeleteBucketIntelligentTieringConfiguration {
bucket: String,
id: String,
},
DeleteBucketInventoryConfiguration {
bucket: String,
id: String,
},
DeleteBucketLifecycle {
bucket: String,
},
DeleteBucketMetricsConfiguration {
bucket: String,
id: String,
},
DeleteBucketOwnershipControls {
bucket: String,
},
DeleteBucketPolicy {
bucket: String,
},
DeleteBucketReplication {
bucket: String,
},
DeleteBucketTagging {
bucket: String,
},
DeleteBucketWebsite {
bucket: String,
},
DeleteObject {
bucket: String,
key: String,
version_id: Option<String>,
},
DeleteObjects {
bucket: String,
},
DeleteObjectTagging {
bucket: String,
key: String,
version_id: Option<String>,
},
DeletePublicAccessBlock {
bucket: String,
},
GetBucketAccelerateConfiguration {
bucket: String,
},
GetBucketAcl {
bucket: String,
},
GetBucketAnalyticsConfiguration {
bucket: String,
id: String,
},
GetBucketCors {
bucket: String,
},
GetBucketEncryption {
bucket: String,
},
GetBucketIntelligentTieringConfiguration {
bucket: String,
id: String,
},
GetBucketInventoryConfiguration {
bucket: String,
id: String,
},
GetBucketLifecycleConfiguration {
bucket: String,
},
GetBucketLocation {
bucket: String,
},
GetBucketLogging {
bucket: String,
},
GetBucketMetricsConfiguration {
bucket: String,
id: String,
},
GetBucketNotificationConfiguration {
bucket: String,
},
GetBucketOwnershipControls {
bucket: String,
},
GetBucketPolicy {
bucket: String,
},
GetBucketPolicyStatus {
bucket: String,
},
GetBucketReplication {
bucket: String,
},
GetBucketRequestPayment {
bucket: String,
},
GetBucketTagging {
bucket: String,
},
GetBucketVersioning {
bucket: String,
},
GetBucketWebsite {
bucket: String,
},
/// There are actually many more query parameters, used to add headers to the answer. They were
/// not added here as they are best handled in a dedicated route.
GetObject {
bucket: String,
key: String,
part_number: Option<u64>,
version_id: Option<String>,
},
GetObjectAcl {
bucket: String,
key: String,
version_id: Option<String>,
},
GetObjectLegalHold {
bucket: String,
key: String,
version_id: Option<String>,
},
GetObjectLockConfiguration {
bucket: String,
},
GetObjectRetention {
bucket: String,
key: String,
version_id: Option<String>,
},
GetObjectTagging {
bucket: String,
key: String,
version_id: Option<String>,
},
GetObjectTorrent {
bucket: String,
key: String,
},
GetPublicAccessBlock {
bucket: String,
},
HeadBucket {
bucket: String,
},
HeadObject {
bucket: String,
key: String,
part_number: Option<u64>,
version_id: Option<String>,
},
ListBucketAnalyticsConfigurations {
bucket: String,
continuation_token: Option<String>,
},
ListBucketIntelligentTieringConfigurations {
bucket: String,
continuation_token: Option<String>,
},
ListBucketInventoryConfigurations {
bucket: String,
continuation_token: Option<String>,
},
ListBucketMetricsConfigurations {
bucket: String,
continuation_token: Option<String>,
},
ListBuckets,
ListMultipartUploads {
bucket: String,
delimiter: Option<char>,
encoding_type: Option<String>,
key_marker: Option<String>,
@@ -355,7 +296,6 @@ pub enum Endpoint {
upload_id_marker: Option<String>,
},
ListObjects {
bucket: String,
delimiter: Option<char>,
encoding_type: Option<String>,
marker: Option<String>,
@@ -363,7 +303,6 @@ pub enum Endpoint {
prefix: Option<String>,
},
ListObjectsV2 {
bucket: String,
// This value should always be 2. It is not checked when constructing the struct
list_type: String,
continuation_token: Option<String>,
@@ -375,7 +314,6 @@ pub enum Endpoint {
start_after: Option<String>,
},
ListObjectVersions {
bucket: String,
delimiter: Option<char>,
encoding_type: Option<String>,
key_marker: Option<String>,
@@ -384,119 +322,90 @@ pub enum Endpoint {
version_id_marker: Option<String>,
},
ListParts {
bucket: String,
key: String,
max_parts: Option<u64>,
part_number_marker: Option<u64>,
upload_id: String,
},
Options,
PutBucketAccelerateConfiguration {
bucket: String,
},
PutBucketAcl {
bucket: String,
},
PutBucketAnalyticsConfiguration {
bucket: String,
id: String,
},
PutBucketCors {
bucket: String,
},
PutBucketEncryption {
bucket: String,
},
PutBucketIntelligentTieringConfiguration {
bucket: String,
id: String,
},
PutBucketInventoryConfiguration {
bucket: String,
id: String,
},
PutBucketLifecycleConfiguration {
bucket: String,
},
PutBucketLogging {
bucket: String,
},
PutBucketMetricsConfiguration {
bucket: String,
id: String,
},
PutBucketNotificationConfiguration {
bucket: String,
},
PutBucketOwnershipControls {
bucket: String,
},
PutBucketPolicy {
bucket: String,
},
PutBucketReplication {
bucket: String,
},
PutBucketRequestPayment {
bucket: String,
},
PutBucketTagging {
bucket: String,
},
PutBucketVersioning {
bucket: String,
},
PutBucketWebsite {
bucket: String,
},
PutObject {
bucket: String,
key: String,
},
PutObjectAcl {
bucket: String,
key: String,
version_id: Option<String>,
},
PutObjectLegalHold {
bucket: String,
key: String,
version_id: Option<String>,
},
PutObjectLockConfiguration {
bucket: String,
},
PutObjectRetention {
bucket: String,
key: String,
version_id: Option<String>,
},
PutObjectTagging {
bucket: String,
key: String,
version_id: Option<String>,
},
PutPublicAccessBlock {
bucket: String,
},
RestoreObject {
bucket: String,
key: String,
version_id: Option<String>,
},
SelectObjectContent {
bucket: String,
key: String,
// This value should always be 2. It is not checked when constructing the struct
select_type: String,
},
UploadPart {
bucket: String,
key: String,
part_number: u64,
upload_id: String,
},
UploadPartCopy {
bucket: String,
key: String,
part_number: u64,
upload_id: String,
@@ -506,12 +415,16 @@ pub enum Endpoint {
impl Endpoint {
/// Determine which S3 endpoint a request is for using the request, and a bucket which was
/// possibly extracted from the Host header.
pub fn from_request<T>(req: &Request<T>, bucket: Option<String>) -> Result<Self, Error> {
/// Returns Self plus bucket name, if endpoint is not Endpoint::ListBuckets
pub fn from_request<T>(
req: &Request<T>,
bucket: Option<String>,
) -> Result<(Self, Option<String>), Error> {
let uri = req.uri();
let path = uri.path().trim_start_matches('/');
let query = uri.query();
if bucket.is_none() && path.is_empty() {
return Ok(Self::ListBuckets);
return Ok((Self::ListBuckets, None));
}
let (bucket, key) = if let Some(bucket) = bucket {
@@ -522,6 +435,10 @@ impl Endpoint {
.unwrap_or((path.to_owned(), ""))
};
if *req.method() == Method::OPTIONS {
return Ok((Self::Options, Some(bucket)));
}
let key = percent_encoding::percent_decode_str(key)
.decode_utf8()?
.into_owned();
@@ -529,29 +446,25 @@ impl Endpoint {
let mut query = QueryParameters::from_query(query.unwrap_or_default())?;
let res = match *req.method() {
Method::GET => Self::from_get(bucket, key, &mut query)?,
Method::HEAD => Self::from_head(bucket, key, &mut query)?,
Method::POST => Self::from_post(bucket, key, &mut query)?,
Method::PUT => Self::from_put(bucket, key, &mut query, req.headers())?,
Method::DELETE => Self::from_delete(bucket, key, &mut query)?,
Method::GET => Self::from_get(key, &mut query)?,
Method::HEAD => Self::from_head(key, &mut query)?,
Method::POST => Self::from_post(key, &mut query)?,
Method::PUT => Self::from_put(key, &mut query, req.headers())?,
Method::DELETE => Self::from_delete(key, &mut query)?,
_ => return Err(Error::BadRequest("Unknown method".to_owned())),
};
if let Some(message) = query.nonempty_message() {
debug!("Unused query parameter: {}", message)
}
Ok(res)
Ok((res, Some(bucket)))
}
/// Determine which endpoint a request is for, knowing it is a GET.
fn from_get(
bucket: String,
key: String,
query: &mut QueryParameters<'_>,
) -> Result<Self, Error> {
fn from_get(key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
s3_match! {
@gen_parser
(query.keyword.take().unwrap_or_default().as_ref(), key, bucket, query, None),
(query.keyword.take().unwrap_or_default().as_ref(), key, query, None),
key: [
EMPTY if upload_id => ListParts (query::upload_id, opt_parse::max_parts, opt_parse::part_number_marker),
EMPTY => GetObject (query_opt::version_id, opt_parse::part_number),
@@ -605,14 +518,10 @@ impl Endpoint {
}
/// Determine which endpoint a request is for, knowing it is a HEAD.
fn from_head(
bucket: String,
key: String,
query: &mut QueryParameters<'_>,
) -> Result<Self, Error> {
fn from_head(key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
s3_match! {
@gen_parser
(query.keyword.take().unwrap_or_default().as_ref(), key, bucket, query, None),
(query.keyword.take().unwrap_or_default().as_ref(), key, query, None),
key: [
EMPTY => HeadObject(opt_parse::part_number, query_opt::version_id),
],
@@ -623,14 +532,10 @@ impl Endpoint {
}
/// Determine which endpoint a request is for, knowing it is a POST.
fn from_post(
bucket: String,
key: String,
query: &mut QueryParameters<'_>,
) -> Result<Self, Error> {
fn from_post(key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
s3_match! {
@gen_parser
(query.keyword.take().unwrap_or_default().as_ref(), key, bucket, query, None),
(query.keyword.take().unwrap_or_default().as_ref(), key, query, None),
key: [
EMPTY if upload_id => CompleteMultipartUpload (query::upload_id),
RESTORE => RestoreObject (query_opt::version_id),
@@ -645,14 +550,13 @@ impl Endpoint {
/// Determine which endpoint a request is for, knowing it is a PUT.
fn from_put(
bucket: String,
key: String,
query: &mut QueryParameters<'_>,
headers: &HeaderMap<HeaderValue>,
) -> Result<Self, Error> {
s3_match! {
@gen_parser
(query.keyword.take().unwrap_or_default().as_ref(), key, bucket, query, headers),
(query.keyword.take().unwrap_or_default().as_ref(), key, query, headers),
key: [
EMPTY if part_number header "x-amz-copy-source" => UploadPartCopy (parse::part_number, query::upload_id),
EMPTY header "x-amz-copy-source" => CopyObject,
@@ -691,14 +595,10 @@ impl Endpoint {
}
/// Determine which endpoint a request is for, knowing it is a DELETE.
fn from_delete(
bucket: String,
key: String,
query: &mut QueryParameters<'_>,
) -> Result<Self, Error> {
fn from_delete(key: String, query: &mut QueryParameters<'_>) -> Result<Self, Error> {
s3_match! {
@gen_parser
(query.keyword.take().unwrap_or_default().as_ref(), key, bucket, query, None),
(query.keyword.take().unwrap_or_default().as_ref(), key, query, None),
key: [
EMPTY if upload_id => AbortMultipartUpload (query::upload_id),
EMPTY => DeleteObject (query_opt::version_id),
@@ -759,21 +659,17 @@ impl Endpoint {
}
/// Get the kind of authorization which is required to perform the operation.
pub fn authorization_type(&self) -> Authorization<'_> {
let bucket = if let Some(bucket) = self.get_bucket() {
bucket
} else {
pub fn authorization_type(&self) -> Authorization {
if let Endpoint::ListBuckets = self {
return Authorization::None;
};
let readonly = s3_match! {
@extract
@match
self,
bucket,
[
GetBucketAccelerateConfiguration,
GetBucketAcl,
GetBucketAnalyticsConfiguration,
GetBucketCors,
GetBucketEncryption,
GetBucketIntelligentTieringConfiguration,
GetBucketInventoryConfiguration,
@@ -810,41 +706,41 @@ impl Endpoint {
ListParts,
SelectObjectContent,
]
}
.is_some();
};
let owner = s3_match! {
@extract
@match
self,
bucket,
[
DeleteBucket,
GetBucketWebsite,
PutBucketWebsite,
DeleteBucketWebsite,
GetBucketCors,
PutBucketCors,
DeleteBucketCors,
]
}
.is_some();
};
if readonly {
Authorization::Read(bucket)
Authorization::Read
} else if owner {
Authorization::Owner(bucket)
Authorization::Owner
} else {
Authorization::Write(bucket)
Authorization::Write
}
}
}
/// What kind of authorization is required to perform a given action
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Authorization<'a> {
pub enum Authorization {
/// No authorization is required
None,
/// Having Read permission on bucket .0 is required
Read(&'a str),
/// Having Write permission on bucket .0 is required
Write(&'a str),
/// Having Owner permission on bucket .0 is required
Owner(&'a str),
/// Having Read permission on bucket
Read,
/// Having Write permission on bucket
Write,
/// Having Owner permission on bucket
Owner,
}
/// This macro is used to generate part of the code in this module. It must be called only one, and
@@ -985,7 +881,7 @@ mod tests {
uri: &str,
bucket: Option<String>,
header: Option<(&str, &str)>,
) -> Endpoint {
) -> (Endpoint, Option<String>) {
let mut req = Request::builder().method(method).uri(uri);
if let Some((k, v)) = header {
req = req.header(k, v)
@@ -1000,13 +896,13 @@ mod tests {
$(
assert!(
matches!(
parse(test_cases!{@actual_method $method}, $uri, Some("my_bucket".to_owned()), None),
parse(test_cases!{@actual_method $method}, $uri, Some("my_bucket".to_owned()), None).0,
Endpoint::$variant { .. }
)
);
assert!(
matches!(
parse(test_cases!{@actual_method $method}, concat!("/my_bucket", $uri), None, None),
parse(test_cases!{@actual_method $method}, concat!("/my_bucket", $uri), None, None).0,
Endpoint::$variant { .. }
)
);
@@ -1025,78 +921,82 @@ mod tests {
(@actual_method OWNER_DELETE) => {{ "DELETE" }};
(@auth HEAD $uri:expr) => {{
assert_eq!(parse("HEAD", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Read("my_bucket"))
assert_eq!(parse("HEAD", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Read)
}};
(@auth GET $uri:expr) => {{
assert_eq!(parse("GET", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Read("my_bucket"))
assert_eq!(parse("GET", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Read)
}};
(@auth OWNER_GET $uri:expr) => {{
assert_eq!(parse("GET", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Owner("my_bucket"))
assert_eq!(parse("GET", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Owner)
}};
(@auth PUT $uri:expr) => {{
assert_eq!(parse("PUT", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Write("my_bucket"))
assert_eq!(parse("PUT", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Write)
}};
(@auth OWNER_PUT $uri:expr) => {{
assert_eq!(parse("PUT", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Owner("my_bucket"))
assert_eq!(parse("PUT", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Owner)
}};
(@auth POST $uri:expr) => {{
assert_eq!(parse("POST", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Write("my_bucket"))
assert_eq!(parse("POST", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Write)
}};
(@auth DELETE $uri:expr) => {{
assert_eq!(parse("DELETE", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Write("my_bucket"))
assert_eq!(parse("DELETE", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Write)
}};
(@auth OWNER_DELETE $uri:expr) => {{
assert_eq!(parse("DELETE", concat!("/my_bucket", $uri), None, None).authorization_type(),
Authorization::Owner("my_bucket"))
assert_eq!(parse("DELETE", concat!("/my_bucket", $uri), None, None).0.authorization_type(),
Authorization::Owner)
}};
}
#[test]
fn test_bucket_extraction() {
assert_eq!(
parse("GET", "/my/key", Some("my_bucket".to_owned()), None).get_bucket(),
parse("GET", "/my_bucket/my/key", None, None).get_bucket()
parse("GET", "/my/key", Some("my_bucket".to_owned()), None).1,
parse("GET", "/my_bucket/my/key", None, None).1
);
assert_eq!(
parse("GET", "/my_bucket/my/key", None, None)
.get_bucket()
.unwrap(),
parse("GET", "/my_bucket/my/key", None, None).1.unwrap(),
"my_bucket"
);
assert!(parse("GET", "/", None, None).get_bucket().is_none());
assert!(parse("GET", "/", None, None).1.is_none());
}
#[test]
fn test_key() {
assert_eq!(
parse("GET", "/my/key", Some("my_bucket".to_owned()), None).get_key(),
parse("GET", "/my_bucket/my/key", None, None).get_key()
parse("GET", "/my/key", Some("my_bucket".to_owned()), None)
.0
.get_key(),
parse("GET", "/my_bucket/my/key", None, None).0.get_key()
);
assert_eq!(
parse("GET", "/my_bucket/my/key", None, None)
.0
.get_key()
.unwrap(),
"my/key"
);
assert_eq!(
parse("GET", "/my_bucket/my/key?acl", None, None)
.0
.get_key()
.unwrap(),
"my/key"
);
assert!(parse("GET", "/my_bucket/?list-type=2", None, None)
.0
.get_key()
.is_none());
assert_eq!(
parse("GET", "/my_bucket/%26%2B%3F%25%C3%A9/something", None, None)
.0
.get_key()
.unwrap(),
"&+?%é/something"
@@ -1134,7 +1034,7 @@ mod tests {
OWNER_DELETE "/" => DeleteBucket
DELETE "/?analytics&id=list1" => DeleteBucketAnalyticsConfiguration
DELETE "/?analytics&id=Id" => DeleteBucketAnalyticsConfiguration
DELETE "/?cors" => DeleteBucketCors
OWNER_DELETE "/?cors" => DeleteBucketCors
DELETE "/?encryption" => DeleteBucketEncryption
DELETE "/?intelligent-tiering&id=Id" => DeleteBucketIntelligentTieringConfiguration
DELETE "/?inventory&id=list1" => DeleteBucketInventoryConfiguration
@@ -1157,7 +1057,7 @@ mod tests {
GET "/?accelerate" => GetBucketAccelerateConfiguration
GET "/?acl" => GetBucketAcl
GET "/?analytics&id=Id" => GetBucketAnalyticsConfiguration
GET "/?cors" => GetBucketCors
OWNER_GET "/?cors" => GetBucketCors
GET "/?encryption" => GetBucketEncryption
GET "/?intelligent-tiering&id=Id" => GetBucketIntelligentTieringConfiguration
GET "/?inventory&id=list1" => GetBucketInventoryConfiguration
@@ -1233,7 +1133,7 @@ mod tests {
PUT "/?acl" => PutBucketAcl
PUT "/?analytics&id=report1" => PutBucketAnalyticsConfiguration
PUT "/?analytics&id=Id" => PutBucketAnalyticsConfiguration
PUT "/?cors" => PutBucketCors
OWNER_PUT "/?cors" => PutBucketCors
PUT "/?encryption" => PutBucketEncryption
PUT "/?intelligent-tiering&id=Id" => PutBucketIntelligentTieringConfiguration
PUT "/?inventory&id=report1" => PutBucketInventoryConfiguration
@@ -1268,11 +1168,11 @@ mod tests {
);
// no bucket, won't work with the rest of the test suite
assert!(matches!(
parse("GET", "/", None, None),
parse("GET", "/", None, None).0,
Endpoint::ListBuckets { .. }
));
assert!(matches!(
parse("GET", "/", None, None).authorization_type(),
parse("GET", "/", None, None).0.authorization_type(),
Authorization::None
));
@@ -1283,16 +1183,8 @@ mod tests {
"/Key+",
Some("my_bucket".to_owned()),
Some(("x-amz-copy-source", "some/key"))
),
Endpoint::CopyObject { .. }
));
assert!(matches!(
parse(
"PUT",
"/my_bucket/Key+",
None,
Some(("x-amz-copy-source", "some/key"))
),
)
.0,
Endpoint::CopyObject { .. }
));
assert!(matches!(
@@ -1302,8 +1194,19 @@ mod tests {
None,
Some(("x-amz-copy-source", "some/key"))
)
.0,
Endpoint::CopyObject { .. }
));
assert!(matches!(
parse(
"PUT",
"/my_bucket/Key+",
None,
Some(("x-amz-copy-source", "some/key"))
)
.0
.authorization_type(),
Authorization::Write("my_bucket")
Authorization::Write
));
// require a header
@@ -1313,16 +1216,8 @@ mod tests {
"/Key+?partNumber=2&uploadId=UploadId",
Some("my_bucket".to_owned()),
Some(("x-amz-copy-source", "some/key"))
),
Endpoint::UploadPartCopy { .. }
));
assert!(matches!(
parse(
"PUT",
"/my_bucket/Key+?partNumber=2&uploadId=UploadId",
None,
Some(("x-amz-copy-source", "some/key"))
),
)
.0,
Endpoint::UploadPartCopy { .. }
));
assert!(matches!(
@@ -1332,8 +1227,19 @@ mod tests {
None,
Some(("x-amz-copy-source", "some/key"))
)
.0,
Endpoint::UploadPartCopy { .. }
));
assert!(matches!(
parse(
"PUT",
"/my_bucket/Key+?partNumber=2&uploadId=UploadId",
None,
Some(("x-amz-copy-source", "some/key"))
)
.0
.authorization_type(),
Authorization::Write("my_bucket")
Authorization::Write
));
// POST request, but with GET semantic for permissions purpose
@@ -1343,17 +1249,19 @@ mod tests {
"/{Key+}?select&select-type=2",
Some("my_bucket".to_owned()),
None
),
)
.0,
Endpoint::SelectObjectContent { .. }
));
assert!(matches!(
parse("POST", "/my_bucket/{Key+}?select&select-type=2", None, None),
parse("POST", "/my_bucket/{Key+}?select&select-type=2", None, None).0,
Endpoint::SelectObjectContent { .. }
));
assert!(matches!(
parse("POST", "/my_bucket/{Key+}?select&select-type=2", None, None)
.0
.authorization_type(),
Authorization::Read("my_bucket")
Authorization::Read
));
}
}
+15 -16
View File
@@ -5,7 +5,7 @@ use hyper::{Body, Request, Response, StatusCode};
use serde::{Deserialize, Serialize};
use crate::error::*;
use crate::s3_xml::{xmlns_tag, IntValue, Value};
use crate::s3_xml::{to_xml_with_header, xmlns_tag, IntValue, Value};
use crate::signature::verify_signed_content;
use garage_model::bucket_table::*;
@@ -13,16 +13,7 @@ use garage_model::garage::Garage;
use garage_table::*;
use garage_util::data::*;
pub async fn handle_get_website(
garage: Arc<Garage>,
bucket_id: Uuid,
) -> Result<Response<Body>, Error> {
let bucket = garage
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.ok_or(Error::NoSuchBucket)?;
pub async fn handle_get_website(bucket: &Bucket) -> Result<Response<Body>, Error> {
let param = bucket
.params()
.ok_or_internal_error("Bucket should not be deleted at this point")?;
@@ -39,7 +30,7 @@ pub async fn handle_get_website(
redirect_all_requests_to: None,
routing_rules: None,
};
let xml = quick_xml::se::to_string(&wc)?;
let xml = to_xml_with_header(&wc)?;
Ok(Response::builder()
.status(StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/xml")
@@ -80,7 +71,10 @@ pub async fn handle_put_website(
content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> {
let body = hyper::body::to_bytes(req.into_body()).await?;
verify_signed_content(content_sha256, &body[..])?;
if let Some(content_sha256) = content_sha256 {
verify_signed_content(content_sha256, &body[..])?;
}
let mut bucket = garage
.bucket_table
@@ -303,7 +297,7 @@ mod tests {
use quick_xml::de::from_str;
#[test]
fn test_deserialize() {
fn test_deserialize() -> Result<(), Error> {
let message = r#"<?xml version="1.0" encoding="UTF-8"?>
<WebsiteConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<ErrorDocument>
@@ -365,7 +359,12 @@ mod tests {
ref_value,
conf
}
// TODO verify result is ok
// TODO cycle back and verify if ok
let message2 = to_xml_with_header(&ref_value)?;
let cleanup = |c: &str| c.replace(char::is_whitespace, "");
assert_eq!(cleanup(message), cleanup(&message2));
Ok(())
}
}
+127 -6
View File
@@ -16,6 +16,12 @@ pub fn xmlns_tag<S: Serializer>(_v: &(), s: S) -> Result<S::Ok, S::Error> {
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct Value(#[serde(rename = "$value")] pub String);
impl From<&str> for Value {
fn from(s: &str) -> Value {
Value(s.to_string())
}
}
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub struct IntValue(#[serde(rename = "$value")] pub i64);
@@ -27,12 +33,6 @@ pub struct Bucket {
pub name: Value,
}
#[derive(Debug, Serialize, PartialEq)]
pub struct DisplayName(#[serde(rename = "$value")] pub String);
#[derive(Debug, Serialize, PartialEq)]
pub struct Id(#[serde(rename = "$value")] pub String);
#[derive(Debug, Serialize, PartialEq)]
pub struct Owner {
#[serde(rename = "DisplayName")]
@@ -187,6 +187,46 @@ pub struct ListMultipartUploadsResult {
pub encoding_type: Option<Value>,
}
#[derive(Debug, Serialize, PartialEq)]
pub struct PartItem {
#[serde(rename = "ETag")]
pub etag: Value,
#[serde(rename = "LastModified")]
pub last_modified: Value,
#[serde(rename = "PartNumber")]
pub part_number: IntValue,
#[serde(rename = "Size")]
pub size: IntValue,
}
#[derive(Debug, Serialize, PartialEq)]
pub struct ListPartsResult {
#[serde(serialize_with = "xmlns_tag")]
pub xmlns: (),
#[serde(rename = "Bucket")]
pub bucket: Value,
#[serde(rename = "Key")]
pub key: Value,
#[serde(rename = "UploadId")]
pub upload_id: Value,
#[serde(rename = "PartNumberMarker")]
pub part_number_marker: Option<IntValue>,
#[serde(rename = "NextPartNumberMarker")]
pub next_part_number_marker: Option<IntValue>,
#[serde(rename = "MaxParts")]
pub max_parts: IntValue,
#[serde(rename = "IsTruncated")]
pub is_truncated: Value,
#[serde(rename = "Part", default)]
pub parts: Vec<PartItem>,
#[serde(rename = "Initiator")]
pub initiator: Initiator,
#[serde(rename = "Owner")]
pub owner: Owner,
#[serde(rename = "StorageClass")]
pub storage_class: Value,
}
#[derive(Debug, Serialize, PartialEq)]
pub struct ListBucketItem {
#[serde(rename = "Key")]
@@ -706,4 +746,85 @@ mod tests {
);
Ok(())
}
#[test]
fn list_parts() -> Result<(), ApiError> {
let result = ListPartsResult {
xmlns: (),
bucket: Value("example-bucket".to_string()),
key: Value("example-object".to_string()),
upload_id: Value(
"XXBsb2FkIElEIGZvciBlbHZpbmcncyVcdS1tb3ZpZS5tMnRzEEEwbG9hZA".to_string(),
),
part_number_marker: Some(IntValue(1)),
next_part_number_marker: Some(IntValue(3)),
max_parts: IntValue(2),
is_truncated: Value("true".to_string()),
parts: vec![
PartItem {
etag: Value("\"7778aef83f66abc1fa1e8477f296d394\"".to_string()),
last_modified: Value("2010-11-10T20:48:34.000Z".to_string()),
part_number: IntValue(2),
size: IntValue(10485760),
},
PartItem {
etag: Value("\"aaaa18db4cc2f85cedef654fccc4a4x8\"".to_string()),
last_modified: Value("2010-11-10T20:48:33.000Z".to_string()),
part_number: IntValue(3),
size: IntValue(10485760),
},
],
initiator: Initiator {
display_name: Value("umat-user-11116a31-17b5-4fb7-9df5-b288870f11xx".to_string()),
id: Value(
"arn:aws:iam::111122223333:user/some-user-11116a31-17b5-4fb7-9df5-b288870f11xx"
.to_string(),
),
},
owner: Owner {
display_name: Value("someName".to_string()),
id: Value(
"75aa57f09aa0c8caeab4f8c24e99d10f8e7faeebf76c078efc7c6caea54ba06a".to_string(),
),
},
storage_class: Value("STANDARD".to_string()),
};
assert_eq!(
to_xml_with_header(&result)?,
"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
<ListPartsResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
<Bucket>example-bucket</Bucket>\
<Key>example-object</Key>\
<UploadId>XXBsb2FkIElEIGZvciBlbHZpbmcncyVcdS1tb3ZpZS5tMnRzEEEwbG9hZA</UploadId>\
<PartNumberMarker>1</PartNumberMarker>\
<NextPartNumberMarker>3</NextPartNumberMarker>\
<MaxParts>2</MaxParts>\
<IsTruncated>true</IsTruncated>\
<Part>\
<ETag>&quot;7778aef83f66abc1fa1e8477f296d394&quot;</ETag>\
<LastModified>2010-11-10T20:48:34.000Z</LastModified>\
<PartNumber>2</PartNumber>\
<Size>10485760</Size>\
</Part>\
<Part>\
<ETag>&quot;aaaa18db4cc2f85cedef654fccc4a4x8&quot;</ETag>\
<LastModified>2010-11-10T20:48:33.000Z</LastModified>\
<PartNumber>3</PartNumber>\
<Size>10485760</Size>\
</Part>\
<Initiator>\
<DisplayName>umat-user-11116a31-17b5-4fb7-9df5-b288870f11xx</DisplayName>\
<ID>arn:aws:iam::111122223333:user/some-user-11116a31-17b5-4fb7-9df5-b288870f11xx</ID>\
</Initiator>\
<Owner>\
<DisplayName>someName</DisplayName>\
<ID>75aa57f09aa0c8caeab4f8c24e99d10f8e7faeebf76c078efc7c6caea54ba06a</ID>\
</Owner>\
<StorageClass>STANDARD</StorageClass>\
</ListPartsResult>"
);
Ok(())
}
}
+47
View File
@@ -0,0 +1,47 @@
use chrono::{DateTime, Utc};
use hmac::{Hmac, Mac, NewMac};
use sha2::Sha256;
use garage_util::data::{sha256sum, Hash};
use crate::error::*;
pub mod payload;
pub mod streaming;
pub const SHORT_DATE: &str = "%Y%m%d";
pub const LONG_DATETIME: &str = "%Y%m%dT%H%M%SZ";
type HmacSha256 = Hmac<Sha256>;
pub fn verify_signed_content(expected_sha256: Hash, body: &[u8]) -> Result<(), Error> {
if expected_sha256 != sha256sum(body) {
return Err(Error::BadRequest(
"Request content hash does not match signed hash".to_string(),
));
}
Ok(())
}
pub fn signing_hmac(
datetime: &DateTime<Utc>,
secret_key: &str,
region: &str,
service: &str,
) -> Result<HmacSha256, crypto_mac::InvalidKeyLength> {
let secret = String::from("AWS4") + secret_key;
let mut date_hmac = HmacSha256::new_varkey(secret.as_bytes())?;
date_hmac.update(datetime.format(SHORT_DATE).to_string().as_bytes());
let mut region_hmac = HmacSha256::new_varkey(&date_hmac.finalize().into_bytes())?;
region_hmac.update(region.as_bytes());
let mut service_hmac = HmacSha256::new_varkey(&region_hmac.finalize().into_bytes())?;
service_hmac.update(service.as_bytes());
let mut signing_hmac = HmacSha256::new_varkey(&service_hmac.finalize().into_bytes())?;
signing_hmac.update(b"aws4_request");
let hmac = HmacSha256::new_varkey(&signing_hmac.finalize().into_bytes())?;
Ok(hmac)
}
pub fn compute_scope(datetime: &DateTime<Utc>, region: &str) -> String {
format!("{}/{}/s3/aws4_request", datetime.format(SHORT_DATE), region,)
}
@@ -1,28 +1,26 @@
use std::collections::HashMap;
use chrono::{DateTime, Duration, NaiveDateTime, Utc};
use hmac::{Hmac, Mac, NewMac};
use hmac::Mac;
use hyper::{Body, Method, Request};
use sha2::{Digest, Sha256};
use garage_table::*;
use garage_util::data::{sha256sum, Hash};
use garage_util::data::Hash;
use garage_model::garage::Garage;
use garage_model::key_table::*;
use super::signing_hmac;
use super::{LONG_DATETIME, SHORT_DATE};
use crate::encoding::uri_encode;
use crate::error::*;
const SHORT_DATE: &str = "%Y%m%d";
const LONG_DATETIME: &str = "%Y%m%dT%H%M%SZ";
type HmacSha256 = Hmac<Sha256>;
pub async fn check_signature(
pub async fn check_payload_signature(
garage: &Garage,
request: &Request<Body>,
) -> Result<(Key, Option<Hash>), Error> {
) -> Result<(Option<Key>, Option<Hash>), Error> {
let mut headers = HashMap::new();
for (key, val) in request.headers() {
headers.insert(key.to_string(), val.to_str()?.to_string());
@@ -36,24 +34,24 @@ pub async fn check_signature(
let authorization = if let Some(authorization) = headers.get("authorization") {
parse_authorization(authorization, &headers)?
} else if let Some(algorithm) = headers.get("x-amz-algorithm") {
parse_query_authorization(algorithm, &headers)?
} else {
parse_query_authorization(&headers)?
let content_sha256 = headers.get("x-amz-content-sha256");
if let Some(content_sha256) = content_sha256.filter(|c| "UNSIGNED-PAYLOAD" != c.as_str()) {
let sha256 = hex::decode(content_sha256)
.ok()
.and_then(|bytes| Hash::try_from(&bytes))
.ok_or_bad_request("Invalid content sha256 hash")?;
return Ok((None, Some(sha256)));
} else {
return Ok((None, None));
}
};
let date = headers
.get("x-amz-date")
.ok_or_bad_request("Missing X-Amz-Date field")?;
let date: NaiveDateTime =
NaiveDateTime::parse_from_str(date, LONG_DATETIME).ok_or_bad_request("Invalid date")?;
let date: DateTime<Utc> = DateTime::from_utc(date, Utc);
if Utc::now() - date > Duration::hours(24) {
return Err(Error::BadRequest("Date is too old".to_string()));
}
let scope = format!(
"{}/{}/s3/aws4_request",
date.format(SHORT_DATE),
authorization.date.format(SHORT_DATE),
garage.config.s3_api.s3_region
);
if authorization.scope != scope {
@@ -76,10 +74,10 @@ pub async fn check_signature(
&authorization.signed_headers,
&authorization.content_sha256,
);
let string_to_sign = string_to_sign(&date, &scope, &canonical_request);
let string_to_sign = string_to_sign(&authorization.date, &scope, &canonical_request);
let mut hmac = signing_hmac(
&date,
&authorization.date,
&key_p.secret_key,
&garage.config.s3_api.s3_region,
"s3",
@@ -97,16 +95,16 @@ pub async fn check_signature(
let content_sha256 = if authorization.content_sha256 == "UNSIGNED-PAYLOAD" {
None
} else if authorization.content_sha256 == "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" {
let bytes = hex::decode(authorization.signature).ok_or_bad_request("Invalid signature")?;
Some(Hash::try_from(&bytes).ok_or_bad_request("Invalid signature")?)
} else {
let bytes = hex::decode(authorization.content_sha256)
.ok_or_bad_request("Invalid content sha256 hash")?;
Some(
Hash::try_from(&bytes[..])
.ok_or_else(|| Error::BadRequest("Invalid content sha256 hash".to_string()))?,
)
Some(Hash::try_from(&bytes).ok_or_bad_request("Invalid content sha256 hash")?)
};
Ok((key, content_sha256))
Ok((Some(key), content_sha256))
}
struct Authorization {
@@ -115,6 +113,7 @@ struct Authorization {
signed_headers: String,
signature: String,
content_sha256: String,
date: DateTime<Utc>,
}
fn parse_authorization(
@@ -149,6 +148,17 @@ fn parse_authorization(
.get("x-amz-content-sha256")
.ok_or_bad_request("Missing X-Amz-Content-Sha256 field")?;
let date = headers
.get("x-amz-date")
.ok_or_bad_request("Missing X-Amz-Date field")?;
let date: NaiveDateTime =
NaiveDateTime::parse_from_str(date, LONG_DATETIME).ok_or_bad_request("Invalid date")?;
let date: DateTime<Utc> = DateTime::from_utc(date, Utc);
if Utc::now() - date > Duration::hours(24) {
return Err(Error::BadRequest("Date is too old".to_string()));
}
let auth = Authorization {
key_id,
scope,
@@ -161,15 +171,16 @@ fn parse_authorization(
.ok_or_bad_request("Could not find Signature in Authorization field")?
.to_string(),
content_sha256: content_sha256.to_string(),
date,
};
Ok(auth)
}
fn parse_query_authorization(headers: &HashMap<String, String>) -> Result<Authorization, Error> {
let algo = headers
.get("x-amz-algorithm")
.ok_or_bad_request("X-Amz-Algorithm not found in query parameters")?;
if algo != "AWS4-HMAC-SHA256" {
fn parse_query_authorization(
algorithm: &str,
headers: &HashMap<String, String>,
) -> Result<Authorization, Error> {
if algorithm != "AWS4-HMAC-SHA256" {
return Err(Error::BadRequest(
"Unsupported authorization method".to_string(),
));
@@ -190,12 +201,36 @@ fn parse_query_authorization(headers: &HashMap<String, String>) -> Result<Author
.map(|x| x.as_str())
.unwrap_or("UNSIGNED-PAYLOAD");
let duration = headers
.get("x-amz-expires")
.ok_or_bad_request("X-Amz-Expires not found in query parameters")?
.parse()
.map_err(|_| Error::BadRequest("X-Amz-Expires is not a number".to_string()))?;
if duration > 7 * 24 * 3600 {
return Err(Error::BadRequest(
"X-Amz-Exprires may not exceed a week".to_string(),
));
}
let date = headers
.get("x-amz-date")
.ok_or_bad_request("Missing X-Amz-Date field")?;
let date: NaiveDateTime =
NaiveDateTime::parse_from_str(date, LONG_DATETIME).ok_or_bad_request("Invalid date")?;
let date: DateTime<Utc> = DateTime::from_utc(date, Utc);
if Utc::now() - date > Duration::seconds(duration) {
return Err(Error::BadRequest("Date is too old".to_string()));
}
Ok(Authorization {
key_id,
scope,
signed_headers: signed_headers.to_string(),
signature: signature.to_string(),
content_sha256: content_sha256.to_string(),
date,
})
}
@@ -222,25 +257,6 @@ fn string_to_sign(datetime: &DateTime<Utc>, scope_string: &str, canonical_req: &
.join("\n")
}
fn signing_hmac(
datetime: &DateTime<Utc>,
secret_key: &str,
region: &str,
service: &str,
) -> Result<HmacSha256, crypto_mac::InvalidKeyLength> {
let secret = String::from("AWS4") + secret_key;
let mut date_hmac = HmacSha256::new_varkey(secret.as_bytes())?;
date_hmac.update(datetime.format(SHORT_DATE).to_string().as_bytes());
let mut region_hmac = HmacSha256::new_varkey(&date_hmac.finalize().into_bytes())?;
region_hmac.update(region.as_bytes());
let mut service_hmac = HmacSha256::new_varkey(&region_hmac.finalize().into_bytes())?;
service_hmac.update(service.as_bytes());
let mut signing_hmac = HmacSha256::new_varkey(&service_hmac.finalize().into_bytes())?;
signing_hmac.update(b"aws4_request");
let hmac = HmacSha256::new_varkey(&signing_hmac.finalize().into_bytes())?;
Ok(hmac)
}
fn canonical_request(
method: &Method,
url_path: &str,
@@ -288,14 +304,3 @@ fn canonical_query_string(uri: &hyper::Uri) -> String {
"".to_string()
}
}
pub fn verify_signed_content(content_sha256: Option<Hash>, body: &[u8]) -> Result<(), Error> {
let expected_sha256 =
content_sha256.ok_or_bad_request("Request content hash not signed, aborting.")?;
if expected_sha256 != sha256sum(body) {
return Err(Error::BadRequest(
"Request content hash does not match signed hash".to_string(),
));
}
Ok(())
}
+319
View File
@@ -0,0 +1,319 @@
use std::pin::Pin;
use chrono::{DateTime, Utc};
use futures::prelude::*;
use futures::task;
use hyper::body::Bytes;
use garage_util::data::Hash;
use hmac::Mac;
use super::sha256sum;
use super::HmacSha256;
use super::LONG_DATETIME;
use crate::error::*;
/// Result of `sha256("")`
const EMPTY_STRING_HEX_DIGEST: &str =
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
fn compute_streaming_payload_signature(
signing_hmac: &HmacSha256,
date: DateTime<Utc>,
scope: &str,
previous_signature: Hash,
content_sha256: Hash,
) -> Result<Hash, Error> {
let string_to_sign = [
"AWS4-HMAC-SHA256-PAYLOAD",
&date.format(LONG_DATETIME).to_string(),
scope,
&hex::encode(previous_signature),
EMPTY_STRING_HEX_DIGEST,
&hex::encode(content_sha256),
]
.join("\n");
let mut hmac = signing_hmac.clone();
hmac.update(string_to_sign.as_bytes());
Hash::try_from(&hmac.finalize().into_bytes()).ok_or_internal_error("Invalid signature")
}
mod payload {
use garage_util::data::Hash;
pub enum Error<I> {
Parser(nom::error::Error<I>),
BadSignature,
}
impl<I> Error<I> {
pub fn description(&self) -> &str {
match *self {
Error::Parser(ref e) => e.code.description(),
Error::BadSignature => "Bad signature",
}
}
}
#[derive(Debug, Clone)]
pub struct Header {
pub size: usize,
pub signature: Hash,
}
impl Header {
pub fn parse(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
use nom::bytes::streaming::tag;
use nom::character::streaming::hex_digit1;
use nom::combinator::map_res;
use nom::number::streaming::hex_u32;
macro_rules! try_parse {
($expr:expr) => {
$expr.map_err(|e| e.map(Error::Parser))?
};
}
let (input, size) = try_parse!(hex_u32(input));
let (input, _) = try_parse!(tag(";")(input));
let (input, _) = try_parse!(tag("chunk-signature=")(input));
let (input, data) = try_parse!(map_res(hex_digit1, hex::decode)(input));
let signature = Hash::try_from(&data).ok_or(nom::Err::Failure(Error::BadSignature))?;
let (input, _) = try_parse!(tag("\r\n")(input));
let header = Header {
size: size as usize,
signature,
};
Ok((input, header))
}
}
}
#[derive(Debug)]
pub enum SignedPayloadStreamError {
Stream(Error),
InvalidSignature,
Message(String),
}
impl SignedPayloadStreamError {
fn message(msg: &str) -> Self {
SignedPayloadStreamError::Message(msg.into())
}
}
impl From<SignedPayloadStreamError> for Error {
fn from(err: SignedPayloadStreamError) -> Self {
match err {
SignedPayloadStreamError::Stream(e) => e,
SignedPayloadStreamError::InvalidSignature => {
Error::BadRequest("Invalid payload signature".into())
}
SignedPayloadStreamError::Message(e) => {
Error::BadRequest(format!("Chunk format error: {}", e))
}
}
}
}
impl<I> From<payload::Error<I>> for SignedPayloadStreamError {
fn from(err: payload::Error<I>) -> Self {
Self::message(err.description())
}
}
impl<I> From<nom::error::Error<I>> for SignedPayloadStreamError {
fn from(err: nom::error::Error<I>) -> Self {
Self::message(err.code.description())
}
}
struct SignedPayload {
header: payload::Header,
data: Bytes,
}
#[pin_project::pin_project]
pub struct SignedPayloadStream<S>
where
S: Stream<Item = Result<Bytes, Error>>,
{
#[pin]
stream: S,
buf: bytes::BytesMut,
datetime: DateTime<Utc>,
scope: String,
signing_hmac: HmacSha256,
previous_signature: Hash,
}
impl<S> SignedPayloadStream<S>
where
S: Stream<Item = Result<Bytes, Error>>,
{
pub fn new(
stream: S,
signing_hmac: HmacSha256,
datetime: DateTime<Utc>,
scope: &str,
seed_signature: Hash,
) -> Result<Self, Error> {
Ok(Self {
stream,
buf: bytes::BytesMut::new(),
datetime,
scope: scope.into(),
signing_hmac,
previous_signature: seed_signature,
})
}
fn parse_next(input: &[u8]) -> nom::IResult<&[u8], SignedPayload, SignedPayloadStreamError> {
use nom::bytes::streaming::{tag, take};
macro_rules! try_parse {
($expr:expr) => {
$expr.map_err(nom::Err::convert)?
};
}
let (input, header) = try_parse!(payload::Header::parse(input));
// 0-sized chunk is the last
if header.size == 0 {
return Ok((
input,
SignedPayload {
header,
data: Bytes::new(),
},
));
}
let (input, data) = try_parse!(take::<_, _, nom::error::Error<_>>(header.size)(input));
let (input, _) = try_parse!(tag::<_, _, nom::error::Error<_>>("\r\n")(input));
let data = Bytes::from(data.to_vec());
Ok((input, SignedPayload { header, data }))
}
}
impl<S> Stream for SignedPayloadStream<S>
where
S: Stream<Item = Result<Bytes, Error>> + Unpin,
{
type Item = Result<Bytes, SignedPayloadStreamError>;
fn poll_next(
self: Pin<&mut Self>,
cx: &mut task::Context<'_>,
) -> task::Poll<Option<Self::Item>> {
use std::task::Poll;
let mut this = self.project();
loop {
let (input, payload) = match Self::parse_next(this.buf) {
Ok(res) => res,
Err(nom::Err::Incomplete(_)) => {
match futures::ready!(this.stream.as_mut().poll_next(cx)) {
Some(Ok(bytes)) => {
this.buf.extend(bytes);
continue;
}
Some(Err(e)) => {
return Poll::Ready(Some(Err(SignedPayloadStreamError::Stream(e))))
}
None => {
return Poll::Ready(Some(Err(SignedPayloadStreamError::message(
"Unexpected EOF",
))));
}
}
}
Err(nom::Err::Error(e)) | Err(nom::Err::Failure(e)) => {
return Poll::Ready(Some(Err(e)))
}
};
// 0-sized chunk is the last
if payload.data.is_empty() {
return Poll::Ready(None);
}
let data_sha256sum = sha256sum(&payload.data);
let expected_signature = compute_streaming_payload_signature(
this.signing_hmac,
*this.datetime,
this.scope,
*this.previous_signature,
data_sha256sum,
)
.map_err(|e| {
SignedPayloadStreamError::Message(format!("Could not build signature: {}", e))
})?;
if payload.header.signature != expected_signature {
return Poll::Ready(Some(Err(SignedPayloadStreamError::InvalidSignature)));
}
*this.buf = input.into();
*this.previous_signature = payload.header.signature;
return Poll::Ready(Some(Ok(payload.data)));
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.stream.size_hint()
}
}
#[cfg(test)]
mod tests {
use futures::prelude::*;
use super::{SignedPayloadStream, SignedPayloadStreamError};
#[tokio::test]
async fn test_interrupted_signed_payload_stream() {
use chrono::{DateTime, Utc};
use garage_util::data::Hash;
let datetime = DateTime::parse_from_rfc3339("2021-12-13T13:12:42+01:00") // TODO UNIX 0
.unwrap()
.with_timezone(&Utc);
let secret_key = "test";
let region = "test";
let scope = crate::signature::compute_scope(&datetime, region);
let signing_hmac =
crate::signature::signing_hmac(&datetime, secret_key, region, "s3").unwrap();
let data: &[&[u8]] = &[b"1"];
let body = futures::stream::iter(data.iter().map(|block| Ok(block.as_ref().into())));
let seed_signature = Hash::default();
let mut stream =
SignedPayloadStream::new(body, signing_hmac, datetime, &scope, seed_signature).unwrap();
assert!(stream.try_next().await.is_err());
match stream.try_next().await {
Err(SignedPayloadStreamError::Message(msg)) if msg == "Unexpected EOF" => {}
item => panic!(
"Unexpected result, expected early EOF error, got {:?}",
item
),
}
}
}
+24 -6
View File
@@ -27,10 +27,7 @@ pub struct BucketParams {
pub creation_date: u64,
/// Map of key with access to the bucket, and what kind of access they give
pub authorized_keys: crdt::Map<String, BucketKeyPerm>,
/// Whether this bucket is allowed for website access
/// (under all of its global alias names),
/// and if so, the website configuration XML document
pub website_config: crdt::Lww<Option<WebsiteConfig>>,
/// Map of aliases that are or have been given to this bucket
/// in the global namespace
/// (not authoritative: this is just used as an indication to
@@ -40,6 +37,13 @@ pub struct BucketParams {
/// in namespaces local to keys
/// key = (access key id, alias name)
pub local_aliases: crdt::LwwMap<(String, String), bool>,
/// Whether this bucket is allowed for website access
/// (under all of its global alias names),
/// and if so, the website configuration XML document
pub website_config: crdt::Lww<Option<WebsiteConfig>>,
/// CORS rules
pub cors_config: crdt::Lww<Option<Vec<CorsRule>>>,
}
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
@@ -48,15 +52,26 @@ pub struct WebsiteConfig {
pub error_document: Option<String>,
}
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
pub struct CorsRule {
pub id: Option<String>,
pub max_age_seconds: Option<u64>,
pub allow_origins: Vec<String>,
pub allow_methods: Vec<String>,
pub allow_headers: Vec<String>,
pub expose_headers: Vec<String>,
}
impl BucketParams {
/// Create an empty BucketParams with no authorized keys and no website accesss
pub fn new() -> Self {
BucketParams {
creation_date: now_msec(),
authorized_keys: crdt::Map::new(),
website_config: crdt::Lww::new(None),
aliases: crdt::LwwMap::new(),
local_aliases: crdt::LwwMap::new(),
website_config: crdt::Lww::new(None),
cors_config: crdt::Lww::new(None),
}
}
}
@@ -65,9 +80,12 @@ impl Crdt for BucketParams {
fn merge(&mut self, o: &Self) {
self.creation_date = std::cmp::min(self.creation_date, o.creation_date);
self.authorized_keys.merge(&o.authorized_keys);
self.website_config.merge(&o.website_config);
self.aliases.merge(&o.aliases);
self.local_aliases.merge(&o.local_aliases);
self.website_config.merge(&o.website_config);
self.cors_config.merge(&o.cors_config);
}
}
+1 -5
View File
@@ -31,11 +31,7 @@ where
fn ok_or_bad_request<M: AsRef<str>>(self, reason: M) -> Result<T, Error> {
match self {
Ok(x) => Ok(x),
Err(e) => Err(Error::BadRequest(format!(
"{}: {}",
reason.as_ref(),
e.to_string()
))),
Err(e) => Err(Error::BadRequest(format!("{}: {}", reason.as_ref(), e))),
}
}
}
+2 -1
View File
@@ -69,9 +69,10 @@ impl Migrate {
state: Deletable::Present(BucketParams {
creation_date: now_msec(),
authorized_keys: Map::new(),
website_config: Lww::new(website),
aliases: LwwMap::new(),
local_aliases: LwwMap::new(),
website_config: Lww::new(website),
cors_config: Lww::new(None),
}),
})
.await?;
+12
View File
@@ -125,3 +125,15 @@ where
}
}
}
impl<T> Default for Lww<T>
where
T: Default,
{
fn default() -> Self {
Self {
ts: 0,
v: T::default(),
}
}
}
+63 -51
View File
@@ -11,8 +11,9 @@ use hyper::{
use crate::error::*;
use garage_api::error::{Error as ApiError, OkOrBadRequest};
use garage_api::error::{Error as ApiError, OkOrBadRequest, OkOrInternalError};
use garage_api::helpers::{authority_to_host, host_to_bucket};
use garage_api::s3_cors::{add_cors_headers, find_matching_cors_rule, handle_options};
use garage_api::s3_get::{handle_get, handle_head};
use garage_model::garage::Garage;
@@ -132,69 +133,80 @@ async fn serve_file(garage: Arc<Garage>, req: &Request<Body>) -> Result<Response
);
let ret_doc = match *req.method() {
Method::HEAD => handle_head(garage.clone(), req, bucket_id, &key).await,
Method::GET => handle_get(garage.clone(), req, bucket_id, &key).await,
Method::OPTIONS => handle_options(req, &bucket).await,
Method::HEAD => handle_head(garage.clone(), req, bucket_id, &key, None).await,
Method::GET => handle_get(garage.clone(), req, bucket_id, &key, None).await,
_ => Err(ApiError::BadRequest("HTTP method not supported".into())),
}
.map_err(Error::from);
if let Err(error) = ret_doc {
if *req.method() == Method::HEAD || !error.http_status_code().is_client_error() {
// Do not return the error document in the following cases:
// - the error is not a 4xx error code
// - the request is a HEAD method
// In this case we just return the error code and the error message in the body,
// by relying on err_to_res that is called above when we return an Err.
return Err(error);
}
match ret_doc {
Err(error) => {
// For a HEAD or OPTIONS method, and for non-4xx errors,
// we don't return the error document as content,
// we return above and just return the error message
// by relying on err_to_res that is called when we return an Err.
if *req.method() == Method::HEAD
|| *req.method() == Method::OPTIONS
|| !error.http_status_code().is_client_error()
{
return Err(error);
}
// Same if no error document is set: just return the error directly
let error_document = match &website_config.error_document {
Some(ed) => ed.trim_start_matches('/').to_owned(),
None => return Err(error),
};
// If no error document is set: just return the error directly
let error_document = match &website_config.error_document {
Some(ed) => ed.trim_start_matches('/').to_owned(),
None => return Err(error),
};
// We want to return the error document
// Create a fake HTTP request with path = the error document
let req2 = Request::builder()
.uri(format!("http://{}/{}", host, &error_document))
.body(Body::empty())
.unwrap();
// We want to return the error document
// Create a fake HTTP request with path = the error document
let req2 = Request::builder()
.uri(format!("http://{}/{}", host, &error_document))
.body(Body::empty())
.unwrap();
match handle_get(garage, &req2, bucket_id, &error_document).await {
Ok(mut error_doc) => {
// The error won't be logged back in handle_request,
// so log it here
info!(
"{} {} {} {}",
req.method(),
req.uri(),
error.http_status_code(),
error
);
match handle_get(garage, &req2, bucket_id, &error_document, None).await {
Ok(mut error_doc) => {
// The error won't be logged back in handle_request,
// so log it here
info!(
"{} {} {} {}",
req.method(),
req.uri(),
error.http_status_code(),
error
);
*error_doc.status_mut() = error.http_status_code();
error.add_headers(error_doc.headers_mut());
*error_doc.status_mut() = error.http_status_code();
error.add_headers(error_doc.headers_mut());
// Preserve error message in a special header
for error_line in error.to_string().split('\n') {
if let Ok(v) = HeaderValue::from_bytes(error_line.as_bytes()) {
error_doc.headers_mut().append("X-Garage-Error", v);
// Preserve error message in a special header
for error_line in error.to_string().split('\n') {
if let Ok(v) = HeaderValue::from_bytes(error_line.as_bytes()) {
error_doc.headers_mut().append("X-Garage-Error", v);
}
}
}
Ok(error_doc)
}
Err(error_doc_error) => {
warn!(
"Couldn't get error document {} for bucket {:?}: {}",
error_document, bucket_id, error_doc_error
);
Err(error)
Ok(error_doc)
}
Err(error_doc_error) => {
warn!(
"Couldn't get error document {} for bucket {:?}: {}",
error_document, bucket_id, error_doc_error
);
Err(error)
}
}
}
} else {
ret_doc
Ok(mut resp) => {
// Maybe add CORS headers
if let Some(rule) = find_matching_cors_rule(&bucket, req)? {
add_cors_headers(&mut resp, rule)
.ok_or_internal_error("Invalid bucket CORS configuration")?;
}
Ok(resp)
}
}
}