Deuxfleurs/garage
48
56
Fork 73
Code Issues 100 Pull requests 13 Releases 30 Wiki Activity

Support for PostObject #222

Merged
lx merged 14 commits from trinity-1686a/garage:post-object into main 2022-02-21 22:02:31 +00:00
Conversation 1 Commits 14 Files changed 12 +84 -61
2 changed files with 84 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 19ac5ce20f - Show all commits

64
src/api/s3_post_object.rs
View file

@ -39,7 +39,8 @@ pub async fn handle_post_object(
.for_field("file", 5 * 1024 * 1024 * 1024),
);
let mut multipart = Multipart::with_constraints(req.into_body(), boundary, constraints);
let (_head, body) = req.into_parts();
let mut multipart = Multipart::with_constraints(body, boundary, constraints);
let mut params = HeaderMap::new();
while let Some(field) = multipart.next_field().await? {
@ -56,6 +57,7 @@ pub async fn handle_post_object(
params.append("x-amz-acl", content);
}
_ => {
// TODO actually that's illegal to have the same param multiple times
params.append(name, content);
}
}
@ -122,8 +124,11 @@ pub async fn handle_post_object(
let conditions = decoded_policy.into_conditions()?;
for (param_key, value) in params.iter() {
let param_key = param_key.as_str();
if param_key.eq_ignore_ascii_case("content-type") {
let mut param_key = param_key.to_string();
param_key.make_ascii_lowercase();
match param_key.as_str() {
"policy" | "x-amz-signature" => (), // this is always accepted, as it's required to validate other fields
"content-type" => {
for cond in &conditions.content_type {
let ok = match cond {
Operation::Equal(s) => value == s,
@ -138,7 +143,8 @@ pub async fn handle_post_object(
)));
}
}
} else if param_key == "key" {
}
"key" => {
let conds = conditions.params.get("key").ok_or_else(|| {
Error::BadRequest(format!("Key '{}' is not allowed in policy", param_key))
})?;
@ -154,8 +160,9 @@ pub async fn handle_post_object(
)));
}
}
} else {
let conds = conditions.params.get(param_key).ok_or_else(|| {
}
_ => {
let conds = conditions.params.get(&param_key).ok_or_else(|| {
Error::BadRequest(format!("Key '{}' is not allowed in policy", param_key))
})?;
for cond in conds {
@ -172,6 +179,9 @@ pub async fn handle_post_object(
}
}
}
}
// TODO check that each policy item is used
let content_type = field
.content_type()
@ -185,7 +195,7 @@ pub async fn handle_post_object(
let headers = get_headers(&params)?;
let stream = field.map(|r| r.map_err(Into::into));
let res = save_stream(
let (_, md5) = save_stream(
garage,
headers,
StreamLimiter::new(stream, conditions.content_length),
@ -196,35 +206,45 @@ pub async fn handle_post_object(
)
.await?;
let resp = if let Some(target) = params
let etag = format!("\"{}\"", md5);
// TODO get uri
// get Host
// append www-form-urlencoded key
let location = "todo";
let resp = if let Some(mut target) = params
.get("success_action_redirect")
.and_then(|h| h.to_str().ok())
.and_then(|u| url::Url::parse(u).ok())
.filter(|u| u.scheme() == "https" || u.scheme() == "http")
{
target
.query_pairs_mut()
.append_pair("bucket", &bucket)
.append_pair("key", &key)
.append_pair("etag", &etag);
let target = target.to_string();
Response::builder()
.status(StatusCode::SEE_OTHER)
.header(header::LOCATION, target.clone())
.header(header::ETAG, etag)
.body(target.into())?
} else {
let action = params
.get("success_action_status")
.and_then(|h| h.to_str().ok())
.unwrap_or("204");
match action {
"200" => Response::builder()
let builder = Response::builder()
.status(StatusCode::OK)
.body(Body::empty())?,
.header(header::LOCATION, location)
.header(header::ETAG, etag);
match action {
"200" => builder.status(StatusCode::OK).body(Body::empty())?,
"201" => {
// TODO body should be an XML document, not sure which yet
Response::builder()
.status(StatusCode::CREATED)
.body(res.into_body())?
builder.status(StatusCode::CREATED).body(Body::from(""))?
}
_ => Response::builder()
.status(StatusCode::NO_CONTENT)
.body(Body::empty())?,
_ => builder.status(StatusCode::NO_CONTENT).body(Body::empty())?,
}
};
@ -254,8 +274,9 @@ impl Policy {
if map.len() != 1 {
return Err(Error::BadRequest("Invalid policy item".to_owned()));
}
let (k, v) = map.into_iter().next().expect("size was verified");
if k.eq_ignore_ascii_case("content-type") {
let (mut k, v) = map.into_iter().next().expect("size was verified");
k.make_ascii_lowercase();
if k == "content-type" {
content_type.push(Operation::Equal(v));
} else {
params.entry(k).or_default().push(Operation::Equal(v));
@ -265,16 +286,17 @@ impl Policy {
if key.remove(0) != '$' {
return Err(Error::BadRequest("Invalid policy item".to_owned()));
}
key.make_ascii_lowercase();
match cond.as_str() {
"eq" => {
if key.eq_ignore_ascii_case("content-type") {
if key == "content-type" {
content_type.push(Operation::Equal(value));
} else {
params.entry(key).or_default().push(Operation::Equal(value));
}
}
"starts-with" => {
if key.eq_ignore_ascii_case("content-type") {
if key == "content-type" {
content_type.push(Operation::StartsWith(value));
} else {
params

9
src/api/s3_put.rs
View file

@ -99,6 +99,7 @@ pub async fn handle_put(
content_sha256,
)
.await
.map(|(uuid, md5)| put_response(uuid, md5))
}
pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
@ -109,7 +110,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
key: &str,
content_md5: Option<String>,
content_sha256: Option<FixedBytes32>,
) -> Result<Response<Body>, Error> {
) -> Result<(Uuid, String), Error> {
// Generate identity of new version
let version_uuid = gen_uuid();
let version_timestamp = now_msec();
@ -150,7 +151,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
let object = Object::new(bucket_id, key.into(), vec![object_version]);
garage.object_table.insert(&object).await?;
return Ok(put_response(version_uuid, data_md5sum_hex));
return Ok((version_uuid, data_md5sum_hex));
}
// Write version identifier in object table so that we have a trace
@ -216,7 +217,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
let object = Object::new(bucket_id, key.into(), vec![object_version]);
garage.object_table.insert(&object).await?;
Ok(put_response(version_uuid, md5sum_hex))
Ok((version_uuid, md5sum_hex))
}
/// Validate MD5 sum against content-md5 header
@ -512,7 +513,7 @@ pub async fn handle_put_part(
let response = Response::builder()
.header("ETag", format!("\"{}\"", data_md5sum_hex))
.body(Body::from(vec![]))
.body(Body::empty())
.unwrap();
Ok(response)
}