Skip to content

fix: Fixed CORS headers in dataproxy, authors in server #202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lfbrehm merged 2 commits into from
Aug 11, 2025
Merged

fix: Fixed CORS headers in dataproxy, authors in server #202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aee7e1a
fix: Fixed CORS headers in dataproxy, authors in server
lfbrehm Aug 11, 2025
6a79c0e
chore: Removed unneeded whitespaces/tracing
lfbrehm Aug 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions components/data_proxy/src/caching/cache.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,9 +576,9 @@
current_path.push('/');
}
current_path.push_str(name);
if with_intermediates {
final_result.push((*id, current_path.clone()));
} else if x == 2 {

Check warning on line 581 in components/data_proxy/src/caching/cache.rs

View workflow job for this annotation

GitHub Actions / clippy

this `if` has identical blocks 

warning: this `if` has identical blocks
   --> components/data_proxy/src/caching/cache.rs:579:51
    |
579 |   ...                   if with_intermediates {
    |  _____________________________________________^
580 | | ...                       final_result.push((*id, current_path.clone()));
581 | | ...                   } else if x == 2 {
    | |_______________________^
    |
note: same as this
   --> components/data_proxy/src/caching/cache.rs:581:46
    |
581 |   ...                   } else if x == 2 {
    |  ________________________________________^
582 | | ...                       final_result.push((*id, current_path.clone()));
583 | | ...                   }
    | |_______________________^
    = help: for further information visit https://rust-lang.github.io/rust-clippy/master/index.html#if_same_then_else
    = note: `#[warn(clippy::if_same_then_else)]` on by default
final_result.push((*id, current_path.clone()));
}
}
Expand Down Expand Up @@ -776,7 +776,7 @@
#[tracing::instrument(level = "trace", skip(self, object))]
pub async fn insert_temp_resource(&self, object: Object) -> Result<()> {
// Return Ok if already exists
if let Ok((rwlock_object, _)) = self.get_resource(&object.id).await {
if let Ok((rwlock_object, _)) = self.get_resource(&object.id) {
let cache_object = rwlock_object.read().await;
if *cache_object == object {
return Ok(());
Expand Down Expand Up @@ -836,7 +836,7 @@
pub async fn upsert_object(&self, object: Object) -> Result<()> {
trace!(?object, "upserting object");

if let Ok((rwlock_object, _)) = self.get_resource(&object.id).await {
if let Ok((rwlock_object, _)) = self.get_resource(&object.id) {
let cache_object = rwlock_object.read().await;
if *cache_object == object {
return Ok(());
Expand Down Expand Up @@ -998,7 +998,7 @@

#[tracing::instrument(level = "trace", skip(self))]
#[allow(clippy::type_complexity)]
pub async fn get_resource(
pub fn get_resource(
&self,
resource_id: &DieselUlid,
) -> Result<(Arc<RwLock<Object>>, Arc<RwLock<Option<ObjectLocation>>>)> {
Expand All @@ -1015,7 +1015,7 @@
resource_id: &DieselUlid,
skip_location: bool,
) -> Result<(Object, Option<ObjectLocation>)> {
let (obj, loc) = self.get_resource(resource_id).await?;
let (obj, loc) = self.get_resource(resource_id)?;
let obj = obj.read().await.clone();
let loc = if !skip_location {
loc.read().await.clone()
Expand Down Expand Up @@ -1111,7 +1111,7 @@

#[tracing::instrument(level = "trace", skip(self))]
pub async fn get_resource_name(&self, id: &DieselUlid) -> Option<String> {
let (res, _) = self.get_resource(id).await.ok()?;
let (res, _) = self.get_resource(id).ok()?;
let res = res.read().await.name.clone();
Some(res)
}
Expand Down Expand Up @@ -1158,7 +1158,7 @@

#[tracing::instrument(level = "trace", skip(self, id))]
pub async fn get_location_cloned(&self, id: &DieselUlid) -> Option<ObjectLocation> {
let (_, loc) = self.get_resource(id).await.ok()?;
let (_, loc) = self.get_resource(id).ok()?;
let loc = loc.read().await;
loc.clone()
}
Expand Down
56 changes: 41 additions & 15 deletions components/data_proxy/src/caching/grpc_query_handler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,31 +500,57 @@
}

#[tracing::instrument(level = "trace", skip(self, obj, token))]
pub async fn add_or_replace_key_value_project(
pub async fn add_cors_to_project(
&self,
token: &str,
obj: DPObject,
kv: Option<(&str, &str)>,
cors_config: crate::structs::CORSConfiguration,
) -> Result<()> {
let remove_cors = obj
let kv = KeyValue {
key: "app.aruna-storage.org/cors".to_string(),
value: serde_json::to_string(&cors_config)?,
variant: KeyValueVariant::Label as i32,
};

let mut req = Request::new(UpdateProjectKeyValuesRequest {
project_id: obj.id.to_string(),
add_key_values: vec![kv],
remove_key_values: vec![],
});

Self::add_token_to_md(req.metadata_mut(), token)?;

self.project_service
.clone()
.update_project_key_values(req)
.await
.map_err(|e| {
error!(error = ?e, msg = e.to_string());
e
})?;
Ok(())
}

#[tracing::instrument(level = "trace", skip(self, obj, token))]
pub async fn remove_cors_from_project(&self, token: &str, obj: DPObject) -> Result<()> {
let cors = match obj
.key_values
.iter()
.filter(|e| e.key == "app.aruna-storage.org/cors")
.cloned()
.collect();

.next()

Check warning on line 541 in components/data_proxy/src/caching/grpc_query_handler.rs

View workflow job for this annotation

GitHub Actions / clippy

unnecessarily eager cloning of iterator items 

warning: unnecessarily eager cloning of iterator items
   --> components/data_proxy/src/caching/grpc_query_handler.rs:536:26
    |
536 |           let cors = match obj
    |  __________________________^
537 | |             .key_values
538 | |             .iter()
539 | |             .filter(|e| e.key == "app.aruna-storage.org/cors")
540 | |             .cloned()
541 | |             .next()
    | |___________________^
    |
    = help: for further information visit https://rust-lang.github.io/rust-clippy/master/index.html#iter_overeager_cloned
    = note: `#[warn(clippy::iter_overeager_cloned)]` on by default
help: try
    |
539 ~             .filter(|e| e.key == "app.aruna-storage.org/cors")
540 +             .next().cloned()
    |
{
Some(cors) => cors,
None => KeyValue {
key: "app.aruna-storage.org/cors".to_string(),
value: String::new(),
variant: KeyValueVariant::Label as i32,
},
};
let mut req = Request::new(UpdateProjectKeyValuesRequest {
project_id: obj.id.to_string(),
add_key_values: kv
.map(|(k, v)| {
vec![KeyValue {
key: k.to_string(),
value: v.to_string(),
variant: KeyValueVariant::Label as i32,
}]
})
.unwrap_or_default(),
remove_key_values: remove_cors,
add_key_values: vec![],
remove_key_values: vec![cors],
});

Self::add_token_to_md(req.metadata_mut(), token)?;
Expand Down
156 changes: 133 additions & 23 deletions components/data_proxy/src/s3_frontend/s3server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,18 @@ use crate::CORS_REGEX;
use anyhow::Result;
use futures_core::future::BoxFuture;
use futures_util::FutureExt;
use http::header::ACCESS_CONTROL_ALLOW_HEADERS;
use http::header::ACCESS_CONTROL_ALLOW_METHODS;
use http::header::ACCESS_CONTROL_ALLOW_ORIGIN;
use http::header::ORIGIN;
use http::HeaderValue;
use http::Method;
use http::StatusCode;
use hyper::service::Service;
use hyper_util::rt::TokioExecutor;
use hyper_util::rt::TokioIo;
use hyper_util::server::conn::auto::Builder as ConnBuilder;
use s3s::header::HOST;
use s3s::s3_error;
use s3s::service::S3Service;
use s3s::service::S3ServiceBuilder;
Expand All @@ -30,10 +35,14 @@ use tracing::info;
pub struct S3Server {
s3service: S3Service,
address: String,
inner_raw: ArunaS3Service,
}

#[derive(Clone)]
pub struct WrappingService(SharedS3Service);
pub struct WrappingService {
shared: SharedS3Service,
inner: ArunaS3Service,
}

impl S3Server {
#[tracing::instrument(level = "trace", skip(address, hostname, backend, cache))]
Expand All @@ -51,7 +60,7 @@ impl S3Server {
})?;

let service = {
let mut b = S3ServiceBuilder::new(s3service);
let mut b = S3ServiceBuilder::new(s3service.clone());
b.set_base_domain(hostname);
b.set_auth(AuthProvider::new(cache).await);
b.build()
Expand All @@ -60,6 +69,7 @@ impl S3Server {
Ok(Self {
s3service: service,
address: address.into(),
inner_raw: s3service,
})
}
#[tracing::instrument(level = "trace", skip(self))]
Expand All @@ -72,7 +82,10 @@ impl S3Server {

let local_addr = listener.local_addr()?;

let service = WrappingService(self.s3service.into_shared());
let service = WrappingService {
shared: self.s3service.into_shared(),
inner: self.inner_raw.clone(),
};

let connection = ConnBuilder::new(TokioExecutor::new());

Expand Down Expand Up @@ -111,16 +124,18 @@ impl Service<hyper::Request<hyper::body::Incoming>> for WrappingService {
fn call(&self, req: hyper::Request<hyper::body::Incoming>) -> Self::Future {
// Catch OPTIONS requests
if req.method() == Method::OPTIONS {
let resp = Box::pin(async {
hyper::Response::builder()
.header("Access-Control-Allow-Origin", "*")
.header("Access-Control-Allow-Methods", "*")
.header("Access-Control-Allow-Headers", "*")
.body(Body::empty())
.map_err(|_| s3_error!(InvalidRequest, "Invalid OPTIONS request"))
let clone = self.clone();
return Box::pin(async move {
match clone.get_bucket_cors(req).await {
Ok(resp) => Ok(resp),
Err(err) => {
error!("{err}");
hyper::Response::builder()
.body(Body::empty())
.map_err(|_| s3_error!(InvalidRequest, "Invalid OPTIONS request"))
}
}
});

return resp;
}

// Check if response gets CORS header pass
Expand All @@ -132,7 +147,7 @@ impl Service<hyper::Request<hyper::body::Incoming>> for WrappingService {
}
}

let service = self.0.clone();
let service = self.shared.clone();
let resp = service.call(req);
let res = resp.map(move |r| {
r.map(|mut r| {
Expand All @@ -151,15 +166,14 @@ impl Service<hyper::Request<hyper::body::Incoming>> for WrappingService {
// Add CORS * if request origin matches exception regex
if origin_exception {
r.headers_mut()
.append("Access-Control-Allow-Origin", HeaderValue::from_static("*"));
r.headers_mut().append(
"Access-Control-Allow-Methods",
HeaderValue::from_static("*"),
);
r.headers_mut().append(
"Access-Control-Allow-Headers",
HeaderValue::from_static("*"),
);
.entry(ACCESS_CONTROL_ALLOW_ORIGIN)
.or_insert(HeaderValue::from_static("*"));
r.headers_mut()
.entry(ACCESS_CONTROL_ALLOW_METHODS)
.or_insert(HeaderValue::from_static("*"));
r.headers_mut()
.entry(ACCESS_CONTROL_ALLOW_HEADERS)
.or_insert(HeaderValue::from_static("*"));
}

// Workaround to return 206 (Partial Content) for range responses
Expand All @@ -181,7 +195,7 @@ impl Service<hyper::Request<hyper::body::Incoming>> for WrappingService {
impl AsRef<S3Service> for WrappingService {
#[tracing::instrument(level = "trace", skip(self))]
fn as_ref(&self) -> &S3Service {
self.0.as_ref()
self.shared.as_ref()
}
}

Expand All @@ -208,3 +222,99 @@ impl<T, S: Clone> Service<T> for MakeService<S> {
ready(Ok(self.0.clone()))
}
}
impl WrappingService {
async fn get_bucket_cors(
&self,
req: hyper::Request<hyper::body::Incoming>,
) -> Result<hyper::Response<Body>, S3Error> {
let origin = req
.headers()
.get(ORIGIN)
.ok_or_else(|| s3_error!(InvalidURI, "No authority provided in URI"))?
.to_str()
.map_err(|e| {
error!("{e}");
s3_error!(InvalidURI, "Invalid URI encoding")
})?;
let authority = req
.headers()
.get(HOST)
.ok_or_else(|| s3_error!(InvalidURI, "No authority provided in URI"))?;
let bucket = authority
.to_str()
.map_err(|e| {
error!("{e}");
s3_error!(InvalidURI, "Invalid URI encoding")
})?
.split(".")
.next()
.ok_or_else(|| s3_error!(InvalidURI, "No bucket found in URI"))?;
let project_id = self
.inner
.cache
.get_path(bucket)
.ok_or_else(|| s3_error!(NoSuchBucket, "No bucket found with name {}", bucket))?;
let (project, _) = self.inner.cache.get_resource(&project_id).map_err(|e| {
error!("{e}");
s3_error!(NoSuchBucket, "No bucket found with id {}", project_id)
})?;
let cors = project
.read()
.await
.key_values
.clone()
.into_iter()
.find(|kv| kv.key == "app.aruna-storage.org/cors")
.map(|kv| kv.value)
.ok_or_else(|| {
s3_error!(
NoSuchBucketPolicy,
"No cors rules for bucket found {}",
bucket
)
})?;
let cors: crate::structs::CORSConfiguration =
serde_json::from_str(&cors).map_err(|_| {
error!(error = "Unable to deserialize cors from JSON");
s3_error!(InvalidObjectState, "Unable to deserialize cors from JSON")
})?;
let rule = cors
.0
.iter()
.find(|rule| {
rule.allowed_origins.contains(&origin.to_string())
|| rule.allowed_origins.contains(&"*".to_string())
})
.ok_or_else(|| s3_error!(NoSuchBucketPolicy, "No cors policy found for bucket"))?;
let mut builder = hyper::Response::builder();
for origin in rule.allowed_origins.iter() {
builder = builder.header(
ACCESS_CONTROL_ALLOW_ORIGIN,
HeaderValue::from_str(origin).map_err(|e| {
error!("{e}");
s3_error!(InvalidPolicyDocument, "Invalid header name for policy")
})?,
);
}
if let Some(headers) = &rule.allowed_headers {
builder = builder.header(
ACCESS_CONTROL_ALLOW_HEADERS,
HeaderValue::from_str(&headers.join(",")).map_err(|e| {
error!("{e}");
s3_error!(InvalidPolicyDocument, "Invalid header name for policy")
})?,
)
}
let resp = builder
.header(
ACCESS_CONTROL_ALLOW_METHODS,
HeaderValue::from_str(&rule.allowed_methods.join(",")).map_err(|e| {
error!("{e}");
s3_error!(InvalidPolicyDocument, "Invalid header name for policy")
})?,
)
.body(Body::empty())
.map_err(|_| s3_error!(InvalidRequest, "Invalid OPTIONS request"))?;
Ok(resp)
}
}
Loading
Loading