diff --git a/Cargo.lock b/Cargo.lock index 911bb8f4..a12146ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2934,7 +2934,6 @@ dependencies = [ "async-stream", "axum", "axum-extra", - "base64", "bytes", "ed25519-dalek", "elegant-departure", @@ -2967,6 +2966,7 @@ dependencies = [ "serde", "serde-vars", "serde_json", + "serde_with", "stresstest", "tempfile", "thiserror", @@ -3000,6 +3000,7 @@ dependencies = [ "quick-xml", "regex", "reqwest 0.13.4", + "ring", "sentry", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 742fd49b..7fbfe466 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -80,6 +80,7 @@ rand = "0.10.1" rand_distr = "0.6.0" regex = "1.12.4" reqwest = { version = "0.13.4", default-features = false } +ring = "0.17.14" rustls = { version = "0.23.40", default-features = false } secrecy = "0.10.3" sentry = "0.48.3" @@ -97,6 +98,7 @@ rdkafka = { version = "0.39.0", features = [ serde = { version = "1.0.228", features = ["derive"] } serde-vars = "0.3.1" serde_json = "1.0.150" +serde_with = "3.21.0" serde_yaml = "0.9.34-deprecated" sketches-ddsketch = "0.3.1" tempfile = "3.27.0" diff --git a/objectstore-server/Cargo.toml b/objectstore-server/Cargo.toml index e6d8a1e3..43f86520 100644 --- a/objectstore-server/Cargo.toml +++ b/objectstore-server/Cargo.toml @@ -16,7 +16,6 @@ argh = { workspace = true } async-stream = { workspace = true } axum = { workspace = true, features = ["multipart"] } axum-extra = { workspace = true, features = ["typed-header"] } -base64 = { workspace = true } bytes = { workspace = true } ed25519-dalek = { workspace = true, features = ["pem"] } elegant-departure = { workspace = true, features = ["tokio"] } @@ -46,6 +45,7 @@ sentry = { workspace = true, features = ["tower-axum-matched-path", "tracing", " serde = { workspace = true } serde-vars = { workspace = true } serde_json = { workspace = true } +serde_with = { workspace = true } thiserror = { workspace = true } thread_local = { workspace = true } tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } diff --git a/objectstore-server/src/auth/service.rs b/objectstore-server/src/auth/service.rs index 946b6215..be18e0e5 100644 --- a/objectstore-server/src/auth/service.rs +++ b/objectstore-server/src/auth/service.rs @@ -226,7 +226,7 @@ impl AuthAwareService { id: ObjectId, session: SessionToken, ) -> ApiResult { - // An offset query can commit an assembled object. + // A status query can detect an upload to be complete and cause logical object creation. self.check_permission(Permission::ObjectWrite, id.context())?; Ok(self.service.upload_offset(id, session).await?) } diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index dca9f45b..753ee997 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -65,9 +65,11 @@ use anyhow::Result; use figment::providers::{Env, Format, Serialized, Yaml}; use objectstore_service::backend::local_fs::FileSystemConfig; use objectstore_service::change_stream::CostTrackerConfig; +use objectstore_service::resumable::Encryptor; use objectstore_types::auth::Permission; use secrecy::{CloneableSecret, SecretBox, SerializableSecret, zeroize::Zeroize}; use serde::{Deserialize, Serialize}; +use serde_with::{Bytes, serde_as}; pub use objectstore_log::{LevelFilter, LogFormat, LoggingConfig}; pub use objectstore_service::backend::{MultipartUploadStorageConfig, StorageConfig}; @@ -575,6 +577,8 @@ pub struct Config { /// - `OS__SERVICE__CONCURRENCY_QUEUE` /// - `OS__SERVICE__CONCURRENCY_TIMEOUT` /// - `OS__SERVICE__BULK_CONCURRENCY_PCT` +/// - `OS__SERVICE__RESUMABLE_TOKEN_ENCRYPTION__ACTIVE_KEY_ID` +/// - `OS__SERVICE__RESUMABLE_TOKEN_ENCRYPTION__KEYS` #[derive(Debug, Deserialize, Serialize)] #[serde(default)] pub struct Service { @@ -629,6 +633,58 @@ pub struct Service { /// /// `60` pub bulk_concurrency_pct: u32, + + /// Persistent encryption keys for resumable-upload session tokens returned to clients. + /// + /// Tokens are always encrypted. When this is absent, Objectstore generates a fresh in-memory + /// AES-256 key at startup, so resumable sessions become invalid after a restart. Configure a + /// persistent keyring for sessions that must survive restarts. Keep old keys configured while + /// their sessions may still be active; removing a key intentionally invalidates those sessions. + /// Values must be raw AES-256 key bytes. + /// + /// ```yaml + /// service: + /// resumable_token_encryption: + /// active_key_id: v1 + /// keys: + /// v1: ${file:/var/run/secrets/objectstore/resumable-upload-v1} + /// ``` + pub resumable_token_encryption: Option, +} + +impl Service { + /// Loads and validates the configured resumable token encryption keys. + pub(crate) fn resumable_token_encryption(&self) -> Result> { + let Some(config) = &self.resumable_token_encryption else { + return Ok(None); + }; + + Encryptor::new(config.active_key_id.clone(), config.keys.clone()).map(Some) + } +} + +/// AES-256-GCM keys used to protect externally visible resumable session tokens. +#[serde_as] +#[derive(Clone, Deserialize, Serialize)] +pub struct ResumableTokenEncryptionConfig { + /// Key used to encrypt newly created sessions. + pub active_key_id: String, + /// Raw, exactly 32-byte AES-256 keys, indexed by rotation ID. + /// + /// File-backed secrets should use `${file:PATH}` so they are loaded during configuration + /// deserialization. + #[serde(default)] + #[serde_as(as = "BTreeMap<_, Bytes>")] + pub keys: BTreeMap>, +} + +impl fmt::Debug for ResumableTokenEncryptionConfig { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ResumableTokenEncryptionConfig") + .field("active_key_id", &self.active_key_id) + .field("key_ids", &self.keys.keys().collect::>()) + .finish() + } } impl Default for Service { @@ -638,6 +694,7 @@ impl Default for Service { concurrency_queue: 0, concurrency_timeout: Duration::from_secs(1), bulk_concurrency_pct: 60, + resumable_token_encryption: None, } } } @@ -866,6 +923,40 @@ mod tests { }); } + #[test] + fn resumable_token_encryption_rejects_invalid_configuration() { + let mut valid = tempfile::NamedTempFile::new().unwrap(); + valid.write_all(&[7; 32]).unwrap(); + let mut short = tempfile::NamedTempFile::new().unwrap(); + short.write_all(&[7; 31]).unwrap(); + for yaml in [ + "service:\n resumable_token_encryption:\n active_key_id: v1\n".to_owned(), + format!( + "service:\n resumable_token_encryption:\n active_key_id: missing\n keys:\n v1: ${{file:{}}}\n", + valid.path().display(), + ), + format!( + "service:\n resumable_token_encryption:\n active_key_id: bad_key\n keys:\n 'bad key': ${{file:{}}}\n", + valid.path().display(), + ), + format!( + "service:\n resumable_token_encryption:\n active_key_id: v1\n keys:\n v1: ${{file:{}}}\n", + short.path().display(), + ), + ] { + let mut tempfile = tempfile::NamedTempFile::new().unwrap(); + tempfile.write_all(yaml.as_bytes()).unwrap(); + figment::Jail::expect_with(|_jail| { + let config = Config::load(Some(tempfile.path())).unwrap(); + assert!( + config.service.resumable_token_encryption().is_err(), + "accepted {yaml}" + ); + Ok(()) + }); + } + } + #[test] fn configured_with_env_and_yaml() { let mut tempfile = tempfile::NamedTempFile::new().unwrap(); @@ -1026,6 +1117,7 @@ mod tests { let dir = tempfile::tempdir().unwrap(); std::fs::write(dir.path().join("relative-password"), "hunter3").unwrap(); + std::fs::write(dir.path().join("resumable-token-key"), [255; 32]).unwrap(); let config_path = dir.path().join("config.yml"); std::fs::write( @@ -1040,6 +1132,11 @@ mod tests { from.env: ${{KAFKA_SASL_PASSWORD}} from.relative.file: ${{file:relative-password}} from.absolute.file: ${{file:{}}} + service: + resumable_token_encryption: + active_key_id: v1 + keys: + v1: ${{file:resumable-token-key}} "#, absolute_secret.display() ), @@ -1064,6 +1161,13 @@ mod tests { sink.override_params["not.a.reference"], "prod-${NOT_A_VAR", "a value that is not a reference is left alone" ); + assert!( + config + .service + .resumable_token_encryption() + .unwrap() + .is_some() + ); Ok(()) }); diff --git a/objectstore-server/src/endpoints/mod.rs b/objectstore-server/src/endpoints/mod.rs index 42a9755d..1b5daf67 100644 --- a/objectstore-server/src/endpoints/mod.rs +++ b/objectstore-server/src/endpoints/mod.rs @@ -32,12 +32,12 @@ //! Clients are still encouraged to send the whole payload in a single request, as that's the most //! efficient and reliable approach. //! The server knows the total size from the session, so it recognizes the chunk carrying the last -//! byte and commits the object itself. +//! byte and completes the upload itself. //! //! Resumable uploads use the object endpoints above, selected by a query parameter: //! `upload_type=resumable` opens a session, and `session=` addresses it from then on. -//! Session creation returns the opaque backend token unchanged; subsequent requests encode it as -//! unpadded base64url in the `session` query parameter. +//! Session creation returns an opaque token encoded once as unpadded base64url; that value can be +//! placed directly in the `session` query parameter. //! The object is named by the request path as usual, and [`objectstore_types::resumable`] //! holds the protocol types. //! @@ -57,14 +57,16 @@ //! `Upload-Offset` header: a byte offset submits the body as the chunk starting there, while //! the `*` wildcard submits an empty body and asks which offset the server holds. Both answer //! `204 No Content` with the authoritative `Upload-Offset` while bytes remain, and -//! `201 Created` with `{"key"}` once the object is committed. +//! `201 Created` with `{"key"}` once the upload is complete and the object is available through +//! the normal object endpoints. The session is terminal at that point. //! The offset in the response may be lower than the end of the last chunk that was sent. //! Backends can e.g. persist only aligned prefixes and discard the remainder, so clients must //! always continue from the returned offset. //! Every chunk requires `Content-Length`, even over HTTP/2, while creation and offset queries //! must not carry a request body. //! -//! An offset query can commit an object, so it requires write permission despite being read-shaped. +//! An offset query can finish pending backend publication work, so it requires write permission +//! despite being read-shaped. //! Termination likewise needs write rather than delete permission: it releases an in-progress upload, //! not an object. //! diff --git a/objectstore-server/src/endpoints/resumable.rs b/objectstore-server/src/endpoints/resumable.rs index 0c9a7059..5e70b25b 100644 --- a/objectstore-server/src/endpoints/resumable.rs +++ b/objectstore-server/src/endpoints/resumable.rs @@ -23,7 +23,8 @@ use objectstore_service::id::{ObjectContext, ObjectId}; use objectstore_service::stream::ClientStream; use objectstore_types::metadata::Metadata; use objectstore_types::resumable::{ - CommitResponse, CreateSessionResponse, HEADER_UPLOAD_OFFSET, UploadOffset, UploadProgress, + CompleteUploadResponse, CreateSessionResponse, HEADER_UPLOAD_OFFSET, UploadOffset, + UploadProgress, }; use crate::auth::AuthAwareService; @@ -113,12 +114,17 @@ async fn create_session_for_id( /// /// [`HEADER_UPLOAD_OFFSET`] selects between the two. A concrete offset submits the request /// body as the chunk starting there; the `*` wildcard submits nothing and asks where the -/// server stands, which also commits an object that was assembled but not yet committed. +/// server stands. A status query can observe completion when the final chunk succeeded but its +/// response was lost. A composed backend can also finish pending idempotent publication work +/// before reporting completion. /// Chunks require `Content-Length`, including over HTTP/2. Offset queries may omit it, but the /// body stream is checked and any bytes are rejected as a malformed request. /// /// Both answer `204 No Content` with the authoritative offset while bytes remain, and -/// `201 Created` with the key once the object is committed. +/// `201 Created` with the key once the upload is complete, the session is terminal, and the object +/// is available through the normal object endpoints. +/// The acknowledged offset may be lower than the submitted chunk's end, so clients should continue +/// from this response (or from a later explicit offset query), never from local byte accounting alone. pub(super) async fn continue_session( service: AuthAwareService, Xt(id): Xt, @@ -189,8 +195,8 @@ fn progress_response(progress: ApiResult, key: String) -> ApiRes [(HEADER_UPLOAD_OFFSET, http::HeaderValue::from(offset))], ) .into_response(), - UploadProgress::Committed => { - (StatusCode::CREATED, Json(CommitResponse { key })).into_response() + UploadProgress::Complete => { + (StatusCode::CREATED, Json(CompleteUploadResponse { key })).into_response() } }; @@ -229,7 +235,7 @@ mod tests { #[tokio::test] async fn commit_answers_created_with_the_key() { - let response = progress_response(Ok(UploadProgress::Committed), "my-key".into()).unwrap(); + let response = progress_response(Ok(UploadProgress::Complete), "my-key".into()).unwrap(); let (status, offset, body) = parts_of(response).await; assert_eq!(status, StatusCode::CREATED); diff --git a/objectstore-server/src/extractors/id.rs b/objectstore-server/src/extractors/id.rs index 1bc62a84..6aec0fa3 100644 --- a/objectstore-server/src/extractors/id.rs +++ b/objectstore-server/src/extractors/id.rs @@ -229,6 +229,7 @@ mod tests { use axum::routing::{get, post}; use objectstore_service::StorageService; use objectstore_service::backend::in_memory::InMemoryBackend; + use objectstore_service::resumable::Encryptor; use tower::ServiceExt; use crate::auth::PublicKeyDirectory; @@ -239,7 +240,10 @@ mod tests { use crate::web::RequestCounter; async fn test_state(config: Config) -> ServiceState { - let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory"))); + let service = StorageService::new( + Box::new(InMemoryBackend::new("in-memory")), + Encryptor::ephemeral().unwrap(), + ); let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await.unwrap()); let rate_limiter = RateLimiter::new(config.rate_limits.clone()); diff --git a/objectstore-server/src/resumable.rs b/objectstore-server/src/resumable.rs index a43f9f94..11ee356a 100644 --- a/objectstore-server/src/resumable.rs +++ b/objectstore-server/src/resumable.rs @@ -3,8 +3,6 @@ use axum::extract::{FromRequestParts, OptionalFromRequestParts, Query}; use axum::http::{HeaderName, HeaderValue, request::Parts}; use axum_extra::headers::{ContentLength, Error as HeaderError, Header}; -use base64::Engine as _; -use base64::engine::general_purpose::URL_SAFE_NO_PAD; use futures_util::TryStreamExt; use objectstore_service::error::Error as ServiceError; use objectstore_service::stream::ClientStream; @@ -177,19 +175,8 @@ where } fn decode_session_token(encoded: &str) -> ApiResult { - let bytes = URL_SAFE_NO_PAD - .decode(encoded) - .map_err(|error| ApiError::map_client("invalid session token", error))?; - - if URL_SAFE_NO_PAD.encode(&bytes) != encoded { - return Err(ApiError::client( - "session token must use unpadded base64 URL encoding", - )); - } - - String::from_utf8(bytes) - .map(SessionToken::from) - .map_err(|error| ApiError::map_client("session token is not valid UTF-8", error)) + SessionToken::from_base64url(encoded) + .map_err(|error| ApiError::map_client("invalid session token", error)) } /// Confirms that a request neither declares nor streams a non-empty body. @@ -255,14 +242,14 @@ mod tests { #[test] fn session_token_decodes_from_unpadded_base64url() { assert_eq!( - decode_session_token("Li4vZXNjYXBl").unwrap().as_ref(), - "../escape" + decode_session_token("Li4vZXNjYXBl").unwrap().as_bytes(), + b"../escape" ); } #[test] fn session_token_rejects_invalid_query_encodings() { - for invalid in ["%%%", "dG9rM24=", "_w"] { + for invalid in ["%%%", "dG9rM24="] { assert!( decode_session_token(invalid).is_err(), "accepted {invalid:?}" diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index 9143cb4d..9b55db37 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -6,12 +6,13 @@ use std::sync::Arc; use std::time::Duration; -use anyhow::Result; +use anyhow::{Context, Result}; use bytes::Bytes; use futures_util::Stream; use objectstore_service::change_stream::ChangeStreamFactory; use objectstore_service::concurrency::ConcurrencyLimiter; use objectstore_service::id::ObjectContext; +use objectstore_service::resumable::Encryptor; use objectstore_service::{StorageService, backend}; use tokio::runtime::Handle; @@ -68,12 +69,19 @@ impl Services { .as_ref() .map(ChangeStreamFactory::new) .unwrap_or_default(); + let resumable_token_encryption = match config.service.resumable_token_encryption()? { + Some(encryption) => encryption, + None => { + Encryptor::ephemeral().context("failed to initialize resumable token encryption")? + } + }; let backend = backend::from_config(config.storage.clone(), &streams).await?; let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency) .with_queue(config.service.concurrency_queue) .with_timeout(config.service.concurrency_timeout) .with_bulk(config.service.bulk_concurrency_pct); - let service = StorageService::new(backend).with_concurrency(concurrency); + let service = + StorageService::new(backend, resumable_token_encryption).with_concurrency(concurrency); service.start(); let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await?); diff --git a/objectstore-server/tests/resumable.rs b/objectstore-server/tests/resumable.rs index fc99e5ad..6764f4ec 100644 --- a/objectstore-server/tests/resumable.rs +++ b/objectstore-server/tests/resumable.rs @@ -1,13 +1,16 @@ //! Integration tests for the resumable upload endpoints. //! -//! No backend implements resumable uploads yet. Until a supporting test backend exists, these -//! tests cover request validation and ensure regular object requests remain unaffected. +//! The test server uses its default filesystem backend, which does not implement resumable +//! uploads. These tests cover request validation and ensure regular object requests remain +//! unaffected. Backend behavior is covered in the service and backend test suites. +//! TODO: Add end-to-end resumable upload coverage once the filesystem backend supports it. +use std::collections::BTreeMap; use std::io::{Read, Write}; use std::net::TcpStream; use anyhow::Result; -use objectstore_server::config::{AuthZ, Config}; +use objectstore_server::config::{AuthZ, Config, ResumableTokenEncryptionConfig, Service}; use objectstore_test::server::TestServer; use objectstore_types::resumable::{HEADER_UPLOAD_LENGTH, HEADER_UPLOAD_OFFSET}; use reqwest::StatusCode; @@ -15,6 +18,9 @@ use reqwest::StatusCode; /// Unpadded base64url for the opaque backend token `some-token`. const SESSION: &str = "c29tZS10b2tlbg"; +/// Protected `some-token`, bound to `test/org.1/objects/my-key` with the test key below. +const PROTECTED_SESSION: &str = "BHRlc3QAAAAAAAAAAAAAAAAa-svQkfdonL5u60b-WT_LzpuGlgG7hWo5euKiKopuFcKkMHnolxLc6JiaHvvMLcQi39wg4playvMM9HNrTtQOvOVVDI9IXylpxwkWmu5yFrCGsPj7BA"; + async fn test_server() -> TestServer { TestServer::with_config(Config { auth: AuthZ { @@ -26,6 +32,24 @@ async fn test_server() -> TestServer { .await } +async fn test_server_with_protected_session() -> Result { + Ok(TestServer::with_config(Config { + auth: AuthZ { + enforce: false, + ..Default::default() + }, + service: Service { + resumable_token_encryption: Some(ResumableTokenEncryptionConfig { + active_key_id: "test".into(), + keys: BTreeMap::from([("test".into(), vec![7; 32])]), + }), + ..Default::default() + }, + ..Default::default() + }) + .await) +} + /// Sends a raw HTTP/1.1 `PUT`, preserving the caller's exact body framing headers. async fn raw_put(server: &TestServer, path: &str, headers: &str, body: &str) -> Result { let url = reqwest::Url::parse(&server.url(path))?; @@ -136,10 +160,10 @@ async fn declined_session_creation_returns_not_implemented() -> Result<()> { #[tokio::test] async fn offset_query_does_not_require_content_length() -> Result<()> { - let server = test_server().await; + let server = test_server_with_protected_session().await?; let response = raw_put( &server, - &format!("/v1/objects/test/org=1/my-key?session={SESSION}"), + &format!("/v1/objects/test/org=1/my-key?session={PROTECTED_SESSION}"), &format!("{HEADER_UPLOAD_OFFSET}: *\r\n"), "", ) @@ -256,11 +280,11 @@ async fn delete_rejects_upload_type() -> Result<()> { #[tokio::test] async fn session_takes_precedence_over_upload_type() -> Result<()> { - let server = test_server().await; + let server = test_server_with_protected_session().await?; let response = reqwest::Client::new() .put(server.url(&format!( - "/v1/objects/test/org=1/my-key?upload_type=resumable&session={SESSION}" + "/v1/objects/test/org=1/my-key?upload_type=resumable&session={PROTECTED_SESSION}" ))) .header(HEADER_UPLOAD_OFFSET, "*") .send() diff --git a/objectstore-service/Cargo.toml b/objectstore-service/Cargo.toml index b946a728..a3c30ad8 100644 --- a/objectstore-service/Cargo.toml +++ b/objectstore-service/Cargo.toml @@ -27,6 +27,7 @@ objectstore-types = { workspace = true } quick-xml = { workspace = true, features = ["serialize"] } regex = { workspace = true } reqwest = { workspace = true, features = ["charset", "http2", "hickory-dns", "json", "multipart", "native-tls-no-alpn", "stream", "system-proxy"] } +ring = { workspace = true } sentry = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 7cdee09f..015252bc 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -260,14 +260,26 @@ trips for objects large enough that re-sending the whole payload is expensive. reports the offset now persisted. 3. After a failure, [`upload_offset`](backend::common::Backend::upload_offset) reports where the backend stands, so the caller resumes from there. -4. The chunk carrying the last byte commits the object. There is no finalize call — - the backend recognizes that chunk from the declared total size. +4. The chunk carrying the last byte completes the upload. There is no separate completion call — + the backend recognizes that chunk from the declared total size and makes the object available + through its normal read path before reporting completion. 5. At any time, an upload can be canceled, which discards what its session holds. +Backend session tokens remain private to the service. Before returning a token to a client, +[`StorageService`] places the canonical object path and backend-defined token string in an +AES-256-GCM envelope. Continuation, offset-query, and cancellation operations authenticate and +open that envelope, reject an object-path mismatch, and pass the original string back to the +backend. + Not all backends support resumable uploads. A backend returns no session when it declines a particular upload; this is a routine outcome rather than an error. Acceptance can depend on the declared size, the metadata, or whether resuming is possible in principle. +The offset returned by every chunk response is authoritative. A backend may accept the full +request body but persist only a prefix (for example, up to an internal alignment boundary), so a +client must not advance by the submitted `Content-Length` on its own. It continues from the +preceding response's offset, or performs an explicit offset query after an ambiguous failure. + ## Multipart Uploads When the configured backend supports it, [`StorageService`] exposes multipart diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs index b4ce87b4..765e5044 100644 --- a/objectstore-service/src/backend/common.rs +++ b/objectstore-service/src/backend/common.rs @@ -4,7 +4,7 @@ use std::fmt; use objectstore_types::metadata::{ExpirationPolicy, Metadata}; use objectstore_types::range::{ByteRange, ContentRange}; -use objectstore_types::resumable::{SessionToken, UploadProgress}; +use objectstore_types::resumable::UploadProgress; use bytes::Bytes; @@ -14,6 +14,7 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::BackendToken; use crate::stream::{ClientStream, PayloadStream}; /// User agent string used for outgoing requests. @@ -79,6 +80,9 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { /// Object metadata and its total length are declared upfront and cannot be mutated /// during the upload. /// + /// The returned string is opaque backend-defined state. [`StorageService`](crate::StorageService) + /// protects it before exposing the session token outside the service layer. + /// /// Returns `Ok(None)` when this backend cannot store the described object resumably. Declining /// is a routine outcome rather than an error, and the default implementation declines. /// @@ -91,21 +95,33 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { id: &ObjectId, metadata: &Metadata, total_length: u64, - ) -> Result> { + ) -> Result> { let _ = (id, metadata, total_length); Ok(None) } /// Writes a chunk of `content_length` bytes at `offset` into an open session. /// - /// `offset` must equal the offset the backend currently holds. + /// A backend may acknowledge fewer bytes than the chunk supplied, for example by persisting + /// only an aligned prefix. Callers must continue from the authoritative offset in the returned + /// [`UploadProgress`], or query [`Self::upload_offset`] after an ambiguous failure. A backend + /// may or may not accept a replay starting before its persisted offset. + /// + /// [`UploadProgress::Complete`] means the upload is terminal and the object is available + /// through this backend's normal read methods. A backend that composes another backend must + /// finish its own publication work before returning that outcome. + /// + /// A `content_length` of zero is valid: it is how a zero-length object is uploaded, and it + /// completes such a session. Against a session that still expects bytes it writes nothing and + /// reports the offset the backend holds. + /// /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session, /// and [`ErrorKind::ChunkExceedsUploadLength`] when the chunk would exceed the total length /// declared when the session was created. async fn put_chunk( &self, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, offset: u64, content_length: u64, stream: ClientStream, @@ -116,8 +132,12 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { /// Reports how far the session has progressed. /// - /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session. - async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + /// This can return [`UploadProgress::Complete`] repeatedly after the final chunk, including + /// when its original response was lost. A composed backend may finish pending idempotent + /// publication work before returning that terminal outcome. + /// + /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify a known session. + async fn upload_offset(&self, id: &ObjectId, session: &BackendToken) -> Result { let _ = (id, session); Err(ErrorKind::Unsupported.into()) } @@ -125,7 +145,7 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { /// Cancels an upload session, discarding whatever was uploaded. /// /// Returns [`ErrorKind::UnknownUploadSession`] when `session` does not identify an open session. - async fn cancel_upload(&self, id: &ObjectId, session: &SessionToken) -> Result<()> { + async fn cancel_upload(&self, id: &ObjectId, session: &BackendToken) -> Result<()> { let _ = (id, session); Err(ErrorKind::Unsupported.into()) } diff --git a/objectstore-service/src/backend/counting.rs b/objectstore-service/src/backend/counting.rs index 66ce6ef5..84919c99 100644 --- a/objectstore-service/src/backend/counting.rs +++ b/objectstore-service/src/backend/counting.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use objectstore_types::metadata::Metadata; use objectstore_types::range::ByteRange; -use objectstore_types::resumable::{SessionToken, UploadProgress}; +use objectstore_types::resumable::UploadProgress; use crate::backend::common::{ Backend, DeleteResponse, GetResponse, MetadataResponse, MultipartUploadBackend, PutResponse, @@ -29,6 +29,7 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::BackendToken; use crate::stream::ClientStream; /// Increments `cogs.usage` by one operation for the given `usecase`. @@ -106,7 +107,7 @@ impl Backend for CountingBackend { id: &ObjectId, metadata: &Metadata, total_length: u64, - ) -> Result> { + ) -> Result> { count(&id.context.usecase); self.inner .create_upload_session(id, metadata, total_length) @@ -116,7 +117,7 @@ impl Backend for CountingBackend { async fn put_chunk( &self, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, offset: u64, content_length: u64, stream: ClientStream, @@ -127,12 +128,12 @@ impl Backend for CountingBackend { .await } - async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + async fn upload_offset(&self, id: &ObjectId, session: &BackendToken) -> Result { count(&id.context.usecase); self.inner.upload_offset(id, session).await } - async fn cancel_upload(&self, id: &ObjectId, session: &SessionToken) -> Result<()> { + async fn cancel_upload(&self, id: &ObjectId, session: &BackendToken) -> Result<()> { count(&id.context.usecase); self.inner.cancel_upload(id, session).await } diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 565c888b..6851af82 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -31,6 +31,7 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{BackendToken, UploadProgress}; use crate::stream::ClientStream; /// Configuration for [`GcsBackend`]. @@ -189,6 +190,7 @@ impl GcsObject { fn metadata_size(&self) -> u64 { self.metadata .iter() + .filter(|(key, _)| !matches!(key, GcsMetaKey::EmulatorIgnored)) .map(|(key, value)| key.to_string().len() as u64 + value.len() as u64) .sum() } @@ -334,7 +336,7 @@ impl std::str::FromStr for GcsMetaKey { type Err = anyhow::Error; fn from_str(s: &str) -> Result { - if matches!(s, "x_emulator_upload" | "x_testbench_upload") { + if s.starts_with("x_emulator_") || s.starts_with("x_testbench_") { return Ok(GcsMetaKey::EmulatorIgnored); } @@ -461,6 +463,45 @@ fn insert_gcs_meta_header( Ok(()) } +/// Special status code returned by GCS when a Resumable Upload is canceled successfully or when +/// making other requests to a session that was recently canceled. +const CLIENT_CLOSED_REQUEST_STATUS: u16 = 499; + +/// Represents a resumable upload session in GCS. +#[derive(Debug)] +struct ResumableUpload { + // URI to use for requests that act on this session, returned by GCS in the `Location` header + // on session creation. + session_uri: Url, + // Total length of the object, declared at session creation time. + total_length: u64, +} + +impl ResumableUpload { + fn new(session_uri: Url, total_length: u64) -> Self { + Self { + session_uri, + total_length, + } + } + + fn into_token(self) -> BackendToken { + format!("{}.{}", self.total_length, self.session_uri) + } + + fn from_token(token: &BackendToken) -> Result { + let (total_length, session_uri) = token + .split_once('.') + .ok_or(ErrorKind::UnknownUploadSession)?; + let total_length = total_length + .parse() + .map_err(|_| ErrorKind::UnknownUploadSession)?; + let session_uri = Url::parse(session_uri).map_err(|_| ErrorKind::UnknownUploadSession)?; + let session = Self::new(session_uri, total_length); + Ok(session) + } +} + /// Returns `true` if the error is a transient backend failure worth retrying. fn error_is_retryable(error: &Error) -> bool { matches!( @@ -729,6 +770,25 @@ impl GcsBackend { }) .await } + + /// Reports an object write to the [`ChangeStream`]. + fn report_object_write( + &self, + id: &ObjectId, + stored_size: Option, + metadata_size: u64, + expires_at: Option, + ) { + match stored_size { + Some(stored_size) => { + self.change_stream + .write(id, stored_size + metadata_size, expires_at) + } + None => { + objectstore_metrics::count!("change_stream.unreported", reason = "no_stored_size") + } + } + } } impl fmt::Debug for GcsBackend { @@ -740,6 +800,96 @@ impl fmt::Debug for GcsBackend { } } +/// Converts GCS's inclusive `Range: bytes=0-N` acknowledgement into the next offset. +fn range_header_to_offset(value: &str, total_length: u64) -> Result { + let end = value + .strip_prefix("bytes=0-") + .filter(|end| !end.is_empty() && end.bytes().all(|byte| byte.is_ascii_digit())) + .and_then(|end| end.parse::().ok()) + .ok_or_else(|| { + Error::new( + ErrorKind::BackendFailure, + "GCS: malformed resumable Range header", + ) + })?; + let offset = end.checked_add(1).ok_or_else(|| { + Error::new( + ErrorKind::BackendFailure, + "GCS: resumable Range header overflows", + ) + })?; + if offset > total_length { + return Err(Error::new( + ErrorKind::BackendFailure, + "GCS: incomplete resumable Range reaches declared upload length", + )); + } + Ok(offset) +} + +/// Interprets a Resumable Upload response, shared by chunk writes and status queries. +/// +/// Returns the progress GCS reported, plus the completed object when this is the response that +/// finished the upload. +async fn range_response_to_upload_progress( + session: &ResumableUpload, + response: reqwest::Response, +) -> Result<(UploadProgress, Option)> { + let status = response.status(); + + match status { + StatusCode::NOT_FOUND => { + response.drain_body().await; + return Err(ErrorKind::UnknownUploadSession.into()); + } + status if status == StatusCode::GONE || status.as_u16() == CLIENT_CLOSED_REQUEST_STATUS => { + response.drain_body().await; + return Err(ErrorKind::UploadSessionGone.into()); + } + _ => {} + } + + let response = response + .check_error("GCS: unexpected resumable upload status") + .await?; + + match status { + StatusCode::OK | StatusCode::CREATED => { + let body = response + .bytes() + .await + .reqwest_context("GCS: read completed resumable upload response")?; + let object = serde_json::from_slice::(&body).context( + ErrorKind::CorruptData, + "GCS: parse completed resumable upload response", + )?; + Ok((UploadProgress::Complete, Some(object))) + } + // GCS calls it "308 Resume Incomplete" + StatusCode::PERMANENT_REDIRECT => { + let offset = match response.headers().get(header::RANGE) { + Some(range) => { + let range = range.to_str().map_err(|_| { + Error::new(ErrorKind::BackendFailure, "GCS: invalid Range header") + })?; + range_header_to_offset(range, session.total_length)? + } + // GCS omits this header while it holds nothing + None => 0, + }; + response.drain_body().await; + Ok((UploadProgress::Incomplete { offset }, None)) + } + _ => { + response.drain_body().await; + Err(Error::new( + ErrorKind::BackendFailure, + format!("GCS: unexpected resumable upload status {status}"), + )) + } + } +} + #[async_trait::async_trait] impl Backend for GcsBackend { fn name(&self) -> &'static str { @@ -795,16 +945,12 @@ impl Backend for GcsBackend { .await?; let stored_size = read_stored_content_length(response).await; - - if let Some(payload_size) = stored_size { - self.change_stream.write( - id, - payload_size + gcs_metadata.metadata_size(), - metadata.time_expires, - ); - } else { - objectstore_metrics::count!("change_stream.unreported", reason = "no_stored_size"); - } + self.report_object_write( + id, + stored_size, + gcs_metadata.metadata_size(), + metadata.time_expires, + ); Ok(()) } @@ -919,6 +1065,209 @@ impl Backend for GcsBackend { Ok(()) } + #[tracing::instrument(level = "debug", fields(?id), skip_all)] + async fn create_upload_session( + &self, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result> { + objectstore_log::debug!("Creating resumable upload session on GCS backend"); + let url = self.upload_url(id, "resumable")?; + let metadata_json = serde_json::to_vec(&GcsObject::from_metadata(metadata)).context( + ErrorKind::Internal, + "GCS: failed to serialize resumable upload metadata", + )?; + let content_type = metadata.content_type.clone(); + + let location = self + .with_retry("create_resumable_upload", || { + let url = url.clone(); + let metadata_json = metadata_json.clone(); + let content_type = content_type.clone(); + async move { + let response = self + .request(Method::POST, url) + .await? + .header(header::CONTENT_TYPE, "application/json") + .header("x-upload-content-type", content_type.as_ref()) + .header("x-upload-content-length", total_length) + .body(metadata_json) + .send_traced() + .await + .check_error("GCS: create resumable upload") + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + response.drain_body().await; + return Err(Error::new( + ErrorKind::BackendFailure, + format!("GCS: unexpected resumable session creation status {status}"), + )); + } + + let location = response + .headers() + .get(header::LOCATION) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned) + .ok_or_else(|| { + Error::new( + ErrorKind::BackendFailure, + "GCS: resumable session response missing valid Location header", + ) + })?; + response.drain_body().await; + Ok(location) + } + }) + .await?; + + let session_uri = Url::parse(&location).map_err(|_| { + Error::new( + ErrorKind::BackendFailure, + "GCS: resumable session Location is not a valid URL", + ) + })?; + let session = ResumableUpload::new(session_uri, total_length); + Ok(Some(session.into_token())) + } + + #[tracing::instrument(level = "debug", fields(?id, offset, content_length), skip_all)] + async fn put_chunk( + &self, + id: &ObjectId, + token: &BackendToken, + offset: u64, + content_length: u64, + stream: ClientStream, + ) -> Result { + objectstore_log::debug!("Uploading resumable chunk to GCS backend"); + let session = ResumableUpload::from_token(token)?; + + let end = offset + .checked_add(content_length) + .filter(|end| *end <= session.total_length) + .ok_or(ErrorKind::ChunkExceedsUploadLength { + offset, + content_length, + upload_length: session.total_length, + })?; + + let content_range = match content_length { + // If `total_length` of this upload is 0, the only way to complete it is to + // put a an empty chunk, which is accomplished by putting `*/0` in this header. + // Otherwise, this request is equivalent to an offset query. + 0 => format!("bytes */{}", session.total_length), + _ => format!("bytes {offset}-{}/{}", end - 1, session.total_length), + }; + + let response = self + .request(Method::PUT, session.session_uri.as_str()) + .await? + .header(header::CONTENT_LENGTH, content_length) + .header(header::CONTENT_RANGE, content_range) + .body(Body::wrap_stream(stream)) + .send_traced() + .await + .reqwest_context("GCS: upload resumable chunk")?; + + range_response_to_upload_progress(&session, response) + .await + .map(|(progress, completed)| { + if let Some(object) = completed { + let stored_size = object.size.as_deref().and_then(|size| size.parse().ok()); + self.report_object_write( + id, + stored_size, + object.metadata_size(), + object.custom_time, + ); + } + progress + }) + } + + #[tracing::instrument(level = "debug", fields(?id), skip_all)] + async fn upload_offset(&self, id: &ObjectId, token: &BackendToken) -> Result { + objectstore_log::debug!("Querying resumable upload offset on GCS backend"); + let session = ResumableUpload::from_token(token)?; + + self.with_retry("query_resumable_upload", || async { + let response = self + .request(Method::PUT, session.session_uri.as_str()) + .await? + .header( + header::CONTENT_RANGE, + format!("bytes */{}", session.total_length), + ) + .send_traced() + .await + .reqwest_context("GCS: query resumable upload")?; + + range_response_to_upload_progress(&session, response) + .await + .map(|(progress, completed)| { + // The final `put_chunk` may have persisted the object but failed while + // reading its response, so completion observed here must be reported too. + if let Some(object) = completed { + let stored_size = object.size.as_deref().and_then(|size| size.parse().ok()); + self.report_object_write( + id, + stored_size, + object.metadata_size(), + object.custom_time, + ); + } + progress + }) + }) + .await + } + + #[tracing::instrument(level = "debug", fields(?id), skip_all)] + async fn cancel_upload(&self, id: &ObjectId, token: &BackendToken) -> Result<()> { + objectstore_log::debug!("Cancelling resumable upload on GCS backend"); + let session = ResumableUpload::from_token(token)?; + let session_uri = session.session_uri; + self.with_retry("cancel_resumable_upload", || { + let session_uri = session_uri.clone(); + async move { + let response = self + .request(Method::DELETE, session_uri) + .await? + .send_traced() + .await + .reqwest_context("GCS: cancel resumable upload")?; + match response.status() { + // Expected status code when canceling a recently created upload. + status if status.as_u16() == CLIENT_CLOSED_REQUEST_STATUS => { + response.drain_body().await; + Ok(()) + } + // The upload was already canceled or never existed. + StatusCode::GONE | StatusCode::NOT_FOUND => { + response.drain_body().await; + Ok(()) + } + _ => { + response + .check_error("GCS: cancel resumable upload") + .await? + .drain_body() + .await; + Err(Error::new( + ErrorKind::BackendFailure, + "GCS: unexpected resumable cancellation status", + )) + } + } + } + }) + .await + } + async fn join(&self) { flush_change_stream(&self.change_stream).await; } @@ -1275,6 +1624,7 @@ mod tests { use anyhow::Result; use objectstore_types::scope::{Scope, Scopes}; + use reqwest::header::{HeaderMap, HeaderValue}; #[cfg(feature = "storage-cogs")] use objectstore_inventory_tracker::OpType; @@ -1289,6 +1639,21 @@ mod tests { use crate::multipart::CompletedPart; use crate::stream; + impl GcsBackend { + async fn create_upload_session( + &self, + id: &ObjectId, + metadata: &Metadata, + total_length: u64, + ) -> Result { + ::create_upload_session(self, id, metadata, total_length) + .await? + .ok_or_else(|| Error::from(ErrorKind::Unsupported).into()) + } + } + + const RESUMABLE_CHUNK_SIZE: usize = 256 * 1024; + // NB: Not run any of these tests, you need to have a GCS emulator running. This is done // automatically in CI. // @@ -1320,6 +1685,37 @@ mod tests { Ok((GcsBackend::new(config, &streams).await?, producer)) } + #[derive(Deserialize)] + struct RetryTestResource { + id: String, + } + + /// Configures one storage-testbench failure and sends its ID on subsequent backend requests. + async fn inject_retry_test( + backend: &mut GcsBackend, + method: &str, + instruction: &str, + ) -> Result<()> { + let retry_test: RetryTestResource = reqwest::Client::new() + .post(backend.endpoint.join("retry_test")?) + .json(&serde_json::json!({ + "instructions": { method: [instruction] }, + "transport": "HTTP", + })) + .send() + .await? + .error_for_status()? + .json() + .await?; + + let mut headers = HeaderMap::new(); + headers.insert("x-retry-test-id", HeaderValue::from_str(&retry_test.id)?); + backend.client = reqwest::Client::builder() + .default_headers(headers) + .build()?; + Ok(()) + } + fn make_id() -> ObjectId { ObjectId::random(ObjectContext { usecase: "testing".into(), @@ -1327,6 +1723,387 @@ mod tests { }) } + fn make_id_with_key(key: &str) -> ObjectId { + ObjectId::new( + ObjectContext { + usecase: "testing".into(), + scopes: Scopes::from_iter([ + Scope::create("organization", "42").unwrap(), + Scope::create("project", "7").unwrap(), + ]), + }, + key.into(), + ) + } + + #[test] + fn resumable_range_reports_next_offset_and_rejects_malformed_values() -> Result<()> { + assert_eq!(range_header_to_offset("bytes=0-0", 10)?, 1); + assert_eq!(range_header_to_offset("bytes=0-262143", 300_000)?, 262_144); + + for malformed in [ + "", + "bytes=1-2", + "bytes=0-", + "bytes=0-*", + "bytes=0-9 ", + "bytes=0-9,bytes=20-30", + ] { + assert!( + range_header_to_offset(malformed, 100).is_err(), + "accepted {malformed:?}" + ); + } + assert_eq!(range_header_to_offset("bytes=0-9", 10)?, 10); + assert!(range_header_to_offset("bytes=0-18446744073709551615", u64::MAX).is_err()); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_zero_length_upload_and_metadata() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id_with_key("resumable-zero"); + let metadata = Metadata { + content_type: "application/x-empty".into(), + custom: BTreeMap::from([("upload".into(), "zero".into())]), + ..Default::default() + }; + + let token = backend.create_upload_session(&id, &metadata, 0).await?; + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Complete + ); + + let (stored_metadata, _, payload) = backend.get_object(&id, None).await?.unwrap(); + assert_eq!(stream::read_to_vec(payload).await?, b""); + assert_eq!(stored_metadata.content_type, metadata.content_type); + assert_eq!(stored_metadata.custom, metadata.custom); + + // An empty chunk is how a client uploads a zero-length object, so it has to finalize the + // session rather than be rejected or send a `Content-Range` naming a byte that is absent. + let chunked_id = make_id_with_key("resumable-zero-chunk"); + let token = backend + .create_upload_session(&chunked_id, &metadata, 0) + .await?; + assert_eq!( + backend + .put_chunk(&chunked_id, &token, 0, 0, stream::single(Vec::new())) + .await?, + UploadProgress::Complete + ); + let (_, _, payload) = backend.get_object(&chunked_id, None).await?.unwrap(); + assert_eq!(stream::read_to_vec(payload).await?, b""); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_empty_chunk_reports_offset_without_writing() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id_with_key("resumable-empty-chunk"); + let token = backend + .create_upload_session(&id, &Metadata::default(), 4) + .await?; + + // An empty chunk cannot advance a session that still expects bytes. It reports the + // authoritative offset instead of failing, and leaves what GCS holds untouched. + assert_eq!( + backend + .put_chunk(&id, &token, 0, 0, stream::single(Vec::new())) + .await?, + UploadProgress::Incomplete { offset: 0 } + ); + let after_write = backend + .put_chunk(&id, &token, 0, 2, stream::single(b"ab".to_vec())) + .await?; + assert!(matches!(after_write, UploadProgress::Incomplete { .. })); + + // Whichever prefix GCS acknowledged, an empty chunk reports that same position rather + // than moving it. GCS documents that a chunk "should be a multiple of 256 KiB ... unless + // it's the last chunk", and that a client "should not assume that the server received all + // bytes sent in any given request". The emulator acknowledges any length, so the position + // itself is not asserted here. + assert_eq!( + backend + .put_chunk(&id, &token, 2, 0, stream::single(Vec::new())) + .await?, + after_write + ); + assert_eq!(backend.upload_offset(&id, &token).await?, after_write); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_single_and_multi_chunk_uploads() -> Result<()> { + let backend = create_test_backend().await?; + + let single_id = make_id_with_key("resumable-single"); + let single = b"single chunk".to_vec(); + let token = backend + .create_upload_session(&single_id, &Metadata::default(), single.len() as u64) + .await?; + assert_eq!( + backend + .put_chunk( + &single_id, + &token, + 0, + single.len() as u64, + stream::single(single.clone()), + ) + .await?, + UploadProgress::Complete + ); + let (_, _, payload) = backend.get_object(&single_id, None).await?.unwrap(); + assert_eq!(stream::read_to_vec(payload).await?, single); + + let multi_id = make_id_with_key("resumable-multi"); + let mut expected = vec![b'a'; RESUMABLE_CHUNK_SIZE]; + expected.extend_from_slice(b"final"); + let token = backend + .create_upload_session(&multi_id, &Metadata::default(), expected.len() as u64) + .await?; + assert_eq!( + backend.upload_offset(&multi_id, &token).await?, + UploadProgress::Incomplete { offset: 0 } + ); + assert_eq!( + backend + .put_chunk( + &multi_id, + &token, + 0, + RESUMABLE_CHUNK_SIZE as u64, + stream::single(expected[..RESUMABLE_CHUNK_SIZE].to_vec()), + ) + .await?, + UploadProgress::Incomplete { + offset: RESUMABLE_CHUNK_SIZE as u64 + } + ); + assert_eq!( + backend.upload_offset(&multi_id, &token).await?, + UploadProgress::Incomplete { + offset: RESUMABLE_CHUNK_SIZE as u64 + } + ); + assert_eq!( + backend + .put_chunk( + &multi_id, + &token, + RESUMABLE_CHUNK_SIZE as u64, + 5, + stream::single(b"final".to_vec()), + ) + .await?, + UploadProgress::Complete + ); + let (_, _, payload) = backend.get_object(&multi_id, None).await?.unwrap(); + assert_eq!(stream::read_to_vec(payload).await?, expected); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_unaligned_chunk_and_rewind_preserve_persisted_bytes() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id_with_key("resumable-rewind"); + let total_length = RESUMABLE_CHUNK_SIZE + 3; + let token = backend + .create_upload_session(&id, &Metadata::default(), total_length as u64) + .await?; + + let prefix = vec![b'a'; RESUMABLE_CHUNK_SIZE]; + assert_eq!( + backend + .put_chunk(&id, &token, 0, prefix.len() as u64, stream::single(prefix),) + .await?, + UploadProgress::Incomplete { + offset: RESUMABLE_CHUNK_SIZE as u64 + } + ); + + // Resend three already-persisted positions with different bytes. GCS ignores that overlap + // without comparing it, then appends the suffix from its authoritative offset. + assert_eq!( + backend + .put_chunk( + &id, + &token, + (RESUMABLE_CHUNK_SIZE - 3) as u64, + 6, + stream::single(b"BADxyz".to_vec()), + ) + .await?, + UploadProgress::Complete + ); + + let (_, _, payload) = backend.get_object(&id, None).await?.unwrap(); + let payload = stream::read_to_vec(payload).await?; + assert_eq!(&payload[RESUMABLE_CHUNK_SIZE - 3..], b"aaaxyz"); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_rejects_oversized_chunks() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id_with_key("resumable-validation"); + let token = backend + .create_upload_session(&id, &Metadata::default(), 10) + .await?; + + let error = backend + .put_chunk(&id, &token, 8, 3, stream::single(b"abc".to_vec())) + .await + .unwrap_err(); + assert_eq!( + error.kind(), + ErrorKind::ChunkExceedsUploadLength { + offset: 8, + content_length: 3, + upload_length: 10 + } + ); + + let error = backend + .put_chunk(&id, &token, u64::MAX, 2, stream::single(b"ab".to_vec())) + .await + .unwrap_err(); + assert!(matches!( + error.kind(), + ErrorKind::ChunkExceedsUploadLength { .. } + )); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_cancel_is_idempotent_and_session_is_not_found() -> Result<()> { + let backend = create_test_backend().await?; + let id = make_id_with_key("resumable-cancel"); + let token = backend + .create_upload_session(&id, &Metadata::default(), 10) + .await?; + + backend.cancel_upload(&id, &token).await?; + backend.cancel_upload(&id, &token).await?; + assert!(matches!( + backend.upload_offset(&id, &token).await, + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + Ok(()) + } + + #[tokio::test] + async fn test_resumable_retries_only_replayable_operations() -> Result<()> { + let mut backend = create_test_backend().await?; + inject_retry_test(&mut backend, "storage.objects.insert", "return-503").await?; + let id = make_id_with_key("resumable-retries"); + + // Creation consumes the injected 503 and succeeds on the backend's retry. + let token = backend + .create_upload_session(&id, &Metadata::default(), 10) + .await?; + + inject_retry_test(&mut backend, "storage.objects.insert", "return-503").await?; + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Incomplete { offset: 0 } + ); + + inject_retry_test(&mut backend, "storage.objects.delete", "return-503").await?; + backend.cancel_upload(&id, &token).await?; + Ok(()) + } + + #[tokio::test] + async fn test_resumable_stream_failures_require_explicit_offset_recovery() -> Result<()> { + // A failure before persistence is returned directly: put_chunk must not retry the body. + let mut backend = create_test_backend().await?; + let id = make_id_with_key("resumable-failure-before"); + let token = backend + .create_upload_session(&id, &Metadata::default(), 4) + .await?; + inject_retry_test(&mut backend, "storage.objects.insert", "return-503").await?; + assert!(matches!( + backend + .put_chunk(&id, &token, 0, 4, stream::single(b"data".to_vec())) + .await, + Err(error) if error.kind() == ErrorKind::BackendUnavailable + )); + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Incomplete { offset: 0 } + ); + + // The emulator persists the first KiB, then fails. Recovery observes that prefix and the + // caller resumes exactly from the returned offset. + let mut backend = create_test_backend().await?; + let id = make_id_with_key("resumable-failure-partial"); + let data = vec![b'p'; 2048]; + let token = backend + .create_upload_session(&id, &Metadata::default(), data.len() as u64) + .await?; + inject_retry_test( + &mut backend, + "storage.objects.insert", + "return-503-after-1K", + ) + .await?; + assert!(matches!( + backend + .put_chunk( + &id, + &token, + 0, + data.len() as u64, + stream::single(data.clone()), + ) + .await, + Err(error) if error.kind() == ErrorKind::BackendUnavailable + )); + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Incomplete { offset: 1024 } + ); + assert_eq!( + backend + .put_chunk( + &id, + &token, + 1024, + 1024, + stream::single(data[1024..].to_vec()), + ) + .await?, + UploadProgress::Complete + ); + + // GCS persists the final bytes, but storage-testbench truncates the successful JSON + // response. The chunk is not retried; an explicit status query observes completion. + let mut backend = create_test_backend().await?; + let id = make_id_with_key("resumable-failure-final"); + let token = backend + .create_upload_session(&id, &Metadata::default(), 5) + .await?; + inject_retry_test( + &mut backend, + "storage.objects.insert", + "return-broken-stream-final-chunk-after-0B", + ) + .await?; + assert!(matches!( + backend + .put_chunk(&id, &token, 0, 5, stream::single(b"final".to_vec())) + .await, + Err(error) if error.kind() == ErrorKind::CorruptData + )); + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Complete + ); + Ok(()) + } + async fn get_generation_matches( backend: &GcsBackend, object_url: Url, @@ -2189,6 +2966,88 @@ mod tests { Ok(()) } + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn resumable_completion_reports_to_change_stream() -> Result<()> { + let (backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id(); + let payload = b"resumable payload".to_vec(); + let metadata = Metadata { + time_expires: Some(SystemTime::now() + Duration::from_secs(3600)), + ..Default::default() + }; + let token = backend + .create_upload_session(&id, &metadata, payload.len() as u64) + .await?; + + assert_eq!( + backend + .put_chunk( + &id, + &token, + 0, + payload.len() as u64, + stream::single::(payload.clone()), + ) + .await?, + UploadProgress::Complete + ); + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].op_type, OpType::Write); + assert_eq!( + records[0].size, + Some(payload.len() as u64 + GcsObject::from_metadata(&metadata).metadata_size()) + ); + assert!(records[0].expiration_time.is_some()); + Ok(()) + } + + #[cfg(feature = "storage-cogs")] + #[tokio::test] + async fn resumable_completion_after_corrupt_response_reports_to_change_stream() -> Result<()> { + let (mut backend, producer) = create_test_backend_with_change_stream().await?; + let id = make_id_with_key("resumable-change-stream-after-corrupt-response"); + let payload = b"final".to_vec(); + let metadata = Metadata::default(); + let token = backend + .create_upload_session(&id, &metadata, payload.len() as u64) + .await?; + inject_retry_test( + &mut backend, + "storage.objects.insert", + "return-broken-stream-final-chunk-after-0B", + ) + .await?; + + assert!(matches!( + backend + .put_chunk( + &id, + &token, + 0, + payload.len() as u64, + stream::single::(payload.clone()), + ) + .await, + Err(error) if error.kind() == ErrorKind::CorruptData + )); + assert_eq!( + backend.upload_offset(&id, &token).await?, + UploadProgress::Complete + ); + + let records = producer.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].op_type, OpType::Write); + assert_eq!( + records[0].size, + Some(payload.len() as u64 + GcsObject::from_metadata(&metadata).metadata_size()) + ); + Ok(()) + } + #[cfg(feature = "storage-cogs")] #[tokio::test] async fn change_stream_size_includes_metadata_keys_and_values() -> Result<()> { diff --git a/objectstore-service/src/backend/testing.rs b/objectstore-service/src/backend/testing.rs index 4b4730e4..95a8f2f5 100644 --- a/objectstore-service/src/backend/testing.rs +++ b/objectstore-service/src/backend/testing.rs @@ -40,7 +40,7 @@ use bytes::Bytes; use objectstore_types::metadata::Metadata; use objectstore_types::range::ByteRange; -use objectstore_types::resumable::{SessionToken, UploadProgress}; +use objectstore_types::resumable::UploadProgress; use crate::backend::common::{ Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, @@ -53,6 +53,7 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::BackendToken; use crate::stream::ClientStream; /// Hooks for [`TestBackend`]. @@ -248,7 +249,7 @@ pub trait Hooks: fmt::Debug + Send + Sync + 'static { id: &ObjectId, metadata: &Metadata, total_length: u64, - ) -> Result> { + ) -> Result> { inner .create_upload_session(id, metadata, total_length) .await @@ -259,7 +260,7 @@ pub trait Hooks: fmt::Debug + Send + Sync + 'static { &self, inner: &InMemoryBackend, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, offset: u64, content_length: u64, stream: ClientStream, @@ -274,7 +275,7 @@ pub trait Hooks: fmt::Debug + Send + Sync + 'static { &self, inner: &InMemoryBackend, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, ) -> Result { inner.upload_offset(id, session).await } @@ -284,7 +285,7 @@ pub trait Hooks: fmt::Debug + Send + Sync + 'static { &self, inner: &InMemoryBackend, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, ) -> Result<()> { inner.cancel_upload(id, session).await } @@ -368,7 +369,7 @@ impl Backend for TestBackend { id: &ObjectId, metadata: &Metadata, total_length: u64, - ) -> Result> { + ) -> Result> { self.hooks .create_upload_session(&self.inner, id, metadata, total_length) .await @@ -377,7 +378,7 @@ impl Backend for TestBackend { async fn put_chunk( &self, id: &ObjectId, - session: &SessionToken, + session: &BackendToken, offset: u64, content_length: u64, stream: ClientStream, @@ -387,11 +388,11 @@ impl Backend for TestBackend { .await } - async fn upload_offset(&self, id: &ObjectId, session: &SessionToken) -> Result { + async fn upload_offset(&self, id: &ObjectId, session: &BackendToken) -> Result { self.hooks.upload_offset(&self.inner, id, session).await } - async fn cancel_upload(&self, id: &ObjectId, session: &SessionToken) -> Result<()> { + async fn cancel_upload(&self, id: &ObjectId, session: &BackendToken) -> Result<()> { self.hooks.cancel_upload(&self.inner, id, session).await } } diff --git a/objectstore-service/src/lib.rs b/objectstore-service/src/lib.rs index a7a82b6d..c81101c0 100644 --- a/objectstore-service/src/lib.rs +++ b/objectstore-service/src/lib.rs @@ -9,6 +9,7 @@ pub mod error; mod gcp_auth; pub mod id; pub mod multipart; +pub mod resumable; pub mod service; pub mod stream; pub mod streaming; diff --git a/objectstore-service/src/resumable.rs b/objectstore-service/src/resumable.rs new file mode 100644 index 00000000..9d1642ef --- /dev/null +++ b/objectstore-service/src/resumable.rs @@ -0,0 +1,367 @@ +//! Utilities for Resumable Uploads. + +use std::collections::BTreeMap; +use std::fmt; + +use ring::aead::{AES_256_GCM, Aad, LessSafeKey, Nonce, UnboundKey}; +use ring::rand::{SecureRandom, SystemRandom}; +use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; + +use crate::error::{Error, ErrorKind, Result, ResultExt as _}; +use crate::id::ObjectId; + +pub use objectstore_types::resumable::{ + SessionToken as EncryptedSessionToken, UploadOffset, UploadProgress, +}; + +/// AES-GCM nonce length in bytes. +const NONCE_LENGTH: usize = 12; +/// AES-GCM authentication tag length in bytes. +const TAG_LENGTH: usize = 16; + +/// Opaque session state encoded and decoded by a storage backend. +pub type BackendToken = String; + +/// Structured token encrypted at the service boundary. +#[derive(Deserialize, Serialize)] +pub(crate) struct SessionToken { + #[serde( + serialize_with = "serialize_object_id", + deserialize_with = "deserialize_object_id" + )] + pub(crate) object_id: ObjectId, + pub(crate) backend_token: BackendToken, +} + +fn serialize_object_id(id: &ObjectId, serializer: S) -> std::result::Result +where + S: Serializer, +{ + serializer.collect_str(&id.as_storage_path()) +} + +fn deserialize_object_id<'de, D>(deserializer: D) -> std::result::Result +where + D: Deserializer<'de>, +{ + let path = String::deserialize(deserializer)?; + ObjectId::from_storage_path(&path) + .ok_or_else(|| de::Error::custom("invalid object storage path")) +} + +/// Encrypts and decrypts Resumable Upload session tokens. +/// +/// The active key encrypts new sessions, while the key ID embedded in an existing token selects +/// its decryption key. +pub struct Encryptor { + active_key_id: String, + active_key: LessSafeKey, + decryption_keys: BTreeMap, + random: SystemRandom, +} + +impl Encryptor { + /// Constructs an encryptor with a fresh process-local AES-256 key. + /// + /// Tokens encrypted with this key cannot be decrypted after a service restart. Configure a + /// persistent keyring with [`Self::new`] when resumable sessions must survive restarts. + /// + /// Returns an error if secure random key generation fails. + pub fn ephemeral() -> anyhow::Result { + let key_id = "ephemeral"; + let random = SystemRandom::new(); + let mut key = [0; 32]; + random + .fill(&mut key) + .map_err(|_| anyhow::anyhow!("failed to generate resumable token encryption key"))?; + Self::new(key_id, BTreeMap::from([(key_id.to_owned(), key.to_vec())])) + } + + /// Validates and constructs an encryptor from raw AES-256 keys. + /// + /// Returns an error for invalid key IDs or sizes, or when the active key is absent. + pub fn new( + active_key_id: impl Into, + keys: BTreeMap>, + ) -> anyhow::Result { + let active_key_id = active_key_id.into(); + validate_key_id(&active_key_id)?; + + let mut validated = BTreeMap::new(); + for (key_id, key) in keys { + validate_key_id(&key_id)?; + anyhow::ensure!( + key.len() == AES_256_GCM.key_len(), + "resumable token encryption key {key_id:?} must contain exactly 32 bytes, got {}", + key.len() + ); + let key = UnboundKey::new(&AES_256_GCM, &key) + .map(LessSafeKey::new) + .map_err(|_| anyhow::anyhow!("invalid resumable token encryption key material"))?; + validated.insert(key_id, key); + } + + let active_key = validated.remove(&active_key_id).ok_or_else(|| { + anyhow::anyhow!( + "active resumable token encryption key {active_key_id:?} is not configured" + ) + })?; + + Ok(Self { + active_key_id, + active_key, + decryption_keys: validated, + random: SystemRandom::new(), + }) + } + + /// Encrypts a structured Resumable Upload session token. + pub(crate) fn encrypt(&self, session: SessionToken) -> Result { + let key_id = self.active_key_id.as_bytes(); + let key_id_length = u8::try_from(key_id.len()).context( + ErrorKind::Internal, + "resumable token encryption key ID exceeds maximum length", + )?; + + let mut header = Vec::with_capacity(1 + key_id.len()); + header.push(key_id_length); + header.extend_from_slice(key_id); + + let mut ciphertext = serde_json::to_vec(&session).context( + ErrorKind::Internal, + "failed to serialize resumable session token", + )?; + + let mut nonce = [0; NONCE_LENGTH]; + self.random.fill(&mut nonce).map_err(|_| { + Error::new( + ErrorKind::Internal, + "failed to generate resumable token nonce", + ) + })?; + self.active_key + .seal_in_place_append_tag( + Nonce::assume_unique_for_key(nonce), + Aad::from(&header), + &mut ciphertext, + ) + .map_err(|_| Error::new(ErrorKind::Internal, "failed to encrypt resumable token"))?; + + let mut envelope = Vec::with_capacity(header.len() + nonce.len() + ciphertext.len()); + envelope.extend_from_slice(&header); + envelope.extend_from_slice(&nonce); + envelope.extend_from_slice(&ciphertext); + Ok(EncryptedSessionToken::new(envelope)) + } + + /// Decrypts a Resumable Upload session token. + pub(crate) fn decrypt(&self, token: EncryptedSessionToken) -> Result { + self.decrypt_inner(token) + .ok_or_else(|| ErrorKind::UnknownUploadSession.into()) + } + + fn decrypt_inner(&self, token: EncryptedSessionToken) -> Option { + let envelope = token.into_bytes(); + let (&key_id_length, rest) = envelope.split_first()?; + let key_id_length = usize::from(key_id_length); + if key_id_length == 0 { + return None; + } + let (key_id, rest) = rest.split_at_checked(key_id_length)?; + let (nonce, ciphertext) = rest.split_at_checked(NONCE_LENGTH)?; + + let key_id = std::str::from_utf8(key_id).ok()?; + let key = if key_id == self.active_key_id { + &self.active_key + } else { + self.decryption_keys.get(key_id)? + }; + let header_length = 1 + key_id_length; + let header = &envelope[..header_length]; + let nonce: [u8; NONCE_LENGTH] = nonce.try_into().ok()?; + if ciphertext.len() < TAG_LENGTH { + return None; + } + let mut ciphertext = ciphertext.to_vec(); + let plaintext = key + .open_in_place( + Nonce::assume_unique_for_key(nonce), + Aad::from(header), + &mut ciphertext, + ) + .ok()?; + serde_json::from_slice(plaintext).ok() + } +} + +impl fmt::Debug for Encryptor { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut key_ids = self.decryption_keys.keys().collect::>(); + key_ids.push(&self.active_key_id); + key_ids.sort_unstable(); + + f.debug_struct("Encryptor") + .field("active_key_id", &self.active_key_id) + .field("key_ids", &key_ids) + .field("keys", &"[redacted]") + .finish() + } +} + +fn validate_key_id(key_id: &str) -> anyhow::Result<()> { + let key_id_length = key_id.len(); + u8::try_from(key_id_length).map_err(|_| { + anyhow::anyhow!( + "resumable token encryption key ID must be at most {} bytes, got {key_id_length}", + u8::MAX + ) + })?; + anyhow::ensure!( + !key_id.is_empty() + && key_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')), + "invalid resumable token encryption key ID {key_id:?}" + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use objectstore_types::scope::{Scope, Scopes}; + + use super::*; + use crate::id::ObjectContext; + + fn id(key: &str) -> ObjectId { + ObjectId::new( + ObjectContext { + usecase: "testing".into(), + scopes: Scopes::from_iter([Scope::create("org", "42").unwrap()]), + }, + key.into(), + ) + } + + fn encryption(active: &str, keys: &[(&str, u8)]) -> Encryptor { + Encryptor::new( + active, + keys.iter() + .map(|(key_id, byte)| (key_id.to_string(), vec![*byte; 32])) + .collect(), + ) + .unwrap() + } + + #[test] + fn encryption_is_randomized_and_round_trips_backend_tokens() { + let encryption = encryption("v1", &[("v1", 7)]); + let id = id("object"); + let backend_token = "backend token".to_owned(); + + let first = encryption + .encrypt(SessionToken { + object_id: id.clone(), + backend_token: backend_token.clone(), + }) + .unwrap(); + let second = encryption + .encrypt(SessionToken { + object_id: id.clone(), + backend_token: backend_token.clone(), + }) + .unwrap(); + assert_ne!(first, second); + let first = encryption.decrypt(first).unwrap(); + let second = encryption.decrypt(second).unwrap(); + assert_eq!(first.object_id, id); + assert_eq!(first.backend_token, backend_token); + assert_eq!(second.object_id, id); + assert_eq!(second.backend_token, backend_token); + } + + #[test] + fn encryption_rejects_tampering_and_plaintext() { + let encryption = encryption("v1", &[("v1", 7)]); + let object = id("object"); + let token = encryption + .encrypt(SessionToken { + object_id: object, + backend_token: "backend token".to_owned(), + }) + .unwrap(); + + let mut tampered = token.clone().into_bytes(); + *tampered.last_mut().unwrap() ^= 1; + assert!(matches!( + encryption.decrypt(EncryptedSessionToken::new(tampered)), + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + assert!(matches!( + encryption.decrypt(EncryptedSessionToken::new(b"backend token")), + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + } + + #[test] + fn rotation_decrypts_old_keys_and_removal_invalidates_them() { + let object = id("object"); + let old = encryption("v1", &[("v1", 1)]); + let old_token = old + .encrypt(SessionToken { + object_id: object.clone(), + backend_token: "backend token".to_owned(), + }) + .unwrap(); + + let rotated = encryption("v2", &[("v1", 1), ("v2", 2)]); + assert_eq!( + rotated.decrypt(old_token.clone()).unwrap().backend_token, + "backend token" + ); + let new_token = rotated + .encrypt(SessionToken { + object_id: object, + backend_token: "new token".to_owned(), + }) + .unwrap(); + assert_eq!(new_token.as_bytes()[1..3], *b"v2"); + + let removed = encryption("v2", &[("v2", 2)]); + assert!(matches!( + removed.decrypt(old_token), + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + } + + #[test] + fn configuration_validates_ids_lengths_and_active_key() { + let error = Encryptor::new("missing", BTreeMap::new()).unwrap_err(); + assert_eq!( + error.to_string(), + "active resumable token encryption key \"missing\" is not configured" + ); + + let error = Encryptor::new("bad key", BTreeMap::from([("bad key".into(), vec![0; 32])])) + .unwrap_err(); + assert_eq!( + error.to_string(), + "invalid resumable token encryption key ID \"bad key\"" + ); + + let error = Encryptor::new("v1", BTreeMap::from([("v1".into(), vec![0; 31])])).unwrap_err(); + assert_eq!( + error.to_string(), + "resumable token encryption key \"v1\" must contain exactly 32 bytes, got 31" + ); + } + + #[test] + fn debug_output_redacts_keys() { + let encryption = encryption("v1", &[("v1", 7)]); + let debug = format!("{encryption:?}"); + assert!(debug.contains("v1")); + assert!(debug.contains("[redacted]")); + assert!(!debug.contains("7, 7")); + } +} diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 9cd48288..5355b455 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use objectstore_types::metadata::Metadata; use objectstore_types::range::{ByteRange, ContentRange}; -use objectstore_types::resumable::{SessionToken, UploadProgress}; +use objectstore_types::resumable::{SessionToken as EncryptedSessionToken, UploadProgress}; use crate::backend::common::Backend; use crate::backend::counting::CountingBackend; @@ -21,6 +21,7 @@ use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, ListPartsResponse, PartNumber, UploadId, UploadPartResponse, }; +use crate::resumable::{BackendToken, Encryptor, SessionToken}; use crate::stream::{ClientStream, PayloadStream}; use crate::streaming::StreamExecutor; @@ -74,6 +75,7 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; pub struct StorageService { inner: Arc, concurrency: ConcurrencyLimiter, + resumable_token_encryption: Arc, } impl StorageService { @@ -83,10 +85,13 @@ impl StorageService { /// each operation run. Single-object operations served directly by `StorageService` are covered /// as we batched operations served by [`StreamExecutor`]. See /// [`backend::counting`](crate::backend::counting) for details. - pub fn new(backend: Box) -> Self { + /// + /// `resumable_token_encryption` protects tokens exposed by the resumable upload methods. + pub fn new(backend: Box, resumable_token_encryption: Encryptor) -> Self { Self { inner: Arc::new(CountingBackend::new(backend)), concurrency: ConcurrencyLimiter::new(DEFAULT_CONCURRENCY_LIMIT), + resumable_token_encryption: Arc::new(resumable_token_encryption), } } @@ -372,33 +377,55 @@ impl StorageService { id: ObjectId, metadata: Metadata, total_length: u64, - ) -> Result> { + ) -> Result> { metadata.validate().kind(ErrorKind::InvalidMetadata)?; let inner = Arc::clone(&self.inner); + let encryption = self.resumable_token_encryption.clone(); self.spawn("create_upload_session", async move { - inner + let session = inner .create_upload_session(&id, &metadata, total_length) - .await + .await?; + session + .map(|backend_token| { + encryption.encrypt(SessionToken { + object_id: id, + backend_token, + }) + }) + .transpose() }) .await } + fn backend_token_for( + &self, + expected_id: &ObjectId, + token: EncryptedSessionToken, + ) -> Result { + let session = self.resumable_token_encryption.decrypt(token)?; + if session.object_id != *expected_id { + return Err(ErrorKind::UnknownUploadSession.into()); + } + Ok(session.backend_token) + } + /// Writes a chunk of `content_length` bytes at `offset` into an open session. /// - /// Commits the object once the chunk carrying the last byte is persisted. + /// Completes the upload once the chunk carrying the last byte is persisted. /// /// # Run-to-completion /// /// Once called, the operation runs to completion even if the returned future is dropped. - /// This matters most for the final chunk, which commits the object. + /// This matters most for the final chunk, which completes the upload. pub async fn put_chunk( &self, id: ObjectId, - session: SessionToken, + session: EncryptedSessionToken, offset: u64, content_length: u64, body: ClientStream, ) -> Result { + let session = self.backend_token_for(&id, session)?; let inner = Arc::clone(&self.inner); self.spawn("put_chunk", async move { inner @@ -408,14 +435,17 @@ impl StorageService { .await } - /// Reports how far a session has progressed, committing the object if it is assembled. + /// Reports how far a session has progressed. /// - /// This can mutate state and therefore requires write permission at the API layer. + /// This can observe completion after the final chunk's response was lost. A composed backend + /// may also finish pending publication work, so this requires write permission at the API + /// layer. pub async fn upload_offset( &self, id: ObjectId, - session: SessionToken, + session: EncryptedSessionToken, ) -> Result { + let session = self.backend_token_for(&id, session)?; let inner = Arc::clone(&self.inner); self.spawn("upload_offset", async move { inner.upload_offset(&id, &session).await @@ -424,7 +454,8 @@ impl StorageService { } /// Cancels an upload session, discarding whatever was uploaded. - pub async fn cancel_upload(&self, id: ObjectId, session: SessionToken) -> Result<()> { + pub async fn cancel_upload(&self, id: ObjectId, session: EncryptedSessionToken) -> Result<()> { + let session = self.backend_token_for(&id, session)?; let inner = Arc::clone(&self.inner); self.spawn("cancel_upload", async move { inner.cancel_upload(&id, &session).await @@ -435,7 +466,7 @@ impl StorageService { #[cfg(test)] mod tests { - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use std::time::Duration; use bytes::BytesMut; @@ -455,6 +486,34 @@ mod tests { use crate::change_stream::ChangeStreamFactory; use crate::stream::{self, ClientStream}; + #[derive(Clone, Debug, Default)] + struct ResumableTokenHooks { + seen_tokens: Arc>>, + } + + #[async_trait::async_trait] + impl Hooks for ResumableTokenHooks { + async fn create_upload_session( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + _metadata: &Metadata, + _total_length: u64, + ) -> Result> { + Ok(Some("backend token".to_owned())) + } + + async fn upload_offset( + &self, + _inner: &InMemoryBackend, + _id: &ObjectId, + session: &BackendToken, + ) -> Result { + self.seen_tokens.lock().unwrap().push(session.to_owned()); + Ok(UploadProgress::Incomplete { offset: 0 }) + } + } + fn make_context() -> ObjectContext { ObjectContext { usecase: "testing".into(), @@ -463,7 +522,10 @@ mod tests { } fn make_service() -> StorageService { - StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) + StorageService::new( + Box::new(InMemoryBackend::new("in-memory")), + Encryptor::ephemeral().unwrap(), + ) } #[tokio::test] @@ -514,7 +576,7 @@ mod tests { let backend = GcsBackend::new(config, &ChangeStreamFactory::default()) .await .unwrap(); - let service = StorageService::new(Box::new(backend)); + let service = StorageService::new(Box::new(backend), Encryptor::ephemeral().unwrap()); let key = service .insert_object( @@ -559,7 +621,7 @@ mod tests { .unwrap(), ); let backend = TieredStorage::new(high_volume, long_term, Box::new(NoopChangeLog)); - let service = StorageService::new(Box::new(backend)); + let service = StorageService::new(Box::new(backend), Encryptor::ephemeral().unwrap()); // A separate GCS backend to directly inspect the long-term storage. let gcs_backend = GcsBackend::new(gcs_config.clone(), &ChangeStreamFactory::default()) @@ -658,7 +720,10 @@ mod tests { #[tokio::test] async fn panic_in_backend_returns_task_failed() { - let service = StorageService::new(Box::new(TestBackend::new(PanicOnGet))); + let service = StorageService::new( + Box::new(TestBackend::new(PanicOnGet)), + Encryptor::ephemeral().unwrap(), + ); let id = ObjectId::new(make_context(), "panic-test".into()); let result = service.get_object(id, None).await; @@ -733,7 +798,7 @@ mod tests { let hv = Box::new(TestBackend::new(GateOnPut::default())); let lt = Box::new(TestBackend::new(GateOnPut::with_pause())); let backend = TieredStorage::new(hv.clone(), lt.clone(), Box::new(NoopChangeLog)); - let service = StorageService::new(Box::new(backend)); + let service = StorageService::new(Box::new(backend), Encryptor::ephemeral().unwrap()); let payload = vec![0xABu8; 2 * 1024 * 1024]; // 2 MiB → long-term path let request = service.insert_object( @@ -775,8 +840,9 @@ mod tests { fn make_limited_service(limit: u32) -> (StorageService, TestBackend) { let backend = TestBackend::new(GateOnPut::with_pause()); - let service = StorageService::new(Box::new(backend.clone())) - .with_concurrency(ConcurrencyLimiter::new(limit)); + let service = + StorageService::new(Box::new(backend.clone()), Encryptor::ephemeral().unwrap()) + .with_concurrency(ConcurrencyLimiter::new(limit)); (service, backend) } @@ -830,7 +896,8 @@ mod tests { #[tokio::test] async fn tasks_limit_returns_configured_limit() { let backend = Box::new(InMemoryBackend::new("cap")); - let service = StorageService::new(backend).with_concurrency(ConcurrencyLimiter::new(7)); + let service = StorageService::new(backend, Encryptor::ephemeral().unwrap()) + .with_concurrency(ConcurrencyLimiter::new(7)); assert_eq!(service.tasks_limit(), 7); } @@ -860,8 +927,11 @@ mod tests { #[tokio::test] async fn permits_released_after_panic() { - let service = StorageService::new(Box::new(TestBackend::new(PanicOnGet))) - .with_concurrency(ConcurrencyLimiter::new(1)); + let service = StorageService::new( + Box::new(TestBackend::new(PanicOnGet)), + Encryptor::ephemeral().unwrap(), + ) + .with_concurrency(ConcurrencyLimiter::new(1)); // First operation panics — the permit must still be released. let id = ObjectId::new(make_context(), "panic-permit".into()); @@ -905,4 +975,80 @@ mod tests { let result = service.create_upload_session(id, metadata, 1024).await; assert!(result.is_err_and(|error| error.kind() == ErrorKind::InvalidMetadata)); } + + #[tokio::test] + async fn resumable_tokens_are_encrypted_by_default() -> Result<()> { + let hooks = ResumableTokenHooks::default(); + let service = StorageService::new( + Box::new(TestBackend::new(hooks.clone())), + Encryptor::ephemeral().unwrap(), + ); + let id = ObjectId::new(make_context(), "resumable".into()); + + let token = service + .create_upload_session(id.clone(), Metadata::default(), 4) + .await? + .expect("test backend supports resumable uploads"); + assert_ne!(token.as_bytes(), b"backend token"); + assert!(matches!( + service + .upload_offset(id.clone(), EncryptedSessionToken::new(b"backend token")) + .await, + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + let other_id = ObjectId::new(make_context(), "other".into()); + assert!(matches!( + service.upload_offset(other_id, token.clone()).await, + Err(error) if error.kind() == ErrorKind::UnknownUploadSession + )); + assert!(hooks.seen_tokens.lock().unwrap().is_empty()); + service.upload_offset(id, token).await?; + assert_eq!( + hooks.seen_tokens.lock().unwrap().as_slice(), + &["backend token"] + ); + Ok(()) + } + + #[tokio::test] + async fn configured_encryption_only_crosses_the_service_boundary() -> Result<()> { + let hooks = ResumableTokenHooks::default(); + let encryption = Encryptor::new( + "v1", + std::collections::BTreeMap::from([("v1".into(), vec![7; 32])]), + ) + .unwrap(); + let service = StorageService::new(Box::new(TestBackend::new(hooks.clone())), encryption); + let id = ObjectId::new(make_context(), "resumable".into()); + + let encrypted = service + .create_upload_session(id.clone(), Metadata::default(), 4) + .await? + .expect("test backend supports resumable uploads"); + assert_ne!(encrypted.as_bytes(), b"backend token"); + service.upload_offset(id, encrypted).await?; + assert_eq!( + hooks.seen_tokens.lock().unwrap().as_slice(), + &["backend token"] + ); + Ok(()) + } + + #[tokio::test] + async fn configured_encryption_rejects_plaintext_tokens() { + let hooks = ResumableTokenHooks::default(); + let encryption = Encryptor::new( + "v1", + std::collections::BTreeMap::from([("v1".into(), vec![7; 32])]), + ) + .unwrap(); + let service = StorageService::new(Box::new(TestBackend::new(hooks.clone())), encryption); + let id = ObjectId::new(make_context(), "resumable".into()); + + let result = service + .upload_offset(id, EncryptedSessionToken::new(b"backend token")) + .await; + assert!(result.is_err_and(|error| error.kind() == ErrorKind::UnknownUploadSession)); + assert!(hooks.seen_tokens.lock().unwrap().is_empty()); + } } diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index 5ce07088..7f6402ba 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -344,6 +344,7 @@ mod tests { use crate::backend::testing::{Hooks, TestBackend}; use crate::concurrency::ConcurrencyLimiter; use crate::error::{Error, ErrorKind}; + use crate::resumable::Encryptor; use crate::service::StorageService; use crate::stream::{self, ClientStream}; @@ -355,8 +356,11 @@ mod tests { } fn make_service_with_limit(limit: u32) -> StorageService { - StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) - .with_concurrency(ConcurrencyLimiter::new(limit)) + StorageService::new( + Box::new(InMemoryBackend::new("in-memory")), + Encryptor::ephemeral().unwrap(), + ) + .with_concurrency(ConcurrencyLimiter::new(limit)) } fn make_service() -> StorageService { @@ -519,8 +523,8 @@ mod tests { resume: Arc::clone(&resume), in_flight: Arc::clone(&in_flight), }); - let service = - StorageService::new(Box::new(gated)).with_concurrency(ConcurrencyLimiter::new(100)); + let service = StorageService::new(Box::new(gated), Encryptor::ephemeral().unwrap()) + .with_concurrency(ConcurrencyLimiter::new(100)); let ops: Vec = (0..10) .map(|i| { @@ -563,13 +567,16 @@ mod tests { async fn bulk_respects_budget() { // Bulk budget = 1 (100% of max=1). Hold the permit via a normal // acquire; the bulk op should wait and eventually time out. - let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) - .with_concurrency( - ConcurrencyLimiter::new(1) - .with_queue(0) - .with_timeout(Duration::from_millis(1)) - .with_bulk(100), - ); + let service = StorageService::new( + Box::new(InMemoryBackend::new("in-memory")), + Encryptor::ephemeral().unwrap(), + ) + .with_concurrency( + ConcurrencyLimiter::new(1) + .with_queue(0) + .with_timeout(Duration::from_millis(1)) + .with_bulk(100), + ); let _held = service.concurrency_limiter().acquire().await.unwrap(); diff --git a/objectstore-types/src/resumable.rs b/objectstore-types/src/resumable.rs index 89bbaf13..d1b6ea1d 100644 --- a/objectstore-types/src/resumable.rs +++ b/objectstore-types/src/resumable.rs @@ -7,16 +7,18 @@ //! The client then sends chunks with [`HEADER_UPLOAD_OFFSET`] set to the byte position at which //! each chunk starts. If an upload is interrupted, the client can send the wildcard offset //! [`UploadOffset::Unknown`] to query the server's authoritative position before resuming. The -//! request that completes the object returns a [`CommitResponse`]. +//! request that completes the upload returns a [`CompleteUploadResponse`]. //! -//! Session tokens are defined by the storage backend and clients must treat their contents as -//! opaque. The token's UTF-8 bytes are encoded as unpadded base64url when the token is placed in a -//! request's `session` query parameter. +//! Session tokens contain the canonical object path and backend state protected by the storage +//! service, and clients must treat their contents as opaque. The token bytes are encoded as +//! unpadded base64url when the token is placed in a request's `session` query parameter. -use std::fmt; use std::str::FromStr; +use std::{borrow::Cow, fmt}; -use serde::{Deserialize, Serialize}; +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; /// Request header declaring the total size of the object, in bytes. /// @@ -35,48 +37,75 @@ const OFFSET_WILDCARD: &str = "*"; /// Identifier for an in-progress resumable upload session. /// -/// The token is an opaque identifier whose contents are defined and interpreted by the storage -/// backend. -#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(transparent)] -pub struct SessionToken(String); +/// Internally, this is an opaque byte string interpreted by the storage service. At the HTTP API +/// boundary it serializes as canonical unpadded base64url, so the serialized value can be placed +/// directly in a subsequent request URL. +#[derive(Clone, PartialEq, Eq)] +pub struct SessionToken(Vec); impl SessionToken { - /// Returns the token as a string slice. - pub fn as_str(&self) -> &str { + /// Wraps opaque session-token bytes. + pub fn new(bytes: impl Into>) -> Self { + Self(bytes.into()) + } + + /// Returns the opaque token bytes. + pub fn as_bytes(&self) -> &[u8] { &self.0 } - /// Consumes the wrapper and returns the backend-defined token. - pub fn into_inner(self) -> String { + /// Consumes the token and returns its opaque bytes. + pub fn into_bytes(self) -> Vec { self.0 } -} -impl AsRef for SessionToken { - fn as_ref(&self) -> &str { - self.as_str() + /// Parses the canonical unpadded-base64url representation used by the HTTP API. + pub fn from_base64url(encoded: &str) -> Result { + let bytes = URL_SAFE_NO_PAD + .decode(encoded) + .map_err(|_| InvalidSessionToken)?; + if URL_SAFE_NO_PAD.encode(&bytes) != encoded { + return Err(InvalidSessionToken); + } + Ok(Self(bytes)) + } + + /// Encodes this token for the HTTP API. + pub fn to_base64url(&self) -> String { + URL_SAFE_NO_PAD.encode(&self.0) } } -impl From for SessionToken { - fn from(token: String) -> Self { - Self(token) +impl fmt::Debug for SessionToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("SessionToken") } } -impl From<&str> for SessionToken { - fn from(token: &str) -> Self { - Self(token.to_owned()) +impl Serialize for SessionToken { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_base64url()) } } -impl fmt::Display for SessionToken { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.0.fmt(f) +impl<'de> Deserialize<'de> for SessionToken { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let encoded = Cow::<'static, String>::deserialize(deserializer)?; + Self::from_base64url(&encoded).map_err(de::Error::custom) } } +/// Error returned for a non-canonical or malformed external session token. +#[derive(Debug, thiserror::Error)] +#[error("session token must use unpadded base64url encoding")] +pub struct InvalidSessionToken; + /// The value of the [`HEADER_UPLOAD_OFFSET`] request header. /// /// In a request, a concrete offset submits a chunk starting at that byte, @@ -124,22 +153,28 @@ impl fmt::Display for UploadOffset { /// How far a resumable upload has progressed. /// -/// Both a chunk write and an offset query can commit an object, so both operations have the same -/// two outcomes. +/// Both a chunk write and an offset query can observe that an upload is complete, so both +/// operations have the same two outcomes. Completion is relative to the backend handling the +/// operation: it means the session is terminal and the object is available through that backend's +/// normal read methods. A backend that composes another backend must finish its own publication +/// work before returning [`UploadProgress::Complete`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum UploadProgress { /// More bytes are expected. The client continues from `offset`. /// /// This offset is authoritative and may be lower than the end of the chunk that was just /// written: backends can persist only a prefix and discard the remainder. It must remain below - /// the session's total length; once every byte has landed, the backend commits the object or + /// the session's total length; once every byte has landed, the backend completes the upload or /// returns an error instead. Incomplete { /// The offset the backend has persisted. offset: u64, }, - /// The last byte arrived and the object is committed and readable. - Committed, + /// The session is terminal and the object is available through the backend's normal reads. + /// + /// This is an observable status rather than a one-time event. A later offset query can return + /// `Complete` again, for example when the response to the final chunk was lost. + Complete, } /// Response from creating a resumable upload session. @@ -151,12 +186,12 @@ pub struct CreateSessionResponse { pub session: SessionToken, } -/// Response from the request that commits the object. +/// Response from the request that completes the upload. /// /// This is either the chunk carrying the last byte, or an offset query against a -/// session whose object was already assembled. +/// session whose final chunk completed but whose response was not observed. #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CommitResponse { +pub struct CompleteUploadResponse { /// The object key. pub key: String, } @@ -166,26 +201,36 @@ mod tests { use super::*; #[test] - fn create_session_response_serializes_token_verbatim() -> Result<(), serde_json::Error> { + fn create_session_response_encodes_token_once() -> Result<(), serde_json::Error> { let response = CreateSessionResponse { key: "key".into(), - session: "../opaque +? ü".into(), + session: SessionToken::new(b"../opaque +? \xc3\xbc"), }; assert_eq!( serde_json::to_string(&response)?, - r#"{"key":"key","session":"../opaque +? ü"}"# + r#"{"key":"key","session":"Li4vb3BhcXVlICs_IMO8"}"# ); Ok(()) } #[test] - fn session_token_exposes_and_recovers_its_inner_value() { - let token = SessionToken::from("opaque-token"); + fn session_token_round_trips_arbitrary_bytes() -> Result<(), serde_json::Error> { + let token = SessionToken::new([0, 1, 2, 0xfe, 0xff]); + let json = serde_json::to_string(&token)?; + assert_eq!(json, r#""AAEC_v8""#); + assert_eq!(serde_json::from_str::(&json)?, token); + Ok(()) + } - assert_eq!(token.as_ref(), "opaque-token"); - assert_eq!(token.to_string(), "opaque-token"); - assert_eq!(token.into_inner(), "opaque-token"); + #[test] + fn session_token_rejects_noncanonical_encodings() { + for invalid in ["%%%", "dG9rM24="] { + assert!( + SessionToken::from_base64url(invalid).is_err(), + "accepted {invalid:?}" + ); + } } #[test]