merge: Merge pull request #391 from arunaengine/fix/blob-hardening

[fix] Blob layer hardening: framing, streams, finalization, path confinement

Author: St4NNi

README.md

@@ -45,6 +45,9 @@ The system is organized around **realms**. A realm is an organizational trust bo
 ### Data, metadata, and access
 
 Each Aruna node exposes an **S3-compatible API**, so researchers can keep using the tools, scripts, workflow systems, and libraries they already have instead of learning a new storage protocol. Buckets are virtual collections that mix local data, replicated data, and references to remote resources. To a user, this looks like one coherent access point. Underneath, Aruna tracks where data actually lives, which permissions apply, and whether an object should be materialized locally or fetched on demand.
+
+> [!NOTE]
+> Object keys for `PutObject`, `CreateMultipartUpload`, `UploadPart`, and `CompleteMultipartUpload` must be non-empty relative paths; they are rejected if they begin with `/`, contain an exact `..` path segment, or contain control characters.
  
 Metadata is part of the core system, not an external catalog bolted on afterwards. Descriptions are stored as **RO-Crate JSON-LD**, so datasets, files, people, instruments, workflows, software, and process runs can be described in a shared format. These descriptions live in a CRDT-based triple store, which allows concurrent edits on different nodes and merges them without a single authority arbitrating the result. Management resources such as users and groups are synchronized through durable document-sync topics, which lets nodes keep working through network outages and reconcile state once they reconnect.
 

api/src/s3/s3_service.rs

@@ -9,7 +9,7 @@ use crate::s3::error::IntoS3Error;
 use crate::s3::util::{
     convert_input, multipart_checksum_type_from_s3, parse_completed_part,
     parse_multipart_checksum_hint, parse_multipart_part_number, parse_upload_id, parse_version_id,
-    s3_checksum_type_from_multipart,
+    s3_checksum_type_from_multipart, validate_object_key,
 };
 use aruna_core::NodeId;
 use aruna_core::stream::{BackendStream, StreamError};
@@ -905,6 +905,7 @@ impl S3 for ArunaS3Service {
             error!(error = "Missing user context");
             s3_error!(UnexpectedContent, "Missing user context")
         })?;
+        validate_object_key(&req.input.key)?;
         let bucket_info = req.extensions.get::<BucketInfo>().cloned();
         let checksum_request = parse_upload_checksum_request(&req.headers)?;
         let replication_auth = AuthContext {
@@ -964,6 +965,7 @@ impl S3 for ArunaS3Service {
             error!(error = "Missing user context");
             s3_error!(UnexpectedContent, "Missing user context")
         })?;
+        validate_object_key(&req.input.key)?;
         let bucket_info = req.extensions.get::<BucketInfo>().cloned();
         let checksum_hint = parse_multipart_checksum_hint(&req.input)?;
 
@@ -1007,6 +1009,7 @@ impl S3 for ArunaS3Service {
             error!(error = "Missing user context");
             s3_error!(UnexpectedContent, "Missing user context")
         })?;
+        validate_object_key(&req.input.key)?;
         let checksum_request = parse_upload_checksum_request(&req.headers)?;
         let upload_id = parse_upload_id(&req.input.upload_id)?;
         let body = req
@@ -1068,6 +1071,7 @@ impl S3 for ArunaS3Service {
             error!(error = "Missing user context");
             s3_error!(UnexpectedContent, "Missing user context")
         })?;
+        validate_object_key(&req.input.key)?;
         let bucket_info = req.extensions.get::<BucketInfo>().cloned();
         let group_id = bucket_info
             .as_ref()

api/src/s3/util.rs

@@ -1,7 +1,9 @@
 use crate::s3::auth::Action;
 use aruna_core::stream::BackendStream;
 use aruna_core::structs::checksum::{ChecksumAlgorithm, ExpectedChecksum};
