aboutsummaryrefslogtreecommitdiff
path: root/src/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/api')
-rw-r--r--src/api/Cargo.toml1
-rw-r--r--src/api/api_server.rs39
-rw-r--r--src/api/s3_delete.rs169
-rw-r--r--src/api/s3_get.rs96
-rw-r--r--src/api/s3_list.rs12
5 files changed, 220 insertions, 97 deletions
diff --git a/src/api/Cargo.toml b/src/api/Cargo.toml
index b8674fba..56e0e2a8 100644
--- a/src/api/Cargo.toml
+++ b/src/api/Cargo.toml
@@ -32,3 +32,4 @@ url = "2.1"
httpdate = "0.3"
percent-encoding = "2.1.0"
roxmltree = "0.11"
+http-range = "0.1"
diff --git a/src/api/api_server.rs b/src/api/api_server.rs
index 913ff0af..32506ccd 100644
--- a/src/api/api_server.rs
+++ b/src/api/api_server.rs
@@ -73,8 +73,7 @@ async fn handler_inner(
req: Request<Body>,
) -> Result<Response<BodyType>, Error> {
let path = req.uri().path().to_string();
- let path = percent_encoding::percent_decode_str(&path)
- .decode_utf8()?;
+ let path = percent_encoding::percent_decode_str(&path).decode_utf8()?;
let (bucket, key) = parse_bucket_key(&path)?;
if bucket.len() == 0 {
@@ -110,7 +109,7 @@ async fn handler_inner(
}
&Method::GET => {
// GetObject query
- Ok(handle_get(garage, &bucket, &key).await?)
+ Ok(handle_get(garage, &req, &bucket, &key).await?)
}
&Method::PUT => {
if params.contains_key(&"partnumber".to_string())
@@ -123,8 +122,8 @@ async fn handler_inner(
} else if req.headers().contains_key("x-amz-copy-source") {
// CopyObject query
let copy_source = req.headers().get("x-amz-copy-source").unwrap().to_str()?;
- let copy_source = percent_encoding::percent_decode_str(&copy_source)
- .decode_utf8()?;
+ let copy_source =
+ percent_encoding::percent_decode_str(&copy_source).decode_utf8()?;
let (source_bucket, source_key) = parse_bucket_key(&copy_source)?;
if !api_key.allow_read(&source_bucket) {
return Err(Error::Forbidden(format!(
@@ -228,22 +227,20 @@ async fn handler_inner(
)
.await?)
}
- &Method::POST => {
- if params.contains_key(&"delete".to_string()) {
- // DeleteObjects
- Ok(handle_delete_objects(garage, bucket, req).await?)
- } else {
- println!(
- "Body: {}",
- std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
- .unwrap_or("<invalid utf8>")
- );
- Err(Error::BadRequest(format!("Unsupported call")))
- }
- }
- _ => {
- Err(Error::BadRequest(format!("Invalid method")))
- }
+ &Method::POST => {
+ if params.contains_key(&"delete".to_string()) {
+ // DeleteObjects
+ Ok(handle_delete_objects(garage, bucket, req).await?)
+ } else {
+ println!(
+ "Body: {}",
+ std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
+ .unwrap_or("<invalid utf8>")
+ );
+ Err(Error::BadRequest(format!("Unsupported call")))
+ }
+ }
+ _ => Err(Error::BadRequest(format!("Invalid method"))),
}
}
}
diff --git a/src/api/s3_delete.rs b/src/api/s3_delete.rs
index 001eb162..e77ab314 100644
--- a/src/api/s3_delete.rs
+++ b/src/api/s3_delete.rs
@@ -1,5 +1,5 @@
-use std::sync::Arc;
use std::fmt::Write;
+use std::sync::Arc;
use hyper::{Body, Request, Response};
@@ -9,10 +9,14 @@ use garage_util::error::Error;
use garage_core::garage::Garage;
use garage_core::object_table::*;
-use crate::http_util::*;
use crate::encoding::*;
+use crate::http_util::*;
-async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Result<(UUID, UUID), Error> {
+async fn handle_delete_internal(
+ garage: &Garage,
+ bucket: &str,
+ key: &str,
+) -> Result<(UUID, UUID), Error> {
let object = match garage
.object_table
.get(&bucket.to_string(), &key.to_string())
@@ -32,16 +36,16 @@ async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Res
let mut must_delete = None;
let mut timestamp = now_msec();
for v in interesting_versions {
- if v.timestamp + 1 > timestamp || must_delete.is_none() {
- must_delete = Some(v.uuid);
- }
+ if v.timestamp + 1 > timestamp || must_delete.is_none() {
+ must_delete = Some(v.uuid);
+ }
timestamp = std::cmp::max(timestamp, v.timestamp + 1);
}
- let deleted_version = match must_delete {
- None => return Err(Error::NotFound),
- Some(v) => v,
- };
+ let deleted_version = match must_delete {
+ None => return Err(Error::NotFound),
+ Some(v) => v,
+ };
let version_uuid = gen_uuid();
@@ -62,8 +66,13 @@ async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Res
return Ok((deleted_version, version_uuid));
}
-pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<Response<BodyType>, Error> {
- let (_deleted_version, delete_marker_version) = handle_delete_internal(&garage, bucket, key).await?;
+pub async fn handle_delete(
+ garage: Arc<Garage>,
+ bucket: &str,
+ key: &str,
+) -> Result<Response<BodyType>, Error> {
+ let (_deleted_version, delete_marker_version) =
+ handle_delete_internal(&garage, bucket, key).await?;
Ok(Response::builder()
.header("x-amz-version-id", hex::encode(delete_marker_version))
@@ -71,76 +80,98 @@ pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Resu
.unwrap())
}
-pub async fn handle_delete_objects(garage: Arc<Garage>, bucket: &str, req: Request<Body>) -> Result<Response<BodyType>, Error> {
- let body = hyper::body::to_bytes(req.into_body()).await?;
- let cmd_xml = roxmltree::Document::parse(&std::str::from_utf8(&body)?)?;
- let cmd = parse_delete_objects_xml(&cmd_xml)
- .map_err(|e| Error::BadRequest(format!("Invald delete XML query: {}", e)))?;
-
- let mut retxml = String::new();
+pub async fn handle_delete_objects(
+ garage: Arc<Garage>,
+ bucket: &str,
+ req: Request<Body>,
+) -> Result<Response<BodyType>, Error> {
+ let body = hyper::body::to_bytes(req.into_body()).await?;
+ let cmd_xml = roxmltree::Document::parse(&std::str::from_utf8(&body)?)?;
+ let cmd = parse_delete_objects_xml(&cmd_xml)
+ .map_err(|e| Error::BadRequest(format!("Invald delete XML query: {}", e)))?;
+
+ let mut retxml = String::new();
writeln!(&mut retxml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
writeln!(&mut retxml, "<DeleteObjectsOutput>").unwrap();
- for obj in cmd.objects.iter() {
- match handle_delete_internal(&garage, bucket, &obj.key).await {
- Ok((deleted_version, delete_marker_version)) => {
- writeln!(&mut retxml, "\t<Deleted>").unwrap();
- writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
- writeln!(&mut retxml, "\t\t<VersionId>{}</VersionId>", hex::encode(deleted_version)).unwrap();
- writeln!(&mut retxml, "\t\t<DeleteMarkerVersionId>{}</DeleteMarkerVersionId>", hex::encode(delete_marker_version)).unwrap();
- writeln!(&mut retxml, "\t</Deleted>").unwrap();
- }
- Err(e) => {
- writeln!(&mut retxml, "\t<Error>").unwrap();
- writeln!(&mut retxml, "\t\t<Code>{}</Code>", e.http_status_code()).unwrap();
- writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
- writeln!(&mut retxml, "\t\t<Message>{}</Message>", xml_escape(&format!("{}", e))).unwrap();
- writeln!(&mut retxml, "\t</Error>").unwrap();
- }
- }
- }
+ for obj in cmd.objects.iter() {
+ match handle_delete_internal(&garage, bucket, &obj.key).await {
+ Ok((deleted_version, delete_marker_version)) => {
+ writeln!(&mut retxml, "\t<Deleted>").unwrap();
+ writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+ writeln!(
+ &mut retxml,
+ "\t\t<VersionId>{}</VersionId>",
+ hex::encode(deleted_version)
+ )
+ .unwrap();
+ writeln!(
+ &mut retxml,
+ "\t\t<DeleteMarkerVersionId>{}</DeleteMarkerVersionId>",
+ hex::encode(delete_marker_version)
+ )
+ .unwrap();
+ writeln!(&mut retxml, "\t</Deleted>").unwrap();
+ }
+ Err(e) => {
+ writeln!(&mut retxml, "\t<Error>").unwrap();
+ writeln!(&mut retxml, "\t\t<Code>{}</Code>", e.http_status_code()).unwrap();
+ writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+ writeln!(
+ &mut retxml,
+ "\t\t<Message>{}</Message>",
+ xml_escape(&format!("{}", e))
+ )
+ .unwrap();
+ writeln!(&mut retxml, "\t</Error>").unwrap();
+ }
+ }
+ }
writeln!(&mut retxml, "</DeleteObjectsOutput>").unwrap();
- Ok(Response::new(Box::new(BytesBody::from(retxml.into_bytes()))))
+ Ok(Response::new(Box::new(BytesBody::from(
+ retxml.into_bytes(),
+ ))))
}
struct DeleteRequest {
- objects: Vec<DeleteObject>,
+ objects: Vec<DeleteObject>,
}
struct DeleteObject {
- key: String,
+ key: String,
}
fn parse_delete_objects_xml(xml: &roxmltree::Document) -> Result<DeleteRequest, String> {
- let mut ret = DeleteRequest{objects: vec![]};
-
- let root = xml.root();
- let delete = match root.first_child() {
- Some(del) => del,
- None => return Err(format!("Delete tag not found")),
- };
- if !delete.has_tag_name("Delete") {
- return Err(format!("Invalid root tag: {:?}", root));
- }
-
- for item in delete.children() {
- if item.has_tag_name("Object") {
- if let Some(key) = item.children().find(|e| e.has_tag_name("Key")) {
- if let Some(key_str) = key.text() {
- ret.objects.push(DeleteObject{key: key_str.to_string()});
- } else {
- return Err(format!("No text for key: {:?}", key));
- }
- } else {
- return Err(format!("No delete key for item: {:?}", item));
- }
- } else {
- return Err(format!("Invalid delete item: {:?}", item));
- }
- }
-
- Ok(ret)
-}
+ let mut ret = DeleteRequest { objects: vec![] };
+
+ let root = xml.root();
+ let delete = match root.first_child() {
+ Some(del) => del,
+ None => return Err(format!("Delete tag not found")),
+ };
+ if !delete.has_tag_name("Delete") {
+ return Err(format!("Invalid root tag: {:?}", root));
+ }
+ for item in delete.children() {
+ if item.has_tag_name("Object") {
+ if let Some(key) = item.children().find(|e| e.has_tag_name("Key")) {
+ if let Some(key_str) = key.text() {
+ ret.objects.push(DeleteObject {
+ key: key_str.to_string(),
+ });
+ } else {
+ return Err(format!("No text for key: {:?}", key));
+ }
+ } else {
+ return Err(format!("No delete key for item: {:?}", item));
+ }
+ } else {
+ return Err(format!("Invalid delete item: {:?}", item));
+ }
+ }
+
+ Ok(ret)
+}
diff --git a/src/api/s3_get.rs b/src/api/s3_get.rs
index 75478e46..3ed0f914 100644
--- a/src/api/s3_get.rs
+++ b/src/api/s3_get.rs
@@ -3,7 +3,7 @@ use std::time::{Duration, UNIX_EPOCH};
use futures::stream::*;
use hyper::body::Bytes;
-use hyper::{Response, StatusCode};
+use hyper::{Body, Request, Response, StatusCode};
use garage_util::error::Error;
@@ -22,6 +22,7 @@ fn object_headers(version: &ObjectVersion) -> http::response::Builder {
.header("Content-Type", version.mime_type.to_string())
.header("Content-Length", format!("{}", version.size))
.header("Last-Modified", date_str)
+ .header("Accept-Ranges", format!("bytes"))
}
pub async fn handle_head(
@@ -59,6 +60,7 @@ pub async fn handle_head(
pub async fn handle_get(
garage: Arc<Garage>,
+ req: &Request<Body>,
bucket: &str,
key: &str,
) -> Result<Response<BodyType>, Error> {
@@ -82,6 +84,25 @@ pub async fn handle_get(
None => return Err(Error::NotFound),
};
+ let range = match req.headers().get("range") {
+ Some(range) => {
+ let range_str = range
+ .to_str()
+ .map_err(|e| Error::BadRequest(format!("Invalid range header: {}", e)))?;
+ let mut ranges = http_range::HttpRange::parse(range_str, last_v.size)
+ .map_err(|_e| Error::BadRequest(format!("Invalid range")))?;
+ if ranges.len() > 1 {
+ return Err(Error::BadRequest(format!("Multiple ranges not supported")));
+ } else {
+ ranges.pop()
+ }
+ }
+ None => None,
+ };
+ if let Some(range) = range {
+ return handle_get_range(garage, last_v, range.start, range.start + range.length).await;
+ }
+
let resp_builder = object_headers(&last_v).status(StatusCode::OK);
match &last_v.data {
@@ -131,3 +152,76 @@ pub async fn handle_get(
}
}
}
+
+pub async fn handle_get_range(
+ garage: Arc<Garage>,
+ version: &ObjectVersion,
+ begin: u64,
+ end: u64,
+) -> Result<Response<BodyType>, Error> {
+ if end > version.size {
+ return Err(Error::BadRequest(format!("Range not included in file")));
+ }
+
+ let resp_builder = object_headers(&version)
+ .header(
+ "Content-Range",
+ format!("bytes {}-{}/{}", begin, end, version.size),
+ )
+ .status(StatusCode::PARTIAL_CONTENT);
+
+ match &version.data {
+ ObjectVersionData::Uploading => Err(Error::Message(format!(
+ "Version is_complete() but data is stil Uploading (internal error)"
+ ))),
+ ObjectVersionData::DeleteMarker => Err(Error::NotFound),
+ ObjectVersionData::Inline(bytes) => {
+ if end as usize <= bytes.len() {
+ let body: BodyType = Box::new(BytesBody::from(
+ bytes[begin as usize..end as usize].to_vec(),
+ ));
+ Ok(resp_builder.body(body)?)
+ } else {
+ Err(Error::Message(format!("Internal error: requested range not present in inline bytes when it should have been")))
+ }
+ }
+ ObjectVersionData::FirstBlock(_first_block_hash) => {
+ let version = garage.version_table.get(&version.uuid, &EmptyKey).await?;
+ let version = match version {
+ Some(v) => v,
+ None => return Err(Error::NotFound),
+ };
+
+ let blocks = version
+ .blocks()
+ .iter()
+ .cloned()
+ .filter(|block| block.offset + block.size > begin && block.offset < end)
+ .collect::<Vec<_>>();
+
+ let body_stream = futures::stream::iter(blocks)
+ .map(move |block| {
+ let garage = garage.clone();
+ async move {
+ let data = garage.block_manager.rpc_get_block(&block.hash).await?;
+ let start_in_block = if block.offset > begin {
+ 0
+ } else {
+ begin - block.offset
+ };
+ let end_in_block = if block.offset + block.size < end {
+ block.size
+ } else {
+ end - block.offset
+ };
+ Ok(Bytes::from(
+ data[start_in_block as usize..end_in_block as usize].to_vec(),
+ ))
+ }
+ })
+ .buffered(2);
+ let body: BodyType = Box::new(StreamBody::new(Box::pin(body_stream)));
+ Ok(resp_builder.body(body)?)
+ }
+ }
+}
diff --git a/src/api/s3_list.rs b/src/api/s3_list.rs
index d4d8161e..b8babbbf 100644
--- a/src/api/s3_list.rs
+++ b/src/api/s3_list.rs
@@ -58,10 +58,10 @@ pub async fn handle_list(
break 'query_loop;
}
if let Some(version) = object.versions().iter().find(|x| x.is_data()) {
- if result_keys.len() + result_common_prefixes.len() >= max_keys {
- truncated = true;
- break 'query_loop;
- }
+ if result_keys.len() + result_common_prefixes.len() >= max_keys {
+ truncated = true;
+ break 'query_loop;
+ }
let common_prefix = if delimiter.len() > 0 {
let relative_key = &object.key[prefix.len()..];
match relative_key.find(delimiter) {
@@ -88,8 +88,8 @@ pub async fn handle_list(
}
}
if objects.len() < max_keys + 1 {
- truncated = false;
- break 'query_loop;
+ truncated = false;
+ break 'query_loop;
}
if objects.len() > 0 {
next_chunk_start = objects[objects.len() - 1].key.clone();
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-06 16:28:52 +0000