diff --git a/server/src/conditional.rs b/server/src/conditional.rs new file mode 100644 index 0000000000..0f73e446e4 --- /dev/null +++ b/server/src/conditional.rs @@ -0,0 +1,167 @@ +use actix_web::HttpRequest; +use actix_web::http::header::EntityTag; +use actix_web::http::header::Header; +use actix_web::http::header::IfMatch; +use actix_web::http::header::IfNoneMatch; + +#[derive(thiserror::Error, Debug, PartialEq, Eq)] +pub enum ConditionalError { + #[error("Invalid header")] + ParseError, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum ConditionalMatch { + IfMatch(String), + IfNoneMatch(String), +} + +pub fn any_match( + req: &HttpRequest, + etag: Option, +) -> Result, ConditionalError> { + match IfMatch::parse(req) { + Ok(IfMatch::Any) => Ok(Some(etag.is_some())), + Ok(IfMatch::Items(items)) => Ok({ + if items.is_empty() { + None + } else { + Some(etag.map_or(false, |e| items.contains(&e))) + } + }), + Err(_) => Err(ConditionalError::ParseError), + } +} + +pub fn none_match( + req: &HttpRequest, + etag: Option, +) -> Result, ConditionalError> { + match IfNoneMatch::parse(req) { + Ok(IfNoneMatch::Any) => Ok(Some(etag.is_none())), + Ok(IfNoneMatch::Items(items)) => Ok({ + if items.is_empty() { + None + } else { + Some(etag.map_or(true, |e| !items.contains(&e))) + } + }), + Err(_) => Err(ConditionalError::ParseError), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use actix_web::test::TestRequest; + + #[test] + fn test_any_match_none_some() { + assert_eq!( + any_match( + &TestRequest::default().to_http_request(), + Some(EntityTag::new_strong("foo".to_owned())) + ), + Ok(None) + ); + } + + #[test] + fn test_any_match_any_some() { + assert_eq!( + any_match( + &TestRequest::default() + .insert_header(("If-Match", "*")) + .to_http_request(), + Some(EntityTag::new_strong("foo".to_owned())) + ), + Ok(Some(true)) + ); + } + + #[test] + fn test_any_match_some_some() { + assert_eq!( + any_match( + &TestRequest::default() + .insert_header(("If-Match", "\"foo\" ,\"bar\"")) + .to_http_request(), + Some(EntityTag::new_strong("foo".to_owned())) + ), + Ok(Some(true)) + ); + } + + #[test] + fn test_none_match_none_none() { + assert_eq!( + none_match(&TestRequest::default().to_http_request(), None), + Ok(None) + ); + } + + #[test] + fn test_none_match_any_some() { + assert_eq!( + none_match( + &TestRequest::default() + .insert_header(("If-None-Match", "*")) + .to_http_request(), + Some(EntityTag::new_strong("foo".to_owned())) + ), + Ok(Some(false)) + ); + } + + #[test] + fn test_none_match_any_none() { + assert_eq!( + none_match( + &TestRequest::default() + .insert_header(("If-None-Match", "*")) + .to_http_request(), + None + ), + Ok(Some(true)) + ); + } + + #[test] + fn test_none_match_some_some() { + assert_eq!( + none_match( + &TestRequest::default() + .insert_header(("If-None-Match", "\"foo\"")) + .to_http_request(), + Some(EntityTag::new_strong("foo".to_owned())) + ), + Ok(Some(false)) + ); + } + + #[test] + fn test_none_match_some_none() { + assert_eq!( + none_match( + &TestRequest::default() + .insert_header(("If-None-Match", "\"foo\"")) + .to_http_request(), + None + ), + Ok(Some(true)) + ); + } + + #[test] + fn test_none_match_some_unknown() { + assert_eq!( + none_match( + &TestRequest::default() + .insert_header(("If-None-Match", "\"foo\"")) + .to_http_request(), + Some(EntityTag::new_strong("bar".to_owned())) + ), + Ok(Some(true)) + ); + } +} diff --git a/server/src/handlers.rs b/server/src/handlers.rs index 06ff317d20..dbe6666231 100644 --- a/server/src/handlers.rs +++ b/server/src/handlers.rs @@ -6,7 +6,7 @@ use actix_web::{ dev::ServiceRequest, http::{ self, - header::{self, ContentLength, ContentType}, + header::{self, ContentLength, ContentType, EntityTag}, }, web::{Data, Header, Path, Payload}, }; @@ -17,14 +17,22 @@ use size::Size; use tracing::*; use uuid::Uuid; +use crate::conditional; use crate::s3::S3Client; -use crate::{blob, merge}; +use crate::{ + blob, + conditional::{ConditionalMatch, any_match, none_match}, + merge, + postgres::ObjectPart, +}; use crate::{ config::CONFIG, postgres::{self, Pool}, }; use crate::{merge::MergeStrategy, recovery}; +const CACHE_CONTROL: &str = "public, max-age=0, must-revalidate"; + #[derive(Deserialize, Debug)] pub struct ObjectPath { workspace: Uuid, @@ -45,6 +53,12 @@ pub enum ApiError { #[error(transparent)] ActixParseError(#[from] actix_web::error::ParseError), + #[error(transparent)] + ConditionalError(#[from] conditional::ConditionalError), + + #[error("Precondition Failed")] + PreconditionFailed, + #[error(transparent)] Other(#[from] anyhow::Error), } @@ -56,6 +70,12 @@ impl actix_web::error::ResponseError for ApiError { match self { ApiError::ActixError(error) => error.error_response(), + ApiError::ConditionalError(_) => HttpResponse::BadRequest().body("Bad Request"), + + ApiError::PreconditionFailed => { + HttpResponse::PreconditionFailed().body("Precondition Failed") + } + _ => { tracing::error!(error=?self, "Internal error in http handler"); HttpResponse::InternalServerError().body("Internal Server Error") @@ -187,6 +207,10 @@ pub async fn put(request: HttpRequest, payload: Payload) -> HandlerResult>().unwrap().to_owned(); let s3 = request.app_data::>().unwrap().to_owned(); + let parts = postgres::find_parts::(&pool, path.workspace, &path.key).await?; + + let conditionals = validate_put_conditionals(request.request(), &parts)?; + let uploaded = blob::upload(&s3, &pool, headers.content_length, payload).await?; merge::validate_put_body(merge_strategy, &uploaded)?; @@ -213,7 +237,8 @@ pub async fn put(request: HttpRequest, payload: Payload) -> HandlerResult HandlerResult(&pool, path.workspace, &path.key).await?; let mut response = if !parts.is_empty() { - let first = parts.first().unwrap(); - let merge_strategy = first.data.merge_strategy.unwrap(); + let conditionals = validate_patch_conditionals(request.request(), &parts)?; + + let merge_strategy = objectpart_strategy(&parts).unwrap(); merge::validate_patch_request(merge_strategy, &headers)?; @@ -288,7 +314,8 @@ pub async fn patch(request: HttpRequest, payload: Payload) -> HandlerResult>(); - recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts).await?; + + recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts, conditionals).await?; postgres::append_part( &pool, @@ -330,29 +357,38 @@ pub async fn get(request: HttpRequest) -> HandlerResult { span.record("huly_key", &path.key); let pool = request.app_data::>().unwrap().to_owned(); - let s3 = request - .app_data::>() - .unwrap() - .to_owned() - .into_inner(); let parts = postgres::find_parts::(&pool, path.workspace, &path.key).await?; let response = if !parts.is_empty() { - let mut response = HttpResponse::Ok(); + let etag = objectpart_etag(&parts).unwrap(); - let headers = parts[0].data.headers.as_ref(); - if let Some(headers) = headers { - for (header, value) in headers.iter() { - response.insert_header((header.as_str(), value.to_owned())); + match none_match(request.request(), Some(etag.clone()))? { + Some(false) => HttpResponse::NotModified() + .insert_header((header::ETAG, etag)) + .insert_header((header::CACHE_CONTROL, CACHE_CONTROL)) + .finish(), + _ => { + let mut response = HttpResponse::Ok(); + + let s3 = request + .app_data::>() + .unwrap() + .to_owned() + .into_inner(); + + let headers = parts[0].data.headers.as_ref(); + if let Some(headers) = headers { + for (header, value) in headers.iter() { + response.insert_header((header.as_str(), value.to_owned())); + } + } + + response.insert_header((header::ETAG, etag)); + response.insert_header((header::CACHE_CONTROL, CACHE_CONTROL)); + response.body(merge::stream(s3, parts).await?) } } - - let etag = parts.last().unwrap().data.etag.to_owned(); - - response.insert_header((header::ETAG, etag)); - - response.body(merge::stream(s3, parts).await?) } else { HttpResponse::NotFound().finish() }; @@ -376,28 +412,37 @@ pub async fn head(request: HttpRequest) -> HandlerResult { let parts = postgres::find_parts::(&pool, path.workspace, &path.key).await?; let response = if !parts.is_empty() { - let mut response = HttpResponse::Ok(); + let etag = objectpart_etag(&parts).unwrap(); - let headers = parts[0].data.headers.as_ref(); - if let Some(headers) = headers { - for (header, value) in headers.iter() { - response.insert_header((header.as_str(), value.to_owned())); + match none_match(request.request(), Some(etag.clone()))? { + Some(false) => HttpResponse::NotModified() + .insert_header((header::ETAG, etag)) + .insert_header((header::CACHE_CONTROL, CACHE_CONTROL)) + .finish(), + _ => { + let mut response = HttpResponse::Ok(); + + let headers = parts[0].data.headers.as_ref(); + if let Some(headers) = headers { + for (header, value) in headers.iter() { + response.insert_header((header.as_str(), value.to_owned())); + } + } + + response.insert_header((header::ETAG, etag)); + response.insert_header((header::CACHE_CONTROL, CACHE_CONTROL)); + + // see https://github.com/actix/examples/blob/master/forms/multipart-s3/src/main.rs#L67-L79 + let content_length = merge::content_length(parts); + match content_length { + Some(content_length) => response.body(SizedStream::new( + content_length as u64, + stream::empty::>().boxed_local(), + )), + None => response.finish(), + } } } - - let etag = parts.last().unwrap().data.etag.to_owned(); - - response.insert_header((header::ETAG, etag)); - - // see https://github.com/actix/examples/blob/master/forms/multipart-s3/src/main.rs#L67-L79 - let content_length = merge::content_length(parts); - match content_length { - Some(content_length) => response.body(SizedStream::new( - content_length as u64, - stream::empty::>().boxed_local(), - )), - None => response.finish(), - } } else { HttpResponse::NotFound().finish() }; @@ -408,3 +453,52 @@ pub async fn head(request: HttpRequest) -> HandlerResult { pub async fn delete(_path: Path) -> HandlerResult { unimplemented!("delete is not implemented") } + +fn objectpart_etag(parts: &Vec>) -> Option { + parts + .last() + .map(|p| EntityTag::new_strong(p.data.etag.to_owned())) +} + +fn objectpart_strategy(parts: &Vec>) -> Option { + parts.first().map(|p| p.data.merge_strategy.unwrap()) +} + +fn validate_patch_conditionals( + req: &HttpRequest, + parts: &Vec>, +) -> Result, ApiError> { + let etag = objectpart_etag(parts); + + match any_match(req, etag)? { + Some(false) => Err(ApiError::PreconditionFailed), + _ => { + let parts_data = parts.iter().map(|p| &p.data).collect::>(); + let parts_etag = recovery::object_etag(parts_data)?; + + Ok(Some(ConditionalMatch::IfMatch(parts_etag))) + } + } +} + +fn validate_put_conditionals( + req: &HttpRequest, + parts: &Vec>, +) -> Result, ApiError> { + let etag = objectpart_etag(parts); + + match any_match(req, etag.clone())? { + Some(true) => { + let parts_data = parts.iter().map(|p| &p.data).collect::>(); + let parts_etag = recovery::object_etag(parts_data)?; + + Ok(Some(ConditionalMatch::IfMatch(parts_etag))) + } + Some(false) => Err(ApiError::PreconditionFailed), + None => match none_match(req, etag.clone())? { + Some(true) => Ok(Some(ConditionalMatch::IfNoneMatch("*".to_owned()))), + Some(false) => Err(ApiError::PreconditionFailed), + None => Ok(None), + }, + } +} diff --git a/server/src/main.rs b/server/src/main.rs index 29d0335696..fbcb1b3b15 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -15,12 +15,13 @@ use uuid::Uuid; use hulyrs::services::jwt::actix::ServiceRequestExt; mod blob; +mod conditional; mod config; mod handlers; mod merge; -mod recovery; mod patch; mod postgres; +mod recovery; mod s3; use config::CONFIG; diff --git a/server/src/recovery.rs b/server/src/recovery.rs index 0f0c807b32..ec05f0e796 100644 --- a/server/src/recovery.rs +++ b/server/src/recovery.rs @@ -1,26 +1,41 @@ use bytes::Bytes; +use crate::conditional::ConditionalMatch; use crate::config::CONFIG; use crate::{handlers::PartData, s3::S3Client}; +pub fn object_etag(parts: Vec<&PartData>) -> anyhow::Result { + let body = Bytes::from(serde_json::to_string(&parts)?); + let digest = md5::compute(body); + Ok(format!("{:x}", digest)) +} + pub async fn set_object( s3: &S3Client, workspace: uuid::Uuid, key: &str, parts: Vec<&PartData>, + conditions: Option, ) -> anyhow::Result<()> { let s3_bucket = &CONFIG.s3_bucket; let key = format!("blob/{}/{}", workspace, key); let body = Bytes::from(serde_json::to_string(&parts)?); - s3.put_object() + let mut cmd = s3 + .put_object() .bucket(s3_bucket) .key(key) .body(body.into()) - .content_type("application/json") - .send() - .await?; + .content_type("application/json"); + + cmd = match conditions { + Some(ConditionalMatch::IfMatch(etag)) => cmd.if_match(etag), + Some(ConditionalMatch::IfNoneMatch(etag)) => cmd.if_none_match(etag), + None => cmd, + }; + + cmd.send().await?; Ok(()) } diff --git a/tests/src/get.rs b/tests/src/get.rs index 7dbe4292d9..22e5de0f54 100644 --- a/tests/src/get.rs +++ b/tests/src/get.rs @@ -37,3 +37,50 @@ pub async fn get_known() -> eyre::Result<()> { Ok(()) } + +#[tanu::test] +pub async fn get_conditional() -> eyre::Result<()> { + let key = random_key(); + let text = random_text(1024); + + let http = Client::new(); + + let res = http.key_put(&key).body(text.clone()).send().await?; + check!(res.status().is_success()); + + let res = http.key_get(&key).send().await?; + check!(res.status().is_success()); + let etag = res.header("etag").expect("ETag not found"); + + // Test without If-None-Match (normal GET) + let res = http.key_get(&key).send().await?; + check!(res.status().is_success()); + check_eq!(text, res.text().await?); + + // Test with If-None-Match: * + let res = http + .key_get(&key) + .header("If-None-Match", "*") + .send() + .await?; + check_eq!(res.status(), http::StatusCode::NOT_MODIFIED); + + // Test with correct ETag + let res = http + .key_get(&key) + .header("If-None-Match", etag) + .send() + .await?; + check_eq!(res.status(), http::StatusCode::NOT_MODIFIED); + + // Test with incorrect ETag + let res = http + .key_get(&key) + .header("If-None-Match", "\"invalid-etag\"") + .send() + .await?; + check!(res.status().is_success()); + check_eq!(text, res.text().await?); + + Ok(()) +} diff --git a/tests/src/head.rs b/tests/src/head.rs index 53bdf6a927..1cdd808e84 100644 --- a/tests/src/head.rs +++ b/tests/src/head.rs @@ -64,3 +64,48 @@ pub async fn head_known_with_jsonpatch() -> eyre::Result<()> { Ok(()) } + +#[tanu::test] +pub async fn head_conditional() -> eyre::Result<()> { + let key = random_key(); + let text = random_text(1024); + + let http = Client::new(); + + let res = http.key_put(&key).body(text.clone()).send().await?; + check!(res.status().is_success()); + + let res = http.key_head(&key).send().await?; + check!(res.status().is_success()); + let etag = res.header("etag").expect("ETag not found"); + + // Test without If-None-Match (normal GET) + let res = http.key_head(&key).send().await?; + check!(res.status().is_success()); + + // Test with If-None-Match: * + let res = http + .key_head(&key) + .header("If-None-Match", "*") + .send() + .await?; + check_eq!(res.status(), http::StatusCode::NOT_MODIFIED); + + // Test with correct ETag + let res = http + .key_head(&key) + .header("If-None-Match", etag) + .send() + .await?; + check_eq!(res.status(), http::StatusCode::NOT_MODIFIED); + + // Test with incorrect ETag + let res = http + .key_head(&key) + .header("If-None-Match", "\"invalid-etag\"") + .send() + .await?; + check!(res.status().is_success()); + + Ok(()) +} diff --git a/tests/src/patch.rs b/tests/src/patch.rs index 38969c0a53..3a161eb6a0 100644 --- a/tests/src/patch.rs +++ b/tests/src/patch.rs @@ -1,11 +1,11 @@ use hulyrs::StatusCode; use serde_json::{self as json, Value, json}; -use tanu::{check, check_eq, eyre, http::Client}; +use tanu::{check, check_eq, check_ne, eyre, http::Client}; use crate::util::*; #[tanu::test((10, 10))] -pub async fn put_and_patch_contact((initial, patch): (usize, usize)) -> eyre::Result<()> { +pub async fn put_and_patch_content((initial, patch): (usize, usize)) -> eyre::Result<()> { let key = random_key(); let initial = random_text(1024 * initial); let patch = random_text(1024 * patch); @@ -234,3 +234,57 @@ async fn get_json_patch_safe() -> eyre::Result<()> { Ok(()) } + +#[derive(PartialEq, Eq)] +pub enum IfMatch { + ETag, + Some(&'static str), + None, +} + +#[tanu::test(1, IfMatch::None, StatusCode::CREATED)] +#[tanu::test(2, IfMatch::ETag, StatusCode::CREATED)] +#[tanu::test(3, IfMatch::Some("*"), StatusCode::CREATED)] +#[tanu::test(4, IfMatch::Some("\"unknown\""), StatusCode::PRECONDITION_FAILED)] +pub async fn put_and_patch_conditional( + _: usize, + if_match: IfMatch, + status: StatusCode, +) -> eyre::Result<()> { + let key = random_key(); + let initial = random_text(1024); + let patch = random_text(1024); + + let http = Client::new(); + + // create new blob + let res = http.key_put(&key).body(initial.clone()).send().await?; + check!(res.status().is_success()); + + // check content + let res = http.key_get(&key).send().await?; + check!(res.status().is_success()); + let etag = res.header("etag").expect("ETag not found"); + + let mut req = http.key_patch(&key).body(patch.clone()); + + req = match if_match { + IfMatch::ETag => req.header("If-Match", etag), + IfMatch::Some(etag) => req.header("If-Match", etag), + IfMatch::None => req, + }; + + let res = req.send().await?; + check_eq!(res.status(), status); + + let res = http.key_get(&key).send().await?; + check!(res.status().is_success()); + + if status.is_success() { + check_ne!(res.header("etag"), Some(etag)); + } else { + check_eq!(res.header("etag"), Some(etag)); + } + + Ok(()) +} diff --git a/tests/src/put.rs b/tests/src/put.rs index 73a588a4f2..1dab0b0512 100644 --- a/tests/src/put.rs +++ b/tests/src/put.rs @@ -341,3 +341,74 @@ pub async fn put_merge_patch( Ok(()) } + +#[derive(PartialEq, Eq)] +pub enum Condition { + ETag, + IfMatch(&'static str), + IfNoneMatch(&'static str), +} + +#[tanu::test(1, Condition::ETag, StatusCode::CREATED)] +#[tanu::test(2, Condition::IfMatch("*"), StatusCode::PRECONDITION_FAILED)] +#[tanu::test(3, Condition::IfNoneMatch("*"), StatusCode::CREATED)] +#[tanu::test(4, Condition::IfNoneMatch("\"unknown\""), StatusCode::CREATED)] +pub async fn put_conditional_create( + _: usize, + condition: Condition, + status: StatusCode, +) -> eyre::Result<()> { + let key = random_key(); + let body = random_text(1024); + + let http = Client::new(); + + let mut req = http.key_put(&key).body(body.clone()); + + req = match condition { + Condition::ETag => req, + Condition::IfMatch(etag) => req.header("If-Match", etag), + Condition::IfNoneMatch(etag) => req.header("If-None-Match", etag), + }; + + let res = req.send().await?; + check_eq!(res.status(), status); + + Ok(()) +} + +#[tanu::test(1, Condition::ETag, StatusCode::CREATED)] +#[tanu::test(2, Condition::IfMatch("*"), StatusCode::CREATED)] +#[tanu::test(3, Condition::IfNoneMatch("*"), StatusCode::PRECONDITION_FAILED)] +#[tanu::test( + 4, + Condition::IfNoneMatch("\"unknown\""), + StatusCode::INTERNAL_SERVER_ERROR +)] +pub async fn put_conditional_update( + _: usize, + condition: Condition, + status: StatusCode, +) -> eyre::Result<()> { + let key = random_key(); + let body = random_text(1024); + + let http = Client::new(); + + let res = http.key_put(&key).body(body.clone()).send().await?; + check!(res.status().is_success()); + let etag = res.header("etag").expect("ETag not found"); + + let mut req = http.key_put(&key).body(body.clone()); + + req = match condition { + Condition::ETag => req.header("If-Match", etag), + Condition::IfMatch(etag) => req.header("If-Match", etag), + Condition::IfNoneMatch(etag) => req.header("If-None-Match", etag), + }; + + let res = req.send().await?; + check_eq!(res.status(), status); + + Ok(()) +}