-use aruna_core::structs::{MultipartChecksumType, MultipartUploadChecksumHint};
+use aruna_core::structs::{
+    MultipartChecksumType, MultipartUploadChecksumHint, ensure_confined_relative_path,
+};
 use aruna_operations::s3::complete_multipart_upload::CompleteMultipartPart;
 use aruna_operations::s3::put_object::PutObjectInput as BlobPutObjectInput;
 use base64::prelude::*;
@@ -10,6 +12,7 @@ use s3s::dto::{
     ChecksumType, CompletedPart, CreateMultipartUploadInput, PartNumber, PutObjectInput,
 };
 use s3s::{S3Error, S3ErrorCode, S3Result, s3_error};
+use std::path::Path;
 use ulid::Ulid;
 
 pub fn get_s3_operation_permission(operation_name: &str) -> Option<Action> {
@@ -123,6 +126,15 @@ pub(crate) fn is_anonymous_object_read_operation(operation_name: &str) -> bool {
     matches!(operation_name, "GetObject")
 }
 
+pub(crate) fn validate_object_key(key: &str) -> S3Result<()> {
+    if key.is_empty() {
+        return Err(s3_error!(InvalidArgument, "Object key must not be empty"));
+    }
+
+    ensure_confined_relative_path(Path::new(key))
+        .map_err(|err| s3_error!(InvalidArgument, "{}", err.to_string()))
+}
+
 pub(crate) fn convert_input(mut input: PutObjectInput) -> S3Result<BlobPutObjectInput, S3Error> {
     match input.body.take() {
         None => Err(s3_error!(InvalidRequest, "Missing body")),
@@ -264,11 +276,43 @@ pub(crate) fn checksum_algorithm_from_s3(
 mod tests {
     use super::{
         get_s3_operation_permission, is_anonymous_object_read_operation,
-        parse_multipart_part_number,
+        parse_multipart_part_number, validate_object_key,
     };
     use crate::s3::auth::Action;
     use s3s::S3ErrorCode;
 
+    #[test]
+    fn validate_object_key_accepts_ordinary_keys() {
+        for key in ["object.bin", "nested/path/object.bin", "a.b..c/keep..dots"] {
+            assert!(
+                validate_object_key(key).is_ok(),
+                "key {key:?} must be valid"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_object_key_rejects_traversal_and_control_keys() {
+        let cases = [
+            "",
+            "/absolute/key",
+            "../escape",
+            "../../etc/passwd",
+            "nested/../../escape",
+            "trailing/..",
+            "with\u{0000}null",
+            "with\u{007f}delete",
+            "with\nnewline",
+        ];
+        for key in cases {
+            assert_eq!(
+                *validate_object_key(key).unwrap_err().code(),
+                S3ErrorCode::InvalidArgument,
+                "key {key:?} must be rejected"
+            );
+        }
+    }
+
     #[test]
     fn anonymous_public_access_only_allows_get_object() {
         assert!(is_anonymous_object_read_operation("GetObject"));

blob/src/bao_tree.rs

@@ -1,10 +1,11 @@
 use crate::error::BlobLibError;
 use crate::hash::Hasher;
+use crate::opendal::abort_partial_writer;
 use aruna_net::streams::{RecvStream, SendStream};
 use bytes::Bytes;
-use futures::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+use futures::{AsyncReadExt, AsyncSeekExt};
 use iroh_io::{AsyncSliceReader, AsyncSliceWriter, AsyncStreamReader, AsyncStreamWriter};
-use opendal::{FuturesAsyncReader, FuturesAsyncWriter, Operator};
+use opendal::{FuturesAsyncReader, Operator};
 use std::{future::Future, io, time::Duration};
 use tokio::io::AsyncWriteExt as TokioAsyncWriteExt;
 use tokio::time::timeout;
@@ -147,47 +148,46 @@ impl AsyncStreamReader for RecvStreamWrapper<'_> {
     }
 }
 
-// ----- FuturesAsyncReader/FuturesAsyncWriter impls for bao_tree ----------
+// ----- FuturesAsyncReader/opendal Writer impls for bao_tree ----------
 pub struct OpenDalWriter {
-    pub writer: FuturesAsyncWriter,
-    pub len: u64,
+    writer: opendal::Writer,
+    operator: Operator,
+    storage_path: String,
     pub hasher: Hasher,
-    pub idle_timeout: Duration,
+    idle_timeout: Duration,
 }
 
 impl AsyncSliceWriter for OpenDalWriter {
     async fn write_at(&mut self, offset: u64, data: &[u8]) -> std::io::Result<()> {
         debug!("[OpenDalWriter] Try to write data with offset {}", offset);
         with_transfer_idle_timeout(
-            self.writer.write_all(data),
+            async {
+                self.writer
+                    .write(data.to_vec())
+                    .await
+                    .map_err(io::Error::from)
+            },
             self.idle_timeout,
             "writing replicated chunk to backend storage",
         )
         .await?;
-        with_transfer_idle_timeout(
-            self.writer.flush(),
-            self.idle_timeout,
-            "flushing replicated chunk to backend storage",
-        )
-        .await?;
         self.hasher.update(data);
         Ok(())
     }
 
     async fn write_bytes_at(&mut self, offset: u64, data: Bytes) -> std::io::Result<()> {
         debug!("[OpenDalWriter] Try to write bytes with offset {}", offset);
         with_transfer_idle_timeout(
-            self.writer.write_all(&data),
+            async {
+                self.writer
+                    .write(data.clone())
+                    .await
+                    .map_err(io::Error::from)
+            },
             self.idle_timeout,
             "writing replicated chunk to backend storage",
         )
         .await?;
-        with_transfer_idle_timeout(
-            self.writer.flush(),
-            self.idle_timeout,
-            "flushing replicated chunk to backend storage",
-        )
-        .await?;
         self.hasher.update(&data);
         Ok(())
     }
@@ -198,12 +198,6 @@ impl AsyncSliceWriter for OpenDalWriter {
     }
 
     async fn sync(&mut self) -> std::io::Result<()> {
-        with_transfer_idle_timeout(
-            self.writer.flush(),
-            self.idle_timeout,
-            "flushing replicated chunk to backend storage",
-        )
-        .await?;
         Ok(())
     }
 }
@@ -212,19 +206,40 @@ impl OpenDalWriter {
     pub async fn new(
         operator: &Operator,
         storage_path: &str,
-        blob_size: u64,
         idle_timeout: Duration,
     ) -> Result<Self, BlobLibError> {
         Ok(Self {
-            writer: operator
-                .writer(storage_path)
-                .await?
-                .into_futures_async_write(),
-            len: blob_size,
+            writer: operator.writer(storage_path).await?,
+            operator: operator.clone(),
+            storage_path: storage_path.to_string(),
             hasher: Hasher::new(),
             idle_timeout,
         })
     }
+
+    pub async fn finalize(mut self) -> std::io::Result<()> {
+        let close_result = with_transfer_idle_timeout(
+            async {
+                self.writer
+                    .close()
+                    .await
+                    .map(|_| ())
+                    .map_err(io::Error::from)
+            },
+            self.idle_timeout,
+            "finalizing replicated blob in backend storage",
+        )
+        .await;
+        if let Err(err) = close_result {
+            abort_partial_writer(&mut self.writer, &self.operator, &self.storage_path).await;
+            return Err(err);
+        }
+        Ok(())
+    }
+
+    pub async fn abort(mut self) {
+        abort_partial_writer(&mut self.writer, &self.operator, &self.storage_path).await;
+    }
 }
 
 pub struct OpenDalReader {

blob/src/blob/backend.rs

@@ -7,7 +7,7 @@ use aruna_core::errors::{BlobError, ConversionError};
 use aruna_core::events::{Event, StorageEvent};
 use aruna_core::handle::Handle;
 use aruna_core::keyspaces::BUCKET_STATS_DB;
-use aruna_core::structs::{Backend, BackendBucket, BackendLocation};
+use aruna_core::structs::{Backend, BackendBucket, BackendLocation, ensure_confined_relative_path};
 use opendal::Operator;
 use std::path::PathBuf;
 use ulid::Ulid;
@@ -198,9 +198,9 @@ pub(super) fn build_backend_path(
     key: &str,
     ulid: Ulid,
 ) -> Result<String, ConversionError> {
-    PathBuf::from(bucket)
-        .join(format!("{}_{}", key, ulid))
-        .into_os_string()
+    let path = PathBuf::from(bucket).join(format!("{}_{}", key, ulid));
+    ensure_confined_relative_path(&path)?;
+    path.into_os_string()
         .into_string()
         .map_err(|_| ConversionError::OsStringError)
 }
@@ -227,9 +227,9 @@ pub(super) fn rebuild_backend_path(
         .rsplit_once('_')
         .map_or(file_name, |(base, _)| base);
 
-    parent
-        .join(format!("{}_{}", base_name, ulid))
-        .into_os_string()
+    let path = parent.join(format!("{}_{}", base_name, ulid));
+    ensure_confined_relative_path(&path)?;
+    path.into_os_string()
         .into_string()
         .map_err(|_| ConversionError::OsStringError)
 }

blob/src/blob/control_plane.rs

@@ -1,16 +1,14 @@
 use super::{BlobEvent, BlobHandler, ControlPlaneTimeoutKind};
+use crate::framing::{MAX_CONTROL_PLANE_FRAME, read_frame, write_frame};
 use crate::messages::{MessageType, ReplicationMessage};
 use aruna_core::errors::BlobError;
 use aruna_core::structs::BackendLocation;
 use aruna_net::streams::{RecvStream, SendStream};
 use std::future::Future;
 use std::time::Duration;
-use tokio::io::{AsyncReadExt as TokioAsyncReadExt, AsyncWriteExt as TokioAsyncWriteExt};
 use tokio::time::timeout;
 use ulid::Ulid;
 
-const MAX_CONTROL_PLANE_MESSAGE_SIZE: usize = 128 * 1024 * 1024;
-
 pub(super) fn control_plane_timeout_event(
     kind: ControlPlaneTimeoutKind,
     action: &'static str,
@@ -84,19 +82,8 @@ pub(super) async fn send_framed_message_with_timeout(
     timeout_duration: Duration,
     action: &'static str,
 ) -> Result<(), BlobEvent> {
-    if payload.len() > MAX_CONTROL_PLANE_MESSAGE_SIZE {
-        return Err(BlobEvent::Error(BlobError::WriteError(format!(
-            "control-plane message exceeds maximum size: {} bytes",
-            payload.len()
-        ))));
-    }
-
     match with_control_plane_timeout(
-        async {
-            TokioAsyncWriteExt::write_u32(sender, payload.len() as u32).await?;
-            TokioAsyncWriteExt::write_all(sender, payload).await?;
-            TokioAsyncWriteExt::flush(sender).await
-        },
+        write_frame(sender, payload, MAX_CONTROL_PLANE_FRAME),
         timeout_duration,
         ControlPlaneTimeoutKind::Write,
         action,
@@ -115,18 +102,7 @@ pub(super) async fn read_framed_message_with_timeout(
     action: &'static str,
 ) -> Result<Vec<u8>, BlobEvent> {
     match with_control_plane_timeout(
-        async {
-            let msg_len = TokioAsyncReadExt::read_u32(receiver).await?;
-            if msg_len as usize > MAX_CONTROL_PLANE_MESSAGE_SIZE {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::InvalidData,
-                    format!("control-plane frame too large: {msg_len} bytes"),
-                ));
-            }
-            let mut buf = vec![0; msg_len as usize];
-            TokioAsyncReadExt::read_exact(receiver, &mut buf).await?;
-            Ok::<Vec<u8>, std::io::Error>(buf)
-        },
+        read_frame(receiver, MAX_CONTROL_PLANE_FRAME),
         timeout_duration,
         ControlPlaneTimeoutKind::Read,
         action,