diff --git a/Cargo.lock b/Cargo.lock index 8dc41e619..026684c0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1446,11 +1446,14 @@ dependencies = [ "reqwest 0.13.4", "rkyv 0.8.18", "saturn", + "sea-orm", "serde", "serde_json", "sha1 0.11.0", + "sha2 0.11.0", "sysinfo 0.39.6", "tempfile", + "thiserror 2.0.20", "tokio", "tokio-stream", "tokio-util", @@ -4023,6 +4026,7 @@ dependencies = [ "reqwest 0.13.4", "serde", "serde_json", + "tempfile", "thiserror 2.0.20", "tokio", "tracing", @@ -4983,6 +4987,7 @@ dependencies = [ "futures", "git-internal", "http", + "io-orbit", "jemallocator", "jupiter", "jupiter-migrate", @@ -4997,6 +5002,7 @@ dependencies = [ "reqwest 0.13.4", "russh 0.63.1", "saturn", + "sea-orm", "serde", "serde_json", "serde_urlencoded", diff --git a/ceres/Cargo.toml b/ceres/Cargo.toml index 2fb5caaa4..b7a9c7209 100644 --- a/ceres/Cargo.toml +++ b/ceres/Cargo.toml @@ -29,6 +29,9 @@ rand = { workspace = true, features = ["thread_rng"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha1 = { workspace = true } +sha2 = { workspace = true, optional = true } +tempfile = { workspace = true, optional = true } +thiserror = { workspace = true, optional = true } tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } reqwest = { workspace = true, features = ["json", "stream"] } @@ -43,9 +46,11 @@ tokio-util = { workspace = true } rkyv = { workspace = true } [features] +fastcdc = ["dep:sha2", "dep:tempfile", "dep:thiserror"] migrate = ["jupiter/migrate"] [dev-dependencies] +sea-orm = { workspace = true } axum = { workspace = true } jupiter-migrate = { workspace = true } orion-client = { workspace = true } diff --git a/ceres/src/application/api_service/mono/lfs.rs b/ceres/src/application/api_service/mono/lfs.rs index 523492528..de825e443 100644 --- a/ceres/src/application/api_service/mono/lfs.rs +++ b/ceres/src/application/api_service/mono/lfs.rs @@ -14,6 +14,11 @@ use crate::lfs::{ }; impl LfsApplicationService { + #[cfg(feature = "fastcdc")] + pub fn media_service(&self) -> &jupiter::service::lfs_service::LfsService { + &self.ctx.storage().lfs_service + } + pub async fn lfs_retrieve_lock(&self, query: LockListQuery) -> Result { handler::lfs_retrieve_lock(self.ctx.storage().lfs_db_storage(), query).await } diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 2a79781ab..d9b445da4 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -180,8 +180,41 @@ pub async fn lfs_delete_lock( } } -/// -/// +/// The basic protocol accepts only SHA-256 object IDs, never storage paths. +/// Keep this check independent of optional transports so private storage keys +/// cannot be addressed through the standard LFS endpoints. +pub fn validate_object_oid(oid: &str) -> Result<(), GitLFSError> { + if oid.len() != 64 + || !oid + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(GitLFSError::GeneralError( + "Invalid LFS object ID: expected 64 lowercase hexadecimal characters".into(), + )); + } + Ok(()) +} + +fn validate_object_size(size: i64) -> Result<(), GitLFSError> { + if size < 0 { + return Err(GitLFSError::GeneralError( + "Invalid LFS object size: expected a non-negative size".into(), + )); + } + Ok(()) +} + +fn validate_request_object(object: &RequestObject) -> Result<(), GitLFSError> { + validate_object_oid(&object.oid)?; + validate_object_size(object.size) +} + +fn lfs_database_error(error: MegaError) -> GitLFSError { + tracing::error!("LFS metadata storage operation failed: {error}"); + GitLFSError::GeneralError("LFS metadata storage operation failed".into()) +} + /// Reference: /// 1. [Git LFS Batch API](https://github.com/git-lfs/git-lfs/blob/main/docs/api/batch.md) pub async fn lfs_process_batch( @@ -195,6 +228,16 @@ pub async fn lfs_process_batch( let file_storage = service.obj_storage.clone(); let db_storage = service.lfs_storage.clone(); for object in objects { + if let Err(error) = validate_request_object(&object) { + response_objects.push(ResponseObject::failed_with_err( + &object, + ObjectError { + code: 400, + message: error.to_string(), + }, + )); + continue; + } let meta_res = lfs_get_meta(&db_storage, &object.oid).await?; let meta = match meta_res { Some(meta) => meta, @@ -205,7 +248,7 @@ pub async fn lfs_process_batch( db_storage .new_lfs_object(meta.clone().into()) .await - .unwrap(); + .map_err(lfs_database_error)?; meta } else { response_objects.push(ResponseObject::failed_with_err( @@ -275,6 +318,7 @@ pub async fn lfs_upload_object( req_obj: &RequestObject, body_bytes: Vec, ) -> Result<(), GitLFSError> { + validate_request_object(req_obj)?; let db_storage: LfsDbStorage = service.lfs_storage.clone(); let meta = if let Some(meta) = lfs_get_meta(&db_storage, &req_obj.oid).await? { @@ -498,7 +542,16 @@ async fn lfs_get_meta( storage: &LfsDbStorage, oid: &str, ) -> Result, GitLFSError> { - Ok(storage.get_lfs_object(oid).await.unwrap().map(|m| m.into())) + validate_object_oid(oid)?; + let meta: Option = storage + .get_lfs_object(oid) + .await + .map_err(lfs_database_error)? + .map(Into::into); + if let Some(meta) = &meta { + validate_object_size(meta.size)?; + } + Ok(meta) } async fn lfs_delete_meta( @@ -640,9 +693,299 @@ async fn delete_lock( #[cfg(test)] mod tests { + use std::sync::Arc; + + use common::config::{LocalConfig, ObjectStorageBackend, ObjectStorageConfig}; + use io_orbit::factory::ObjectStorageFactory; + use jupiter::storage::base_storage::{BaseStorage, StorageConnector}; + use sea_orm::{ConnectionTrait, Database}; + use super::*; use crate::lfs::lfs_structs::{Action, Ref, ResCondition, ResponseObject}; + async fn lfs_fixture() -> (tempfile::TempDir, LfsService) { + let dir = tempfile::tempdir().unwrap(); + let db = Database::connect("sqlite::memory:").await.unwrap(); + db.execute_unprepared( + "CREATE TABLE lfs_objects (oid TEXT PRIMARY KEY, size BIGINT NOT NULL, exist BOOLEAN NOT NULL)", + ) + .await + .unwrap(); + let config = ObjectStorageConfig { + storage_type: ObjectStorageBackend::Local, + local: LocalConfig { + root_dir: dir.path().to_string_lossy().into_owned(), + }, + ..Default::default() + }; + let service = LfsService { + lfs_storage: LfsDbStorage { + base: BaseStorage::new(Arc::new(db)), + }, + obj_storage: ObjectStorageFactory::build(&config).await.unwrap(), + }; + (dir, service) + } + + fn batch_request(operation: Operation, objects: Vec) -> BatchRequest { + BatchRequest { + operation, + transfers: vec!["basic".into()], + objects, + hash_algo: "sha256".into(), + } + } + + #[tokio::test] + async fn basic_rejects_invalid_identifiers_and_sizes_before_database_access() { + let (_dir, service) = lfs_fixture().await; + let legacy_oid = "b".repeat(64); + service + .lfs_storage + .new_lfs_object(callisto::lfs_objects::Model { + oid: legacy_oid.clone(), + size: -1, + exist: true, + }) + .await + .unwrap(); + assert!(matches!( + lfs_download_object(service.clone(), legacy_oid.clone()).await, + Err(GitLFSError::GeneralError(message)) if message.starts_with("Invalid") + )); + assert!(matches!( + lfs_upload_object( + &service, + &RequestObject { oid: legacy_oid, ..Default::default() }, + vec![], + ).await, + Err(GitLFSError::GeneralError(message)) if message.starts_with("Invalid") + )); + // Validating these requests must not depend on a working metadata DB. + service + .lfs_storage + .get_connection() + .execute_unprepared("DROP TABLE lfs_objects") + .await + .unwrap(); + let mut requests: Vec = [ + String::new(), + "../secret".into(), + "media-v1/known-scope/chunks/known-hash".into(), + "media-v1%2Fknown-scope%2Fpending%2Fmanifest".into(), + "a".repeat(63), + "a".repeat(65), + "A".repeat(64), + "g".repeat(64), + ] + .into_iter() + .map(|oid| RequestObject { + oid, + size: 1, + ..Default::default() + }) + .collect(); + requests.push(RequestObject { + oid: "a".repeat(64), + size: -1, + ..Default::default() + }); + + for operation in [Operation::Upload, Operation::Download] { + let objects = requests + .iter() + .map(|request| RequestObject { + oid: request.oid.clone(), + size: request.size, + ..Default::default() + }) + .collect(); + let response = lfs_process_batch( + &service, + batch_request(operation, objects), + "http://localhost", + ) + .await + .unwrap(); + for object in response.objects { + assert_eq!(object.error.unwrap().code, 400); + assert!(object.actions.is_none()); + } + } + for request in requests { + assert!(matches!( + lfs_upload_object(&service, &request, vec![]).await, + Err(GitLFSError::GeneralError(message)) if message.starts_with("Invalid") + )); + if request.size >= 0 { + assert!(matches!( + lfs_download_object(service.clone(), request.oid).await, + Err(GitLFSError::GeneralError(message)) if message.starts_with("Invalid") + )); + } + } + } + + #[tokio::test] + async fn basic_cannot_read_or_overwrite_a_private_key_registered_in_legacy_metadata() { + let (_dir, service) = lfs_fixture().await; + let private_key = "media-v1/known-scope/chunks/known-hash"; + let key = lfs_object_key(private_key); + let original = b"private chunk".to_vec(); + service + .obj_storage + .inner + .put_stream(&key, original.clone().into_stream(), ObjectMeta::default()) + .await + .unwrap(); + service + .lfs_storage + .new_lfs_object(callisto::lfs_objects::Model { + oid: private_key.into(), + size: original.len() as i64, + exist: true, + }) + .await + .unwrap(); + + let request = RequestObject { + oid: private_key.into(), + size: original.len() as i64, + ..Default::default() + }; + assert!( + lfs_upload_object(&service, &request, b"replacement".to_vec()) + .await + .is_err() + ); + assert!( + lfs_download_object(service.clone(), private_key.into()) + .await + .is_err() + ); + let batch = lfs_process_batch( + &service, + batch_request(Operation::Download, vec![request]), + "http://localhost", + ) + .await + .unwrap(); + assert_eq!(batch.objects[0].error.as_ref().unwrap().code, 400); + assert!(batch.objects[0].actions.is_none()); + + let (mut stream, _) = service.obj_storage.inner.get_stream(&key).await.unwrap(); + let mut actual = Vec::new(); + while let Some(bytes) = stream.next().await { + actual.extend_from_slice(&bytes.unwrap()); + } + assert_eq!(actual, original); + } + + #[tokio::test] + async fn basic_metadata_failures_are_errors_instead_of_panics() { + let (_dir, service) = lfs_fixture().await; + service + .lfs_storage + .get_connection() + .execute_unprepared( + "CREATE TRIGGER reject_lfs_insert BEFORE INSERT ON lfs_objects BEGIN SELECT RAISE(FAIL, 'test insert failure'); END", + ) + .await + .unwrap(); + let request = || RequestObject { + oid: "a".repeat(64), + size: 0, + ..Default::default() + }; + let result = lfs_process_batch( + &service, + batch_request(Operation::Upload, vec![request()]), + "http://localhost", + ) + .await; + assert!(matches!( + result, + Err(GitLFSError::GeneralError(message)) if message == "LFS metadata storage operation failed" + )); + + service + .lfs_storage + .get_connection() + .execute_unprepared("DROP TABLE lfs_objects") + .await + .unwrap(); + assert!(matches!( + lfs_download_object(service.clone(), request().oid).await, + Err(GitLFSError::GeneralError(message)) if message == "LFS metadata storage operation failed" + )); + assert!(matches!( + lfs_upload_object(&service, &request(), vec![]).await, + Err(GitLFSError::GeneralError(message)) if message == "LFS metadata storage operation failed" + )); + } + + #[tokio::test] + async fn basic_valid_sha256_objects_keep_upload_and_download_actions() { + let (_dir, service) = lfs_fixture().await; + for (oid, data) in [ + ( + "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824", + b"hello".as_slice(), + ), + ( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + b"".as_slice(), + ), + ] { + let request = || RequestObject { + oid: oid.into(), + size: data.len() as i64, + ..Default::default() + }; + let upload = lfs_process_batch( + &service, + batch_request(Operation::Upload, vec![request()]), + "http://localhost", + ) + .await + .unwrap(); + assert!( + upload.objects[0] + .actions + .as_ref() + .unwrap() + .contains_key(&Action::Upload) + ); + lfs_upload_object(&service, &request(), data.to_vec()) + .await + .unwrap(); + let download = lfs_process_batch( + &service, + batch_request(Operation::Download, vec![request()]), + "http://localhost", + ) + .await + .unwrap(); + assert!( + download.objects[0] + .actions + .as_ref() + .unwrap() + .contains_key(&Action::Download) + ); + let mut stream = Box::pin( + lfs_download_object(service.clone(), oid.into()) + .await + .unwrap(), + ); + let mut actual = Vec::new(); + while let Some(bytes) = stream.next().await { + actual.extend_from_slice(&bytes.unwrap()); + } + assert_eq!(actual, data); + } + } + fn lock(id: &str) -> Lock { Lock { id: id.to_string(), diff --git a/ceres/src/lfs/media/chunker.rs b/ceres/src/lfs/media/chunker.rs new file mode 100644 index 000000000..5644a853a --- /dev/null +++ b/ceres/src/lfs/media/chunker.rs @@ -0,0 +1,186 @@ +//! Wire-compatible port of Libra's deterministic `fastcdc-v1` chunker +//! (libra main 92e1d64a, lore.md §6). External FastCDC crate variants do not +//! necessarily use these frozen parameters and must not replace it in place. +//! +//! No external crate: a gear-hash rolling fingerprint with normalized chunking +//! and hard min/max clamps. The parameters and the GEAR table are **FROZEN** for +//! v1 — changing any of them changes chunk boundaries and would be a breaking +//! `fastcdc-v2` (never an in-place edit), because §6.1 requires byte-identical +//! boundaries across versions/clones/clients. +//! +//! Determinism: for the frozen (GEAR, MIN, AVG, MAX, masks), boundaries are a +//! pure deterministic function of the input bytes — the same bytes always yield +//! the same `Vec`. The streaming [`chunk_reader`] and the in-memory +//! [`chunk_bytes`] share one code path, so they can never disagree. + +use std::io::{self, Read}; + +use super::sha256_hex; + +/// The frozen chunker algorithm identifier recorded in the manifest. +pub const ALGORITHM: &str = "fastcdc-v1"; + +/// Minimum chunk size: no boundary may fire before this many bytes (512 KiB). +pub const MIN_SIZE: usize = 512 * 1024; +/// Target average chunk size (2 MiB); `log2(AVG_SIZE) == 21` sets the mask width. +pub const AVG_SIZE: usize = 2 * 1024 * 1024; +/// Maximum chunk size: a boundary is forced here (8 MiB; matches the §6.4 +/// capability example `max_chunk_size = 8388608`). +pub const MAX_SIZE: usize = 8 * 1024 * 1024; + +// Normalized chunking (FastCDC): use a STRICTER mask before the average size +// (biases toward larger chunks, pushing the cut toward AVG) and a LOOSER mask +// after it (prevents runaway chunks). avg_bits = log2(AVG_SIZE) = 21. +// MASK_STRICT uses avg_bits + 2 = 23 high bits; MASK_LOOSE uses avg_bits - 2 = +// 19 high bits. Masking the well-mixed HIGH bits of the gear hash. +const MASK_STRICT: u64 = ((1u64 << 23) - 1) << 41; +const MASK_LOOSE: u64 = ((1u64 << 19) - 1) << 45; + +/// Deterministic 256-entry gear table, built at compile time from a fixed +/// splitmix64 sequence (frozen seed + constants) so it is fully reproducible +/// and carries no external data file. +const GEAR: [u64; 256] = build_gear(); + +const fn build_gear() -> [u64; 256] { + let mut table = [0u64; 256]; + let mut state: u64 = 0x9E37_79B9_7F4A_7C15; // frozen seed + let mut i = 0; + while i < 256 { + // splitmix64 + state = state.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = state; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^= z >> 31; + table[i] = z; + i += 1; + } + table +} + +/// One content-defined chunk: its byte range within the media object and the +/// lowercase-hex SHA-256 of its RAW (uncompressed) bytes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Chunk { + pub offset: u64, + pub length: u64, + pub chunk_hash: String, +} + +/// Chunk a byte stream. Returns the ordered chunk list. Degenerate inputs (a +/// FROZEN contract, pinned by tests): empty input → zero chunks; input shorter +/// than [`MIN_SIZE`] → exactly one chunk equal to the whole input (no boundary +/// can fire before MIN). +pub fn chunk_reader(mut reader: R) -> io::Result> { + let mut out = Vec::new(); + let mut buf = [0u8; 65536]; + // Bytes of the chunk currently being accumulated (bounded by MAX_SIZE). + let mut cur: Vec = Vec::with_capacity(MAX_SIZE.min(1 << 20)); + let mut offset: u64 = 0; + let mut fingerprint: u64 = 0; + loop { + let n = reader.read(&mut buf)?; + if n == 0 { + break; + } + for &byte in &buf[..n] { + cur.push(byte); + fingerprint = (fingerprint << 1).wrapping_add(GEAR[byte as usize]); + let len = cur.len(); + let cut = if len < MIN_SIZE { + false + } else if len < AVG_SIZE { + (fingerprint & MASK_STRICT) == 0 + } else if len < MAX_SIZE { + (fingerprint & MASK_LOOSE) == 0 + } else { + true // forced boundary at MAX_SIZE + }; + if cut { + out.push(Chunk { + offset, + length: len as u64, + chunk_hash: sha256_hex(&cur), + }); + offset += len as u64; + cur.clear(); + fingerprint = 0; + } + } + } + if !cur.is_empty() { + out.push(Chunk { + offset, + length: cur.len() as u64, + chunk_hash: sha256_hex(&cur), + }); + } + Ok(out) +} + +/// In-memory convenience over [`chunk_reader`] (infallible for a slice). +pub fn chunk_bytes(data: &[u8]) -> Vec { + // INVARIANT: reading from an in-memory `Cursor<&[u8]>` never returns an I/O + // error, so `chunk_reader` cannot fail here. + chunk_reader(io::Cursor::new(data)).expect("in-memory chunking cannot fail") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_input_yields_zero_chunks() { + assert!(chunk_bytes(&[]).is_empty()); + } + + #[test] + fn short_input_is_a_single_chunk() { + let data = vec![7u8; MIN_SIZE - 1]; + let chunks = chunk_bytes(&data); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0].offset, 0); + assert_eq!(chunks[0].length, (MIN_SIZE - 1) as u64); + assert_eq!(chunks[0].chunk_hash, sha256_hex(&data)); + } + + #[test] + fn deterministic_and_contiguous_and_covers_input() { + // A pseudo-random-but-fixed input larger than MAX so multiple chunks fire. + let mut data = Vec::with_capacity(MAX_SIZE * 3); + let mut x: u64 = 0x1234_5678_9ABC_DEF0; + while data.len() < MAX_SIZE * 3 { + x = x + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + data.push((x >> 33) as u8); + } + let a = chunk_bytes(&data); + let b = chunk_bytes(&data); + assert_eq!(a, b, "same bytes must produce byte-identical chunks"); + assert!(a.len() > 1, "a >MAX input must split into multiple chunks"); + // Contiguity + full coverage; each chunk within [.., MAX_SIZE]. + let mut expected_offset = 0u64; + for c in &a { + assert_eq!(c.offset, expected_offset, "chunks must be contiguous"); + assert!(c.length >= 1 && c.length <= MAX_SIZE as u64); + assert_eq!( + c.chunk_hash, + sha256_hex(&data[c.offset as usize..(c.offset + c.length) as usize]) + ); + expected_offset += c.length; + } + assert_eq!( + expected_offset as usize, + data.len(), + "chunks must cover input" + ); + } + + #[test] + fn streaming_and_in_memory_agree() { + let data = vec![0xABu8; MAX_SIZE + 123]; + let streamed = chunk_reader(io::Cursor::new(&data[..])).unwrap(); + assert_eq!(streamed, chunk_bytes(&data)); + } +} diff --git a/ceres/src/lfs/media/mod.rs b/ceres/src/lfs/media/mod.rs new file mode 100644 index 000000000..130d022d3 --- /dev/null +++ b/ceres/src/lfs/media/mod.rs @@ -0,0 +1,388 @@ +//! Opt-in FastCDC transport. Chunks are private to an authenticated user and +//! repository; only a finalized manifest permits reads. Full LFS objects remain +//! the interoperable source of truth. See docs/lfs-api.md for deployment limits. +use std::{ + collections::HashSet, + sync::LazyLock, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use bytes::Bytes; +use futures::StreamExt; +use io_orbit::object_storage::{ObjectByteStream, ObjectKey, ObjectMeta, ObjectNamespace}; +use jupiter::{service::lfs_service::LfsService, utils::into_obj_stream::IntoObjectStream}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tokio::{ + io::{AsyncSeekExt, AsyncWriteExt}, + sync::Semaphore, +}; + +pub mod chunker; +pub mod protocol; +use protocol::{ManifestResponse, MediaManifest, PrepareResponse, valid_hash}; + +#[cfg(test)] +mod tests; + +const PENDING_TTL: Duration = Duration::from_secs(24 * 3600); +static FINALIZERS: LazyLock = LazyLock::new(|| Semaphore::new(2)); + +#[derive(Debug, thiserror::Error)] +pub enum MediaError { + #[error("Invalid media request: {0}")] + Invalid(String), + #[error("Media object not found")] + NotFound, + #[error("Media manifest conflicts with the finalized object")] + Conflict, + #[error("Media storage failed: {0}")] + Storage(String), + #[error("Media I/O failed: {0}")] + Io(#[from] std::io::Error), + #[error("Media JSON failed: {0}")] + Json(#[from] serde_json::Error), +} + +pub fn sha256_hex(bytes: &[u8]) -> String { + hex::encode(Sha256::digest(bytes)) +} + +/// Construct only after authentication. Identity is taken from the validated +/// access token, never from request JSON. Repository is the original LFS URI. +#[derive(Debug, Clone)] +pub struct MediaScope(String); + +impl MediaScope { + pub fn new(actor: &str, repo: &str) -> Result { + if actor.is_empty() + || repo.is_empty() + || !repo.starts_with('/') + || repo.contains('\\') + || repo.contains('%') + || repo.contains('?') + || repo + .split('/') + .skip(1) + .any(|p| p.is_empty() || p == "." || p == "..") + { + return Err(MediaError::Invalid( + "a canonical repository and authenticated actor are required".into(), + )); + } + Ok(Self(sha256_hex(&serde_json::to_vec(&(actor, repo))?))) + } + + fn key(&self, suffix: &str) -> ObjectKey { + ObjectKey { + namespace: ObjectNamespace::Media, + key: format!("media-v1/{}/{suffix}", self.0), + } + } +} + +#[derive(Serialize, Deserialize)] +struct Pending { + created_at: u64, + manifest: MediaManifest, +} + +fn now() -> Result { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .map_err(|e| MediaError::Storage(format!("system clock: {e}"))) +} + +fn storage_error(e: impl std::fmt::Display) -> MediaError { + MediaError::Storage(e.to_string()) +} + +async fn read_bounded( + service: &LfsService, + key: &ObjectKey, + limit: usize, +) -> Result, MediaError> { + if !service + .obj_storage + .inner + .exists(key) + .await + .map_err(storage_error)? + { + return Err(MediaError::NotFound); + } + let (mut stream, _) = service + .obj_storage + .inner + .get_stream(key) + .await + .map_err(storage_error)?; + let mut data = Vec::new(); + while let Some(bytes) = stream.next().await { + let bytes = bytes?; + if bytes.len() > limit.saturating_sub(data.len()) { + return Err(MediaError::Invalid( + "stored media payload exceeds its size limit".into(), + )); + } + data.extend_from_slice(&bytes); + } + Ok(data) +} + +async fn put(service: &LfsService, key: &ObjectKey, data: Vec) -> Result<(), MediaError> { + let size = data.len() as i64; + service + .obj_storage + .inner + .put_stream( + key, + data.into_stream(), + ObjectMeta { + size, + ..Default::default() + }, + ) + .await + .map_err(storage_error) +} + +async fn pending( + service: &LfsService, + scope: &MediaScope, + id: &str, +) -> Result { + if !valid_hash(id) { + return Err(MediaError::NotFound); + } + let data = read_bounded( + service, + &scope.key(&format!("pending/{id}")), + protocol::MAX_MANIFEST_SIZE + 1024, + ) + .await?; + let entry: Pending = serde_json::from_slice(&data)?; + if now()?.saturating_sub(entry.created_at) > PENDING_TTL.as_secs() { + return Err(MediaError::NotFound); + } + if entry.manifest.id()? != id { + return Err(MediaError::Conflict); + } + Ok(entry.manifest) +} + +pub async fn prepare( + service: &LfsService, + scope: &MediaScope, + mut manifest: MediaManifest, +) -> Result { + manifest.validate()?; + manifest.fallback_oid = Some(manifest.media_oid.clone()); + let id = manifest.id()?; + let mut missing = Vec::new(); + let mut seen = HashSet::new(); + for chunk in &manifest.chunks { + if seen.insert(&chunk.chunk_hash) { + match read_chunk(service, scope, &chunk.chunk_hash, chunk.length).await { + Ok(_) => (), + Err(MediaError::NotFound | MediaError::Invalid(_)) => { + missing.push(chunk.chunk_hash.clone()) + } + Err(e) => return Err(e), + } + } + } + let data = serde_json::to_vec(&Pending { + created_at: now()?, + manifest, + })?; + if data.len() > protocol::MAX_MANIFEST_SIZE { + return Err(MediaError::Invalid("manifest too large".into())); + } + put(service, &scope.key(&format!("pending/{id}")), data).await?; + Ok(PrepareResponse { + manifest_id: id, + missing_chunks: missing, + }) +} + +async fn read_chunk( + service: &LfsService, + scope: &MediaScope, + hash: &str, + length: u64, +) -> Result, MediaError> { + let bytes = read_bounded( + service, + &scope.key(&format!("chunks/{hash}")), + chunker::MAX_SIZE, + ) + .await?; + if bytes.len() as u64 != length || sha256_hex(&bytes) != hash { + return Err(MediaError::Invalid("chunk size or SHA-256 mismatch".into())); + } + Ok(bytes) +} + +pub async fn upload_chunk( + service: &LfsService, + scope: &MediaScope, + id: &str, + hash: &str, + data: Vec, +) -> Result<(), MediaError> { + let manifest = pending(service, scope, id).await?; + let chunk = manifest + .chunks + .iter() + .find(|c| c.chunk_hash == hash) + .ok_or(MediaError::NotFound)?; + if data.len() as u64 != chunk.length || sha256_hex(&data) != hash { + return Err(MediaError::Invalid("chunk size or SHA-256 mismatch".into())); + } + put(service, &scope.key(&format!("chunks/{hash}")), data).await +} + +pub async fn get_manifest( + service: &LfsService, + scope: &MediaScope, + oid: &str, +) -> Result { + if !valid_hash(oid) { + return Err(MediaError::NotFound); + } + let bytes = read_bounded( + service, + &scope.key(&format!("finalized/{oid}")), + protocol::MAX_MANIFEST_SIZE, + ) + .await?; + let manifest: MediaManifest = serde_json::from_slice(&bytes)?; + if manifest.media_oid != oid { + return Err(MediaError::Conflict); + } + Ok(ManifestResponse { + manifest_id: manifest.id()?, + manifest, + }) +} + +pub async fn download_chunk( + service: &LfsService, + scope: &MediaScope, + oid: &str, + hash: &str, +) -> Result { + let response = get_manifest(service, scope, oid).await?; + let chunk = response + .manifest + .chunks + .iter() + .find(|c| c.chunk_hash == hash) + .ok_or(MediaError::NotFound)?; + Ok(Bytes::from( + read_chunk(service, scope, hash, chunk.length).await?, + )) +} + +/// Reconstruct into a temporary file, verify every chunk, full SHA-256 and the +/// frozen CDC boundaries, persist the standard fallback, then publish metadata. +/// Publication is an atomic single-object PUT. Canonical content IDs and frozen +/// chunking make concurrent valid finalizations identical (no last-writer loss). +pub async fn finalize( + service: &LfsService, + scope: &MediaScope, + id: &str, +) -> Result<(), MediaError> { + let _permit = FINALIZERS.acquire().await.map_err(storage_error)?; + let manifest = pending(service, scope, id).await?; + if let Some(meta) = service + .lfs_storage + .get_lfs_object(&manifest.media_oid) + .await + .map_err(storage_error)? + && meta.size != manifest.media_size as i64 + { + return Err(MediaError::Conflict); + } + match get_manifest(service, scope, &manifest.media_oid).await { + Ok(existing) if existing.manifest_id != id => return Err(MediaError::Conflict), + Ok(_) | Err(MediaError::NotFound) => (), + Err(e) => return Err(e), + } + let temp = tempfile::tempfile()?; + let mut file = tokio::fs::File::from_std(temp); + let mut digest = Sha256::new(); + for chunk in &manifest.chunks { + let bytes = read_chunk(service, scope, &chunk.chunk_hash, chunk.length).await?; + digest.update(&bytes); + file.write_all(&bytes).await?; + } + if hex::encode(digest.finalize()) != manifest.media_oid { + return Err(MediaError::Invalid("full media SHA-256 mismatch".into())); + } + file.flush().await?; + file.rewind().await?; + let mut file = file.into_std().await; + let (mut file, chunks) = tokio::task::spawn_blocking(move || { + let chunks = chunker::chunk_reader(&mut file)?; + Ok::<_, std::io::Error>((file, chunks)) + }) + .await + .map_err(storage_error)??; + if chunks.len() != manifest.chunks.len() + || chunks.iter().zip(&manifest.chunks).any(|(a, b)| { + a.offset != b.offset || a.length != b.length || a.chunk_hash != b.chunk_hash + }) + { + return Err(MediaError::Invalid( + "manifest does not use frozen fastcdc-v1 boundaries".into(), + )); + } + std::io::Seek::rewind(&mut file)?; + let stream: ObjectByteStream = Box::pin(tokio_util::io::ReaderStream::new( + tokio::fs::File::from_std(file), + )); + let key = ObjectKey { + namespace: ObjectNamespace::Lfs, + key: manifest.media_oid.clone(), + }; + service + .obj_storage + .inner + .put_stream_bounded( + &key, + stream, + ObjectMeta { + size: manifest.media_size as i64, + ..Default::default() + }, + ) + .await + .map_err(storage_error)?; + service + .lfs_storage + .new_lfs_object(callisto::lfs_objects::Model { + oid: manifest.media_oid.clone(), + size: manifest.media_size as i64, + exist: true, + }) + .await + .map_err(storage_error)?; + let meta = service + .lfs_storage + .get_lfs_object(&manifest.media_oid) + .await + .map_err(storage_error)? + .ok_or_else(|| MediaError::Storage("LFS fallback metadata was not persisted".into()))?; + if meta.size != manifest.media_size as i64 { + return Err(MediaError::Conflict); + } + put( + service, + &scope.key(&format!("finalized/{}", manifest.media_oid)), + serde_json::to_vec(&manifest)?, + ) + .await +} diff --git a/ceres/src/lfs/media/protocol.rs b/ceres/src/lfs/media/protocol.rs new file mode 100644 index 000000000..d71143dc5 --- /dev/null +++ b/ceres/src/lfs/media/protocol.rs @@ -0,0 +1,126 @@ +//! Wire contract shared with Libra's feature-gated `utils::media` implementation. +use serde::{Deserialize, Serialize}; + +use super::{MediaError, chunker, sha256_hex}; + +pub const MAX_MANIFEST_SIZE: usize = 10 * 1024 * 1024; +pub const MAX_CHUNKS: usize = 8192; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ChunkEntry { + pub offset: u64, + pub length: u64, + pub chunk_hash: String, + pub encoded_length: u64, + pub compression: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub checksum: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CreatedBy { + pub client: String, + pub version: String, + pub capabilities: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MediaManifest { + pub version: u32, + pub algorithm: String, + pub hash_algorithm: String, + pub media_oid: String, + pub media_size: u64, + pub chunks: Vec, + pub created_by: CreatedBy, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fallback_oid: Option, +} + +pub fn valid_hash(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +impl MediaManifest { + pub fn validate(&self) -> Result<(), MediaError> { + if self.version != 1 + || self.algorithm != chunker::ALGORITHM + || self.hash_algorithm != "sha256" + || !valid_hash(&self.media_oid) + || self.chunks.len() > MAX_CHUNKS + || self + .fallback_oid + .as_ref() + .is_some_and(|oid| oid != &self.media_oid) + { + return Err(MediaError::Invalid( + "unsupported or invalid media manifest".into(), + )); + } + let mut offset = 0u64; + for chunk in &self.chunks { + if chunk.offset != offset + || chunk.length == 0 + || chunk.length > chunker::MAX_SIZE as u64 + || !valid_hash(&chunk.chunk_hash) + || chunk.compression != "none" + || chunk.encoded_length != chunk.length + || chunk.checksum.is_some() + { + return Err(MediaError::Invalid( + "invalid chunk range, hash or encoding".into(), + )); + } + offset = offset + .checked_add(chunk.length) + .ok_or_else(|| MediaError::Invalid("chunk offset overflow".into()))?; + } + if offset != self.media_size { + return Err(MediaError::Invalid( + "chunk lengths do not match media_size".into(), + )); + } + Ok(()) + } + + /// Content identity excludes client provenance. A verified frozen chunking + /// of a given media OID has exactly one ID, including across client versions. + pub fn id(&self) -> Result { + self.validate()?; + let bytes = serde_json::to_vec(&( + self.version, + &self.algorithm, + &self.hash_algorithm, + &self.media_oid, + self.media_size, + &self.chunks, + ))?; + Ok(sha256_hex(&bytes)) + } +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct PrepareResponse { + pub manifest_id: String, + pub missing_chunks: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ManifestResponse { + pub manifest_id: String, + pub manifest: MediaManifest, +} + +pub fn capabilities() -> serde_json::Value { + serde_json::json!({ + "version": "1", "chunked_lfs": true, + "chunk_algorithms": [chunker::ALGORITHM], "hash_algorithms": ["sha256"], + "max_chunk_size": chunker::MAX_SIZE, "max_manifest_size": MAX_MANIFEST_SIZE, + "supports_batch_exists": true, "supports_range_read": false, + "supports_standard_lfs_fallback": true, + "scope": "authenticated-user-and-repository" + }) +} diff --git a/ceres/src/lfs/media/tests.rs b/ceres/src/lfs/media/tests.rs new file mode 100644 index 000000000..d8bdadca6 --- /dev/null +++ b/ceres/src/lfs/media/tests.rs @@ -0,0 +1,304 @@ +use std::sync::Arc; + +use common::config::{LocalConfig, ObjectStorageBackend, ObjectStorageConfig}; +use io_orbit::factory::ObjectStorageFactory; +use jupiter::storage::{ + base_storage::{BaseStorage, StorageConnector}, + lfs_db_storage::LfsDbStorage, +}; +use sea_orm::{ConnectionTrait, Database}; + +use super::{ + protocol::{ChunkEntry, CreatedBy}, + *, +}; + +async fn fixture() -> (tempfile::TempDir, LfsService, MediaScope) { + let dir = tempfile::tempdir().unwrap(); + let db = Database::connect("sqlite::memory:").await.unwrap(); + db.execute_unprepared("CREATE TABLE lfs_objects (oid TEXT PRIMARY KEY, size BIGINT NOT NULL, exist BOOLEAN NOT NULL)").await.unwrap(); + let config = ObjectStorageConfig { + storage_type: ObjectStorageBackend::Local, + local: LocalConfig { + root_dir: dir.path().to_string_lossy().into_owned(), + }, + ..Default::default() + }; + let service = LfsService { + lfs_storage: LfsDbStorage { + base: BaseStorage::new(Arc::new(db)), + }, + obj_storage: ObjectStorageFactory::build(&config).await.unwrap(), + }; + ( + dir, + service, + MediaScope::new("alice", "/project/demo.git").unwrap(), + ) +} + +fn sample(data: &[u8]) -> MediaManifest { + MediaManifest { + version: 1, + algorithm: chunker::ALGORITHM.into(), + hash_algorithm: "sha256".into(), + media_oid: sha256_hex(data), + media_size: data.len() as u64, + chunks: chunker::chunk_bytes(data) + .into_iter() + .map(|c| ChunkEntry { + offset: c.offset, + length: c.length, + chunk_hash: c.chunk_hash, + encoded_length: c.length, + compression: "none".into(), + checksum: None, + }) + .collect(), + created_by: CreatedBy { + client: "libra".into(), + version: "fixture".into(), + capabilities: vec![chunker::ALGORITHM.into()], + }, + fallback_oid: None, + } +} + +async fn upload_all( + service: &LfsService, + scope: &MediaScope, + manifest: &MediaManifest, + data: &[u8], +) -> String { + let prepared = prepare(service, scope, manifest.clone()).await.unwrap(); + for hash in prepared.missing_chunks { + let c = manifest + .chunks + .iter() + .find(|c| c.chunk_hash == hash) + .unwrap(); + upload_chunk( + service, + scope, + &prepared.manifest_id, + &hash, + data[c.offset as usize..(c.offset + c.length) as usize].to_vec(), + ) + .await + .unwrap(); + } + prepared.manifest_id +} + +#[tokio::test] +async fn roundtrip_resume_dedup_isolation_and_standard_fallback() { + let (_dir, service, scope) = fixture().await; + let mut seed = 0x1234_5678_9abc_def0u64; + let data: Vec = (0..12 * 1024 * 1024) + .map(|_| { + seed = seed + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + (seed >> 33) as u8 + }) + .collect(); + let manifest = sample(&data); + assert!(manifest.chunks.len() > 1); + assert!( + manifest + .chunks + .windows(2) + .any(|c| c[0].length != c[1].length) + ); + let first = prepare(&service, &scope, manifest.clone()).await.unwrap(); + assert!(matches!( + get_manifest(&service, &scope, &manifest.media_oid).await, + Err(MediaError::NotFound) + )); + assert!( + finalize(&service, &scope, &first.manifest_id) + .await + .is_err() + ); + let chunk = &manifest.chunks[0]; + assert!( + upload_chunk( + &service, + &scope, + &first.manifest_id, + &chunk.chunk_hash, + vec![0; chunk.length as usize] + ) + .await + .is_err() + ); + upload_chunk( + &service, + &scope, + &first.manifest_id, + &chunk.chunk_hash, + data[..chunk.length as usize].to_vec(), + ) + .await + .unwrap(); + let resumed = prepare(&service, &scope, manifest.clone()).await.unwrap(); + assert_eq!(resumed.manifest_id, first.manifest_id); + assert!(!resumed.missing_chunks.contains(&chunk.chunk_hash)); + assert!(matches!( + download_chunk(&service, &scope, &manifest.media_oid, &chunk.chunk_hash).await, + Err(MediaError::NotFound) + )); + let id = upload_all(&service, &scope, &manifest, &data).await; + finalize(&service, &scope, &id).await.unwrap(); + finalize(&service, &scope, &id).await.unwrap(); + assert!( + prepare(&service, &scope, manifest.clone()) + .await + .unwrap() + .missing_chunks + .is_empty() + ); + assert_eq!( + get_manifest(&service, &scope, &manifest.media_oid) + .await + .unwrap() + .manifest_id, + id + ); + let stream = + crate::lfs::handler::lfs_download_object(service.clone(), manifest.media_oid.clone()) + .await + .unwrap(); + let mut stream = Box::pin(stream); + let mut whole = Vec::new(); + while let Some(bytes) = stream.next().await { + whole.extend_from_slice(&bytes.unwrap()); + } + assert_eq!(whole, data); + for other in [ + MediaScope::new("bob", "/project/demo.git").unwrap(), + MediaScope::new("alice", "/project/other.git").unwrap(), + ] { + assert!(matches!( + get_manifest(&service, &other, &manifest.media_oid).await, + Err(MediaError::NotFound) + )); + assert!(matches!( + upload_chunk(&service, &other, &id, &chunk.chunk_hash, vec![]).await, + Err(MediaError::NotFound) + )); + assert_eq!( + prepare(&service, &other, manifest.clone()) + .await + .unwrap() + .missing_chunks + .len(), + manifest.chunks.len() + ); + } + // A new file revision reuses unchanged content-defined chunks, not merely + // a repeated upload of the exact same object. + let mut edited = data.clone(); + edited[0] ^= 1; + let changed = sample(&edited); + let delta = prepare(&service, &scope, changed.clone()).await.unwrap(); + assert_ne!(changed.media_oid, manifest.media_oid); + assert_eq!( + delta.missing_chunks, + vec![changed.chunks[0].chunk_hash.clone()] + ); + let changed_id = upload_all(&service, &scope, &changed, &edited).await; + finalize(&service, &scope, &changed_id).await.unwrap(); + assert_eq!( + get_manifest(&service, &scope, &manifest.media_oid) + .await + .unwrap() + .manifest_id, + id + ); +} + +#[tokio::test] +async fn rejects_wrong_media_hash_and_noncanonical_chunking_without_publication() { + let (_dir, service, scope) = fixture().await; + let data = b"hello world"; + let mut manifest = sample(data); + manifest.media_oid = "a".repeat(64); + let id = upload_all(&service, &scope, &manifest, data).await; + assert!(finalize(&service, &scope, &id).await.is_err()); + assert!(matches!( + get_manifest(&service, &scope, &manifest.media_oid).await, + Err(MediaError::NotFound) + )); + let mut manifest = sample(data); + manifest.chunks = [(&data[..5], 0), (&data[5..], 5)] + .into_iter() + .map(|(bytes, offset)| ChunkEntry { + offset, + length: bytes.len() as u64, + encoded_length: bytes.len() as u64, + chunk_hash: sha256_hex(bytes), + compression: "none".into(), + checksum: None, + }) + .collect(); + let id = upload_all(&service, &scope, &manifest, data).await; + assert!(finalize(&service, &scope, &id).await.is_err()); + assert!(matches!( + get_manifest(&service, &scope, &manifest.media_oid).await, + Err(MediaError::NotFound) + )); +} + +#[tokio::test] +async fn empty_object_roundtrip_and_expired_pending_rejected() { + let (_dir, service, scope) = fixture().await; + let manifest = sample(&[]); + let id = upload_all(&service, &scope, &manifest, &[]).await; + finalize(&service, &scope, &id).await.unwrap(); + assert!( + get_manifest(&service, &scope, &manifest.media_oid) + .await + .unwrap() + .manifest + .chunks + .is_empty() + ); + let expired = Pending { + created_at: 0, + manifest, + }; + put( + &service, + &scope.key(&format!("pending/{id}")), + serde_json::to_vec(&expired).unwrap(), + ) + .await + .unwrap(); + assert!(matches!( + finalize(&service, &scope, &id).await, + Err(MediaError::NotFound) + )); +} + +#[test] +fn rejects_malformed_manifests_and_scope_paths() { + for repo in ["", "relative", "/../secret", "/a//b", "/%2fsecret", "/a\\b"] { + assert!(MediaScope::new("alice", repo).is_err()); + } + assert!(MediaScope::new("", "/repo").is_err()); + let original = sample(b"hello"); + for change in 0..7 { + let mut m = original.clone(); + match change { + 0 => m.chunks[0].length = 0, + 1 => m.chunks[0].length = u64::MAX, + 2 => m.chunks[0].offset = 1, + 3 => m.chunks[0].encoded_length = 0, + 4 => m.chunks[0].chunk_hash = "../secret".into(), + 5 => m.fallback_oid = Some("b".repeat(64)), + _ => m.chunks[0].compression = "gzip".into(), + } + assert!(m.validate().is_err()); + } +} diff --git a/ceres/src/lfs/mod.rs b/ceres/src/lfs/mod.rs index ce9bd7eff..ecb479e99 100644 --- a/ceres/src/lfs/mod.rs +++ b/ceres/src/lfs/mod.rs @@ -1,2 +1,4 @@ pub mod handler; pub mod lfs_structs; +#[cfg(feature = "fastcdc")] +pub mod media; diff --git a/docs/lfs-api.md b/docs/lfs-api.md index 3fafde899..b59f39a08 100644 --- a/docs/lfs-api.md +++ b/docs/lfs-api.md @@ -42,7 +42,153 @@ Relative to either base path above: | `POST` | `/locks/verify` | Verify locks before push | | `POST` | `/locks/{id}/unlock` | Delete lock | -Chunk download endpoints (`/objects/{oid}/chunks/...`) are **not** exposed in the current router. +Chunk download endpoints (`/objects/{oid}/chunks/...`) are **not** exposed in the current router. + +## Optional FastCDC transport with Libra + +Build both programs with `--features fastcdc` (Mega: `cargo build -p mono +--features fastcdc`; Libra: `cargo build --features fastcdc`). Both features are +off by default. Libra's normal LFS upload/download paths then probe the extension; +`libra config lfs.fastcdc false` disables its use for that repository. Standard +batch requests, LFS pointers and full object URLs retain the `basic` protocol. + +The extension uses Libra's **frozen in-tree `fastcdc-v1`** algorithm, not the +third-party `fastcdc::v2020` algorithm: 512 KiB minimum, 2 MiB target, 8 MiB maximum, +fixed SplitMix64 gear table, SHA-256 over raw chunks and the complete file. The +server re-chunks the reconstructed file before finalization to verify boundaries. +See `ceres/src/lfs/media/chunker.rs`, ported from Libra commit `92e1d64a`. + +The base is **`.git/info/lfs/libra/media/v1`**, for example +`/project/demo.git/info/lfs/libra/media/v1`. The URI rewrite retains the original +repository path in a request extension. Scope-less `/api/v1/lfs` and `/info/lfs` +requests cannot access media data. Canonical repository paths are required; +encoded path aliases and dot segments are rejected. + +The content-type table above describes standard LFS endpoints. The media +extension uses `application/json` for manifest uploads and successful JSON +responses, and `application/octet-stream` for raw chunk bodies. + +All extension endpoints require `Authorization: Bearer `. +Libra uses its host-scoped stored token, also for capability discovery. Missing +or invalid tokens make discovery fall back to standard LFS. No credentials or +storage object keys are returned in error bodies. + +Use a **Mono-issued** token from the server's existing authenticated +`POST /api/v1/user/token/generate` flow. `libra auth login` only saves that token +locally; it neither issues a Mega token nor converts a GitHub PAT or browser +session cookie into one. With Mega listening on localhost port 8000, run the +feature-built Libra binary in a Libra repository: + +```bash +libra config remote.origin.url http://localhost:8000/project/demo.git +libra auth login --host http://localhost:8000 +# Paste the Mono access token at the hidden prompt. +libra auth status --host http://localhost:8000 +libra config lfs.fastcdc true +libra media probe --remote origin +``` + +The token scope must match the remote's **host and port**. Non-loopback remotes +require HTTPS; HTTP token attachment is allowed only for loopback. `--host` takes +an origin without the repository path. In scripts, supply the token on stdin +using `--with-token`, never as an argv value or inside the remote URL. Compiling +the feature does not replace an independently installed `libra` on PATH. +An unset `lfs.fastcdc` permits negotiation in a feature-enabled build; `true` +explicitly enables it and `false` disables it for that repository. `media probe` +checks remote capabilities, not this repository setting, so a `chunked` result +alone does not confirm that normal LFS transfers will use the extension. + +| Method | Relative path | Contract | +|--------|---------------|----------| +| GET | `/capabilities` | Version, frozen algorithm, size limits and fallback support | +| POST | `/manifests` | Validate manifest, persist Pending descriptor, return `manifest_id` and `missing_chunks` (batch existence query) | +| PUT | `/manifests/{id}/chunks/{hash}` | Upload one referenced raw chunk; verify size and SHA-256 | +| POST | `/manifests/{id}/finalize` | Verify chunks, whole hash and CDC boundaries; store full LFS fallback; publish Finalized manifest | +| GET | `/manifests/by-media/{oid}` | Return only a Finalized manifest and its `manifest_id` | +| GET | `/manifests/by-media/{oid}/chunks/{hash}` | Read a chunk referenced by that Finalized object, with integrity verification | + +The manifest JSON matches Libra `MediaManifest`: `version`, `algorithm`, +`hash_algorithm`, `media_oid`, `media_size`, `chunks`, `created_by`, `fallback_oid`. +Each chunk has `offset`, `length`, `chunk_hash`, `encoded_length`, `compression`; +v1 only allows `compression: "none"` and omits the reserved `checksum` field. +Chunks must cover the file without gaps/overlaps, have positive bounded lengths, +and use lowercase SHA-256 hashes. Empty files have zero chunks. Maximum manifest +body: 10 MiB; maximum chunk count: 8192. Chunk request bodies are bounded at 8 MiB. + +`manifest_id` is SHA-256 of the compact JSON array +`[version,algorithm,hash_algorithm,media_oid,media_size,chunks]`. Client provenance +does not affect identity. Frozen boundary validation means two valid manifests +for the same content have the same ID. Finalized publication is an atomic object +store PUT, after the complete fallback and database metadata are persisted; a +crash before publication leaves no readable manifest. Repeating prepare/upload/ +finalize is safe. An interrupted client repeats prepare and uploads only missing +chunks. Pending descriptors expire after 24 hours; repeat prepare to resume later. + +Libra caches verified chunks outside the Git object database. Downloads reuse +intact cached chunks, repair corrupted cached chunks, and replace the destination +atomically only after the full SHA-256 verifies. A missing Finalized manifest +uses the complete LFS object instead; authentication/integrity failures after +selecting a manifest are errors, not silent fallback. + +### Security and operational boundary + +This is an **opt-in transport**, not completion of the entire Lore §6 server plan. +Mega's existing LFS API has no complete repository ACL implementation, so media +storage is conservatively isolated by **authenticated user + repository**. Users +cannot enumerate or download each other's chunks, even when hashes are known. +Every chunk operation also requires a Pending manifest ID or Finalized media OID; +there is no bare global chunk-hash endpoint. Another user's download uses the +standard full-object fallback. The existing standard LFS access policy is unchanged. + +Private backend keys are in the dedicated `media` namespace under +`media-v1//`, with separate +`pending`, `chunks`, and `finalized` prefixes. Full fallback objects keep their +existing OID keys. Back up both LFS objects and media metadata together. There is +currently **no automatic orphan GC, quota accounting, shared repository ACL, +obliteration, or byte-range hydration**. Expired descriptors and orphan chunks +remain on disk; do not turn on broad production access without retention/quota +policy. Never configure a blanket chunk expiry rule: finalized objects can share +chunks within their scope. Finalization limits concurrent full-file staging to two +operations per server process and uses temporary files. Complete fallback objects +use a bounded multipart writer (8 MiB parts, one in flight); empty objects use an +empty single PUT. Local and cloud storage publish only completed writes. An +S3-compatible backend must support multipart upload for this extension; failures +are returned without falling back to whole-file buffering. Configure the backend's +incomplete-multipart lifecycle cleanup for process crashes. + +### Tests and cross-repository verification + +```text +# Mega: storage, LFS service and actual HTTP router/authentication tests +cargo test -p io-orbit --lib +cargo test -p ceres --features fastcdc --lib lfs:: +cargo test -p mono --features fastcdc --lib api::router::lfs_router:: + +# Libra: chunker, manifest, cache, fallback and corruption tests +cargo test --features fastcdc --lib utils::media:: +cargo test --features fastcdc --test media_fastcdc_test +``` + +For the two-process interop test, set `MEGA_FASTCDC_READY_FILE` to the **same +absolute temporary file path** in two terminals. Start Mega's isolated server: + +```text +cargo test -p mono --features fastcdc --lib serve_libra_interop -- --ignored --nocapture +``` + +After the ready file appears, run in Libra: + +```text +cargo test --features fastcdc --test media_fastcdc_test mega_fastcdc_http_interop -- --ignored --nocapture +``` + +The fixture uses the production media routes, basic LFS service handlers, +URI rewrite and access-token lookup, +with private temporary storage/SQLite and fixed **test-only** tokens. It listens +only on loopback and stops after ten minutes or `POST /__test/stop`. The test covers +Libra's normal LFS batch/upload/download entry points, interrupted upload replay, +dedup, cached-chunk resume/repair, cross-user isolation and full-object fallback, +empty files, and recovery after fallback persistence but before manifest publication. ## Examples diff --git a/io-orbit/Cargo.toml b/io-orbit/Cargo.toml index a2c843b71..6339527ff 100644 --- a/io-orbit/Cargo.toml +++ b/io-orbit/Cargo.toml @@ -22,3 +22,6 @@ reqwest = { workspace = true } tokio = { workspace = true, features = ["time", "macros", "rt"] } serde = { workspace = true } serde_json = { workspace = true } + +[dev-dependencies] +tempfile = { workspace = true } diff --git a/io-orbit/src/adapter.rs b/io-orbit/src/adapter.rs index 0732e5a0e..0a9e8f9be 100644 --- a/io-orbit/src/adapter.rs +++ b/io-orbit/src/adapter.rs @@ -45,6 +45,9 @@ pub struct ObjectStoreAdapter { pub upload_strategy: UploadStrategy, } +// S3/GCS require non-final parts >= 5 MiB; R2 also requires equal part sizes. +const MULTIPART_PART_SIZE: usize = 8 * 1024 * 1024; + /// Supported backend implementations for object storage. /// /// Each variant wraps a specific `object_store` backend in an [`Arc`] so that @@ -88,6 +91,17 @@ impl MegaObjectStorage for ObjectStoreAdapter { } } + async fn put_stream_bounded( + &self, + key: &ObjectKey, + data: ObjectByteStream, + _meta: ObjectMeta, + ) -> Result<(), MegaError> { + // Local, GCS and S3 implement atomic multipart publication. Do not use + // the configured SinglePut policy, which buffers the entire object. + self.put_multipart(&key.to_object_store_path(), data).await + } + async fn get_stream( &self, key: &ObjectKey, @@ -171,7 +185,7 @@ impl MegaObjectStorage for ObjectStoreAdapter { async fn exists(&self, key: &ObjectKey) -> Result { let path = key.to_object_store_path(); - Ok(self.to_store().head(&path).await.is_ok()) + Self::exists_in(self.to_store(), &path).await } async fn delete(&self, key: &ObjectKey) -> Result<(), MegaError> { @@ -526,6 +540,17 @@ impl LogStorage for ObjectStoreAdapter { } impl ObjectStoreAdapter { + async fn exists_in( + store: &dyn ObjectStore, + path: &object_store::path::Path, + ) -> Result { + match store.head(path).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(error) => Err(IoOrbitError::from(error).into()), + } + } + fn to_store(&self) -> &dyn ObjectStore { let store: &dyn ObjectStore = match &self.store { BackendStore::S3(s3) => s3.as_ref(), @@ -635,18 +660,64 @@ impl ObjectStoreAdapter { async fn put_multipart( &self, path: &object_store::path::Path, + data: ObjectByteStream, + ) -> Result<(), MegaError> { + Self::put_multipart_to(self.to_store(), path, data).await + } + + /// Use one fixed-size upload buffer and await each part before reading more. + /// The backend publishes the object atomically on `complete`; a backend that + /// rejects multipart is an error, not a request to buffer the whole object. + async fn put_multipart_to( + store: &dyn ObjectStore, + path: &object_store::path::Path, mut data: ObjectByteStream, ) -> Result<(), MegaError> { - let mut upload = self - .to_store() + // An empty stream must use an empty PUT: S3 cannot complete a multipart + // upload with no parts. Ignore zero-length input items for this decision. + let first = loop { + match data.try_next().await? { + Some(chunk) if chunk.is_empty() => continue, + Some(chunk) => break chunk, + None => { + store + .put(path, PutPayload::new()) + .await + .map_err(IoOrbitError::from)?; + return Ok(()); + } + } + }; + let mut upload = store .put_multipart(path) .await .map_err(IoOrbitError::from)?; let res = async { - while let Some(chunk) = data.try_next().await? { + let mut buffer = BytesMut::with_capacity(MULTIPART_PART_SIZE); + let mut chunk = first; + loop { + let mut remaining = chunk.as_ref(); + while !remaining.is_empty() { + let take = remaining.len().min(MULTIPART_PART_SIZE - buffer.len()); + buffer.extend_from_slice(&remaining[..take]); + remaining = &remaining[take..]; + if buffer.len() == MULTIPART_PART_SIZE { + upload + .put_part(std::mem::take(&mut buffer).freeze().into()) + .await + .map_err(IoOrbitError::from)?; + buffer.reserve(MULTIPART_PART_SIZE); + } + } + match data.try_next().await? { + Some(next) => chunk = next, + None => break, + } + } + if !buffer.is_empty() { upload - .put_part(chunk.into()) + .put_part(buffer.freeze().into()) .await .map_err(IoOrbitError::from)?; } @@ -657,11 +728,16 @@ impl ObjectStoreAdapter { } .await; - if res.is_err() { - upload.abort().await.map_err(IoOrbitError::from)?; + if let Err(error) = res { + if let Err(abort_error) = upload.abort().await { + return Err(MegaError::ObjStorage(format!( + "{error}; failed to abort multipart upload: {abort_error}" + ))); + } + return Err(error); } - res + Ok(()) } /// Upload an object using a *single PUT* request. @@ -686,7 +762,7 @@ impl ObjectStoreAdapter { /// - backends without stable multipart support /// /// For large objects (Git packfiles, LFS blobs, etc.), - /// `put_stream` + `put_multipart` MUST be used instead. + /// `put_stream_bounded` MUST be used instead when bounded memory is required. async fn put_single( &self, path: &object_store::path::Path, @@ -746,3 +822,421 @@ impl ObjectStoreAdapter { } } } + +#[cfg(test)] +mod tests { + use std::{fmt, io, sync::Mutex}; + + use futures::stream::BoxStream; + use object_store::{ + CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, PutMultipartOptions, + PutResult, UploadPart, memory::InMemory, path::Path, + }; + + use super::*; + + #[derive(Debug, Default)] + struct UploadStats { + started: usize, + part_lengths: Vec, + finished_parts: usize, + single_lengths: Vec, + complete_calls: usize, + abort_calls: usize, + } + + #[derive(Clone, Copy, Debug, Default)] + struct Failures { + head: bool, + part: bool, + complete: bool, + abort: bool, + unsupported: bool, + } + + #[derive(Debug, Default)] + struct StrictStore { + inner: InMemory, + stats: Arc>, + failures: Failures, + } + + impl fmt::Display for StrictStore { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("StrictStore") + } + } + + fn injected_error(message: &'static str) -> object_store::Error { + object_store::Error::Generic { + store: "StrictStore", + source: Box::new(io::Error::other(message)), + } + } + + #[derive(Debug)] + struct StrictUpload { + inner: Box, + stats: Arc>, + failures: Failures, + } + + #[async_trait::async_trait] + impl MultipartUpload for StrictUpload { + fn put_part(&mut self, payload: PutPayload) -> UploadPart { + let length = payload.content_length(); + { + let mut stats = self.stats.lock().unwrap(); + if let Some(previous) = stats.part_lengths.last() { + assert_eq!(*previous, MULTIPART_PART_SIZE, "short non-final part"); + } + assert!(length > 0 && length <= MULTIPART_PART_SIZE); + stats.part_lengths.push(length); + } + if self.failures.part { + return Box::pin(async { Err(injected_error("injected part failure")) }); + } + let part = self.inner.put_part(payload); + let stats = Arc::clone(&self.stats); + Box::pin(async move { + part.await?; + stats.lock().unwrap().finished_parts += 1; + Ok(()) + }) + } + + async fn complete(&mut self) -> object_store::Result { + self.stats.lock().unwrap().complete_calls += 1; + if self.failures.complete { + return Err(injected_error("injected complete failure")); + } + self.inner.complete().await + } + + async fn abort(&mut self) -> object_store::Result<()> { + self.stats.lock().unwrap().abort_calls += 1; + if self.failures.abort { + return Err(injected_error("injected abort failure")); + } + self.inner.abort().await + } + } + + #[async_trait::async_trait] + impl ObjectStore for StrictStore { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { + // Bounded writes may use a single PUT only for an empty object. + assert_eq!(payload.content_length(), 0); + self.stats.lock().unwrap().single_lengths.push(0); + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> object_store::Result> { + self.stats.lock().unwrap().started += 1; + if self.failures.unsupported { + return Err(object_store::Error::NotSupported { + source: Box::new(io::Error::other("multipart disabled")), + }); + } + Ok(Box::new(StrictUpload { + inner: self.inner.put_multipart_opts(location, opts).await?, + stats: Arc::clone(&self.stats), + failures: self.failures, + })) + } + + async fn get_opts( + &self, + location: &Path, + options: GetOptions, + ) -> object_store::Result { + if self.failures.head { + return Err(injected_error("injected head failure")); + } + self.inner.get_opts(location, options).await + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + fn list( + &self, + prefix: Option<&Path>, + ) -> BoxStream<'static, object_store::Result> { + self.inner.list(prefix) + } + + async fn list_with_delimiter( + &self, + prefix: Option<&Path>, + ) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + async fn copy_opts( + &self, + from: &Path, + to: &Path, + options: CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } + } + + #[tokio::test] + async fn exists_distinguishes_not_found_from_storage_failures() { + let store = InMemory::new(); + let path = Path::from("object"); + assert!(!ObjectStoreAdapter::exists_in(&store, &path).await.unwrap()); + + store.put(&path, PutPayload::new()).await.unwrap(); + assert!(ObjectStoreAdapter::exists_in(&store, &path).await.unwrap()); + + let failing = StrictStore { + failures: Failures { + head: true, + ..Default::default() + }, + ..Default::default() + }; + let error = ObjectStoreAdapter::exists_in(&failing, &path) + .await + .unwrap_err(); + assert!(matches!(error, MegaError::ObjStorage(_))); + assert!(error.to_string().contains("injected head failure")); + } + + #[tokio::test] + async fn multipart_aggregates_fixed_parts_with_backpressure() { + let bytes = Bytes::from( + (0..2 * MULTIPART_PART_SIZE + 17) + .map(|index| (index as u8).wrapping_mul(31)) + .collect::>(), + ); + for length in [2 * MULTIPART_PART_SIZE, bytes.len()] { + for input_size in [4096, bytes.len()] { + let store = StrictStore::default(); + let path = Path::from("object"); + let expected = bytes.slice(..length); + let input = expected.clone(); + let stats = Arc::clone(&store.stats); + let source = stream::unfold(0, move |offset| { + let input = input.clone(); + let stats = Arc::clone(&stats); + async move { + // The adapter must finish full parts before polling for + // more bytes, even when inputs are tiny ReaderStream items. + assert_eq!( + stats.lock().unwrap().finished_parts, + offset / MULTIPART_PART_SIZE + ); + if offset == input.len() { + None + } else { + let end = (offset + input_size).min(input.len()); + Some((Ok(input.slice(offset..end)), end)) + } + } + }); + ObjectStoreAdapter::put_multipart_to(&store, &path, Box::pin(source)) + .await + .unwrap(); + assert_eq!( + store.get(&path).await.unwrap().bytes().await.unwrap(), + expected + ); + let stats = store.stats.lock().unwrap(); + let mut lengths = vec![MULTIPART_PART_SIZE; 2]; + if length % MULTIPART_PART_SIZE != 0 { + lengths.push(length % MULTIPART_PART_SIZE); + } + assert_eq!(stats.part_lengths, lengths); + assert_eq!(stats.complete_calls, 1); + assert_eq!(stats.abort_calls, 0); + assert!(stats.single_lengths.is_empty()); + } + } + } + + #[tokio::test] + async fn multipart_empty_object_uses_single_put() { + for empty_items in [0, 3] { + let store = StrictStore::default(); + let path = Path::from("empty"); + let source = stream::iter((0..empty_items).map(|_| Ok(Bytes::new()))); + ObjectStoreAdapter::put_multipart_to(&store, &path, Box::pin(source)) + .await + .unwrap(); + assert_eq!(store.head(&path).await.unwrap().size, 0); + let stats = store.stats.lock().unwrap(); + assert_eq!(stats.started, 0); + assert_eq!(stats.single_lengths, [0]); + } + } + + #[tokio::test] + async fn multipart_failures_abort_and_preserve_existing_object() { + for failures in [ + Failures::default(), + Failures { + part: true, + ..Default::default() + }, + Failures { + complete: true, + ..Default::default() + }, + Failures { + abort: true, + ..Default::default() + }, + ] { + let store = StrictStore { + failures, + ..Default::default() + }; + let path = Path::from("existing"); + store.inner.put(&path, "keep me".into()).await.unwrap(); + let mut items = vec![Ok(Bytes::from(vec![1; MULTIPART_PART_SIZE]))]; + if !failures.part && !failures.complete { + items.push(Err(io::Error::other("injected stream failure"))); + } + let error = + ObjectStoreAdapter::put_multipart_to(&store, &path, Box::pin(stream::iter(items))) + .await + .unwrap_err() + .to_string(); + let phase = if failures.part { + "part" + } else if failures.complete { + "complete" + } else { + "stream" + }; + assert!( + error.contains(&format!("injected {phase} failure")), + "{error}" + ); + if failures.abort { + assert!(error.contains("injected abort failure"), "{error}"); + } + assert_eq!( + store.get(&path).await.unwrap().bytes().await.unwrap(), + "keep me" + ); + let stats = store.stats.lock().unwrap(); + assert_eq!(stats.abort_calls, 1); + assert_eq!(stats.complete_calls, usize::from(failures.complete)); + assert!(stats.single_lengths.is_empty()); + } + } + + #[tokio::test] + async fn multipart_unsupported_does_not_fall_back_to_buffering() { + let store = StrictStore { + failures: Failures { + unsupported: true, + ..Default::default() + }, + ..Default::default() + }; + let path = Path::from("unsupported"); + let source = stream::once(async { Ok(Bytes::from_static(b"data")) }); + assert!( + ObjectStoreAdapter::put_multipart_to(&store, &path, Box::pin(source)) + .await + .is_err() + ); + let stats = store.stats.lock().unwrap(); + assert_eq!(stats.started, 1); + assert!(stats.single_lengths.is_empty()); + } + + #[tokio::test] + async fn bounded_local_upload_streams_then_atomically_replaces() { + let dir = tempfile::tempdir().unwrap(); + let local = Arc::new(LocalFileSystem::new_with_prefix(dir.path()).unwrap()); + let adapter = ObjectStoreAdapter { + store: BackendStore::Local(Arc::clone(&local)), + upload_strategy: UploadStrategy::SinglePut, + }; + let key = ObjectKey { + namespace: ObjectNamespace::Lfs, + key: "a".repeat(64), + }; + let path = key.to_object_store_path(); + local.put(&path, "old object".into()).await.unwrap(); + let target = local.path_to_filesystem(&path).unwrap(); + let observed_path = path.clone(); + let observed_local = Arc::clone(&local); + let observed_parent = target.parent().unwrap().to_path_buf(); + let first = stream::once(async { Ok(Bytes::from(vec![7; MULTIPART_PART_SIZE])) }); + let last = stream::once(async move { + assert_eq!( + observed_local + .get(&observed_path) + .await + .unwrap() + .bytes() + .await + .unwrap(), + "old object" + ); + // A full part has already reached the staging file before EOF; + // SinglePut buffering would leave only the original small file. + assert!(std::fs::read_dir(observed_parent).unwrap().any(|entry| { + // Inspect an open handle, not a potentially stale directory + // entry's cached length while the file is still open. + std::fs::File::open(entry.unwrap().path()) + .unwrap() + .metadata() + .unwrap() + .len() + == MULTIPART_PART_SIZE as u64 + })); + Ok(Bytes::from_static(b"tail")) + }); + adapter + .put_stream_bounded(&key, Box::pin(first.chain(last)), ObjectMeta::default()) + .await + .unwrap(); + let expected = local.get(&path).await.unwrap().bytes().await.unwrap(); + assert_eq!( + &expected[..MULTIPART_PART_SIZE], + vec![7; MULTIPART_PART_SIZE] + ); + assert_eq!(&expected[MULTIPART_PART_SIZE..], b"tail"); + + let failed = stream::iter(vec![ + Ok(Bytes::from(vec![3; MULTIPART_PART_SIZE])), + Err(io::Error::other("injected stream failure")), + ]); + assert!( + adapter + .put_stream_bounded(&key, Box::pin(failed), ObjectMeta::default()) + .await + .is_err() + ); + assert_eq!( + local.get(&path).await.unwrap().bytes().await.unwrap(), + expected + ); + assert_eq!( + std::fs::read_dir(target.parent().unwrap()).unwrap().count(), + 1 + ); + } +} diff --git a/io-orbit/src/object_storage.rs b/io-orbit/src/object_storage.rs index 559f525d1..e3e5a619b 100644 --- a/io-orbit/src/object_storage.rs +++ b/io-orbit/src/object_storage.rs @@ -43,6 +43,8 @@ impl ObjectKey { pub enum ObjectNamespace { Git, Lfs, + /// Private media manifests and chunks, separate from public complete LFS objects. + Media, Log, /// Artifact protocol objects (`docs/artifacts-protocol.md`), keyed by UUID string. Artifact, @@ -53,6 +55,7 @@ impl ObjectNamespace { match self { ObjectNamespace::Git => "git", ObjectNamespace::Lfs => "lfs", + ObjectNamespace::Media => "media", ObjectNamespace::Log => "log", ObjectNamespace::Artifact => "artifact", } @@ -135,6 +138,27 @@ pub trait MegaObjectStorage: Send + Sync { meta: ObjectMeta, ) -> Result<(), MegaError>; + /// Atomically replace an object without buffering its entire contents. + /// + /// Implementations must bound upload buffering independently of object size, + /// apply backpressure to the input, and publish only after the stream succeeds. + /// A caller should also bound each input item, since the implementation must + /// retain that item while consuming it. Unlike `put_stream`, this operation + /// does not inherit create-only or single-PUT upload policies. + /// + /// Unsupported backends must fail explicitly, never fall back to buffering + /// the full object. The default rejects the operation without consuming data. + async fn put_stream_bounded( + &self, + _key: &ObjectKey, + _data: ObjectByteStream, + _meta: ObjectMeta, + ) -> Result<(), MegaError> { + Err(MegaError::ObjStorage( + "storage backend does not support bounded streaming writes".to_owned(), + )) + } + /// Retrieve a single object from the storage backend. /// /// # Returns @@ -309,6 +333,15 @@ mod tests { assert_eq!(key.default_sharding(), "lfs/ab/cd/ef/1234567890"); } + #[test] + fn test_media_namespace_is_separate_from_lfs() { + let key = ObjectKey { + namespace: ObjectNamespace::Media, + key: "abcdef1234567890".to_owned(), + }; + assert_eq!(key.default_sharding(), "media/ab/cd/ef/1234567890"); + } + #[test] fn test_s3_key_git() { let key = ObjectKey { diff --git a/jupiter/src/storage/lfs_db_storage.rs b/jupiter/src/storage/lfs_db_storage.rs index d87aef5e2..a4107a5ad 100644 --- a/jupiter/src/storage/lfs_db_storage.rs +++ b/jupiter/src/storage/lfs_db_storage.rs @@ -2,7 +2,7 @@ use std::ops::Deref; use callisto::{lfs_locks, lfs_objects}; use common::errors::MegaError; -use sea_orm::{EntityTrait, InsertResult, IntoActiveModel, Set}; +use sea_orm::{EntityTrait, InsertResult, IntoActiveModel, Set, TryInsertResult}; use crate::storage::base_storage::{BaseStorage, StorageConnector}; @@ -21,24 +21,23 @@ impl Deref for LfsDbStorage { impl LfsDbStorage { pub async fn new_lfs_object(&self, object: lfs_objects::Model) -> Result { let res = lfs_objects::Entity::insert(object.into_active_model()) + .on_conflict_do_nothing() .exec(self.get_connection()) - .await; - Ok(res.is_ok()) + .await?; + Ok(matches!(res, TryInsertResult::Inserted(_))) } pub async fn get_lfs_object(&self, oid: &str) -> Result, MegaError> { let result = lfs_objects::Entity::find_by_id(oid) .one(self.get_connection()) - .await - .unwrap(); + .await?; Ok(result) } pub async fn delete_lfs_object(&self, oid: String) -> Result<(), MegaError> { lfs_objects::Entity::delete_by_id(oid) .exec(self.get_connection()) - .await - .unwrap(); + .await?; Ok(()) } diff --git a/mono/Cargo.toml b/mono/Cargo.toml index c0731a209..bbb451a0a 100644 --- a/mono/Cargo.toml +++ b/mono/Cargo.toml @@ -74,6 +74,9 @@ utoipa-swagger-ui = { workspace = true, features = ["axum"] } once_cell = { workspace = true } base64 = { workspace = true } +[features] +fastcdc = ["ceres/fastcdc"] + [target.'cfg(not(windows))'.dependencies] jemallocator = { workspace = true } @@ -82,5 +85,7 @@ jemallocator = { workspace = true } mimalloc = { workspace = true } [dev-dependencies] +io-orbit = { workspace = true } +sea-orm = { workspace = true } tempfile = { workspace = true } jupiter-migrate = { workspace = true } diff --git a/mono/src/api/router/lfs_router.rs b/mono/src/api/router/lfs_router.rs index 988c74dc5..2c11907cc 100644 --- a/mono/src/api/router/lfs_router.rs +++ b/mono/src/api/router/lfs_router.rs @@ -61,15 +61,25 @@ use crate::api::{MonoApiServiceState, api_doc::LFS_TAG}; const LFS_CONTENT_TYPE: &str = "application/vnd.git-lfs+json"; const LFS_STREAM_CONTENT_TYPE: &str = "application/octet-stream"; +#[cfg(feature = "fastcdc")] +mod media; + +/// Original repository path, captured before the standard LFS URI rewrite. +#[derive(Clone, Debug)] +pub struct LfsRepository(pub String); + pub fn lfs_routes() -> OpenApiRouter { - OpenApiRouter::new() + let router = OpenApiRouter::new() .routes(routes!(lfs_upload_object)) .routes(routes!(lfs_download_object)) .routes(routes!(list_locks)) .routes(routes!(create_lock)) .routes(routes!(list_locks_for_verification)) .routes(routes!(delete_lock)) - .routes(routes!(lfs_process_batch)) + .routes(routes!(lfs_process_batch)); + #[cfg(feature = "fastcdc")] + let router = router.nest("/libra/media/v1", media::router()); + router } /// The [LFS Server Discovery](https://github.com/git-lfs/git-lfs/blob/main/docs/api/server-discovery.md) @@ -396,22 +406,15 @@ pub async fn lfs_upload_object( Path(oid): Path, req: Request, ) -> Result, (StatusCode, String)> { + let body_bytes = match read_lfs_upload_body(&oid, req).await { + Ok(bytes) => bytes, + Err((code, message)) => return Ok(lfs_error_response(code, message)), + }; let req_obj = RequestObject { oid, ..Default::default() }; - // Collect bytes asynchronously from the stream into a Vec - let body_bytes: Vec = req - .into_body() - .into_data_stream() - .try_fold(Vec::new(), |mut acc, chunk| async move { - acc.extend_from_slice(&chunk); - Ok(acc) - }) - .await - .unwrap(); - let result = state .services() .lfs() @@ -429,6 +432,22 @@ pub async fn lfs_upload_object( } } +async fn read_lfs_upload_body( + oid: &str, + req: Request, +) -> Result, (StatusCode, String)> { + // Reject invalid paths before polling an untrusted upload body. + ceres::lfs::handler::validate_object_oid(oid).map_err(map_lfs_error)?; + req.into_body() + .into_data_stream() + .try_fold(Vec::new(), |mut acc, chunk| async move { + acc.extend_from_slice(&chunk); + Ok(acc) + }) + .await + .map_err(|_| (StatusCode::BAD_REQUEST, "Invalid LFS request body".into())) +} + #[cfg(test)] mod tests { use std::collections::HashMap; @@ -437,6 +456,35 @@ mod tests { use super::*; + #[tokio::test] + async fn invalid_upload_oid_is_rejected_without_reading_body() { + let body = Body::from_stream(futures::stream::poll_fn( + |_| -> std::task::Poll>>> { + panic!("invalid object IDs must be rejected before polling the body") + }, + )); + let (code, message) = + read_lfs_upload_body("media-v1/known-scope/chunks/known-hash", Request::new(body)) + .await + .unwrap_err(); + let response = lfs_error_response(code, message); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()["Content-Type"], LFS_CONTENT_TYPE); + } + + #[tokio::test] + async fn unreadable_upload_body_returns_lfs_error() { + let body = Body::from_stream(futures::stream::once(async { + Err::, _>(std::io::Error::other("test body failure")) + })); + let (code, message) = read_lfs_upload_body(&"a".repeat(64), Request::new(body)) + .await + .unwrap_err(); + let response = lfs_error_response(code, message); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()["Content-Type"], LFS_CONTENT_TYPE); + } + #[test] fn test_map_lfs_error_not_found() { // Test "Not found" error mapping diff --git a/mono/src/api/router/lfs_router/media.rs b/mono/src/api/router/lfs_router/media.rs new file mode 100644 index 000000000..f67255341 --- /dev/null +++ b/mono/src/api/router/lfs_router/media.rs @@ -0,0 +1,192 @@ +//! Authenticated, bounded HTTP adapter for the opt-in media protocol. +use axum::{ + Extension, Json, + body::{Body, to_bytes}, + extract::{FromRef, Path, State}, + http::{Request, StatusCode}, + response::{IntoResponse, Response}, +}; +use ceres::lfs::media::{ + self, MediaError, MediaScope, + protocol::{MAX_MANIFEST_SIZE, MediaManifest}, +}; +use jupiter::{service::lfs_service::LfsService, storage::user_storage::UserStorage}; +use utoipa_axum::{router::OpenApiRouter, routes}; + +use super::{LFS_CONTENT_TYPE, LFS_STREAM_CONTENT_TYPE, LfsRepository}; +use crate::api::{MonoApiServiceState, api_doc::LFS_TAG, oauth::AccessTokenUser}; + +#[derive(Clone)] +pub(super) struct MediaState { + lfs: LfsService, + users: UserStorage, +} + +#[cfg(test)] +mod tests; + +impl FromRef for MediaState { + fn from_ref(state: &MonoApiServiceState) -> Self { + Self { + lfs: state.services().lfs().media_service().clone(), + users: UserStorage::from_ref(state), + } + } +} + +impl FromRef for UserStorage { + fn from_ref(state: &MediaState) -> Self { + state.users.clone() + } +} + +pub(super) fn router() -> OpenApiRouter +where + S: Clone + Send + Sync + 'static, + MediaState: FromRef, + UserStorage: FromRef, +{ + OpenApiRouter::new() + .routes(routes!(capabilities)) + .routes(routes!(prepare)) + .routes(routes!(upload_chunk)) + .routes(routes!(finalize)) + .routes(routes!(manifest)) + .routes(routes!(download_chunk)) +} + +fn scope(user: &AccessTokenUser, repo: Option<&LfsRepository>) -> Result { + let repo = repo.ok_or(MediaError::NotFound)?; + MediaScope::new(&user.0.campsite_user_id, &repo.0) +} + +// Keep errors small until Axum builds the HTTP response at the handler boundary. +#[derive(Debug)] +enum MediaHttpError { + Media(MediaError), + Body, +} + +impl From for MediaHttpError { + fn from(error: MediaError) -> Self { + Self::Media(error) + } +} + +impl IntoResponse for MediaHttpError { + fn into_response(self) -> Response { + match self { + Self::Media(err) => error(err), + Self::Body => ( + StatusCode::PAYLOAD_TOO_LARGE, + Json(serde_json::json!({"message":"media body exceeds limit or cannot be read"})), + ) + .into_response(), + } + } +} + +fn error(err: MediaError) -> Response { + let status = match &err { + MediaError::Invalid(_) => StatusCode::BAD_REQUEST, + MediaError::NotFound => StatusCode::NOT_FOUND, + MediaError::Conflict => StatusCode::CONFLICT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + // Do not return storage keys or backend diagnostics (which include scope IDs). + if status.is_server_error() { + tracing::error!("media storage operation failed: {err}"); + } + let message = if status.is_server_error() { + "media storage operation failed".to_owned() + } else { + err.to_string() + }; + ( + status, + [("Content-Type", LFS_CONTENT_TYPE)], + Json(serde_json::json!({"message": message})), + ) + .into_response() +} + +async fn body(req: Request, max: usize) -> Result, MediaHttpError> { + to_bytes(req.into_body(), max) + .await + .map(|b| b.to_vec()) + .map_err(|_| MediaHttpError::Body) +} + +#[utoipa::path(get, path = "/capabilities", responses((status = 200, description = "FastCDC capabilities"), (status = 401, description = "Access token required")), tag = LFS_TAG)] +async fn capabilities( + user: AccessTokenUser, + repo: Option>, +) -> Result, MediaHttpError> { + scope(&user, repo.as_ref().map(|value| &value.0))?; + Ok(Json(media::protocol::capabilities())) +} + +#[utoipa::path(post, path = "/manifests", responses((status = 200, description = "Prepared manifest and missing chunk hashes")), tag = LFS_TAG)] +async fn prepare( + user: AccessTokenUser, + repo: Option>, + State(state): State, + req: Request, +) -> Result { + let scope = scope(&user, repo.as_ref().map(|value| &value.0))?; + let bytes = body(req, MAX_MANIFEST_SIZE).await?; + let manifest: MediaManifest = serde_json::from_slice(&bytes) + .map_err(|_| MediaError::Invalid("malformed manifest JSON".into()))?; + let response = media::prepare(&state.lfs, &scope, manifest).await?; + Ok(Json(response).into_response()) +} + +#[utoipa::path(put, path = "/manifests/{id}/chunks/{hash}", params(("id" = String, Path), ("hash" = String, Path)), responses((status = 204, description = "Chunk verified and stored")), tag = LFS_TAG)] +async fn upload_chunk( + user: AccessTokenUser, + repo: Option>, + State(state): State, + Path((id, hash)): Path<(String, String)>, + req: Request, +) -> Result { + let scope = scope(&user, repo.as_ref().map(|value| &value.0))?; + let bytes = body(req, media::chunker::MAX_SIZE).await?; + media::upload_chunk(&state.lfs, &scope, &id, &hash, bytes).await?; + Ok(StatusCode::NO_CONTENT) +} + +#[utoipa::path(post, path = "/manifests/{id}/finalize", params(("id" = String, Path)), responses((status = 204, description = "Verified manifest and standard LFS fallback published")), tag = LFS_TAG)] +async fn finalize( + user: AccessTokenUser, + repo: Option>, + State(state): State, + Path(id): Path, +) -> Result { + let scope = scope(&user, repo.as_ref().map(|value| &value.0))?; + media::finalize(&state.lfs, &scope, &id).await?; + Ok(StatusCode::NO_CONTENT) +} + +#[utoipa::path(get, path = "/manifests/by-media/{oid}", params(("oid" = String, Path)), responses((status = 200, description = "Finalized manifest"), (status = 404, description = "No readable manifest")), tag = LFS_TAG)] +async fn manifest( + user: AccessTokenUser, + repo: Option>, + State(state): State, + Path(oid): Path, +) -> Result { + let scope = scope(&user, repo.as_ref().map(|value| &value.0))?; + let response = media::get_manifest(&state.lfs, &scope, &oid).await?; + Ok(Json(response).into_response()) +} + +#[utoipa::path(get, path = "/manifests/by-media/{oid}/chunks/{hash}", params(("oid" = String, Path), ("hash" = String, Path)), responses((status = 200, description = "Verified chunk bytes")), tag = LFS_TAG)] +async fn download_chunk( + user: AccessTokenUser, + repo: Option>, + State(state): State, + Path((oid, hash)): Path<(String, String)>, +) -> Result { + let scope = scope(&user, repo.as_ref().map(|value| &value.0))?; + let bytes = media::download_chunk(&state.lfs, &scope, &oid, &hash).await?; + Ok(([("Content-Type", LFS_STREAM_CONTENT_TYPE)], bytes).into_response()) +} diff --git a/mono/src/api/router/lfs_router/media/tests.rs b/mono/src/api/router/lfs_router/media/tests.rs new file mode 100644 index 000000000..7bb16fb1b --- /dev/null +++ b/mono/src/api/router/lfs_router/media/tests.rs @@ -0,0 +1,274 @@ +use std::sync::Arc; + +use axum::{ + Router, + body::to_bytes, + http::{HeaderMap, Request}, + routing::{get, post}, +}; +use ceres::lfs::{ + handler, + lfs_structs::{BatchRequest, BatchResponse, RequestObject}, +}; +use common::config::{LocalConfig, ObjectStorageBackend, ObjectStorageConfig}; +use io_orbit::factory::ObjectStorageFactory; +use jupiter::storage::{ + base_storage::{BaseStorage, StorageConnector}, + lfs_db_storage::LfsDbStorage, +}; +use sea_orm::{ConnectionTrait, Database}; +use tower::{Layer, ServiceExt}; + +use super::*; + +async fn fixture() -> (tempfile::TempDir, Router) { + let dir = tempfile::tempdir().unwrap(); + let db = Database::connect("sqlite::memory:").await.unwrap(); + db.execute_unprepared("CREATE TABLE lfs_objects (oid TEXT PRIMARY KEY, size BIGINT NOT NULL, exist BOOLEAN NOT NULL)").await.unwrap(); + db.execute_unprepared("CREATE TABLE access_token (id BIGINT PRIMARY KEY, campsite_user_id TEXT NOT NULL, token TEXT NOT NULL, created_at TEXT NOT NULL, github_login TEXT)").await.unwrap(); + db.execute_unprepared("INSERT INTO access_token VALUES (1, 'alice', 'test-alice', '2026-01-01 00:00:00', NULL), (2, 'bob', 'test-bob', '2026-01-01 00:00:00', NULL)").await.unwrap(); + let base = BaseStorage::new(Arc::new(db)); + let config = ObjectStorageConfig { + storage_type: ObjectStorageBackend::Local, + local: LocalConfig { + root_dir: dir.path().to_string_lossy().into_owned(), + }, + ..Default::default() + }; + let state = MediaState { + lfs: LfsService { + lfs_storage: LfsDbStorage { base: base.clone() }, + obj_storage: ObjectStorageFactory::build(&config).await.unwrap(), + }, + users: UserStorage { base }, + }; + let media: Router = router::().with_state(state.clone()).into(); + // The fixture uses the real basic handlers too, so Libra can exercise its + // ordinary Batch -> upload/download entry points rather than bypass them. + let basic = Router::new() + .route("/info/lfs/objects/batch", post(basic_batch)) + .route( + "/info/lfs/objects/{oid}", + get(basic_download).put(basic_upload), + ) + .with_state(state); + let app = Router::new() + .nest("/info/lfs/libra/media/v1", media) + .merge(basic); + let app = tower::util::MapRequestLayer::new( + crate::server::http_server::rewrite_lfs_request_uri::, + ) + .layer(app); + (dir, Router::new().fallback_service(app)) +} + +async fn basic_batch( + State(state): State, + headers: HeaderMap, + Json(request): Json, +) -> Json { + let host = headers.get("host").unwrap().to_str().unwrap(); + Json( + handler::lfs_process_batch(&state.lfs, request, &format!("http://{host}")) + .await + .unwrap(), + ) +} + +async fn basic_download(State(state): State, Path(oid): Path) -> Body { + Body::from_stream(handler::lfs_download_object(state.lfs, oid).await.unwrap()) +} + +async fn basic_upload( + State(state): State, + Path(oid): Path, + request: Request, +) -> StatusCode { + let bytes = to_bytes(request.into_body(), 32 * 1024 * 1024) + .await + .unwrap(); + handler::lfs_upload_object( + &state.lfs, + &RequestObject { + oid, + size: bytes.len() as i64, + ..Default::default() + }, + bytes.to_vec(), + ) + .await + .unwrap(); + StatusCode::OK +} + +#[tokio::test] +async fn media_router_requires_token_and_preserves_repository_scope() { + let (_dir, app) = fixture().await; + let path = "/project/demo.git/info/lfs/libra/media/v1/capabilities"; + for token in [None, Some("unknown-token")] { + let mut request = Request::builder().uri(path); + if let Some(token) = token { + request = request.header("Authorization", format!("Bearer {token}")); + } + let response = app + .clone() + .oneshot(request.body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + let response = app + .clone() + .oneshot( + Request::builder() + .uri(path) + .header("Authorization", "Bearer test-alice") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let value: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), 65536).await.unwrap()).unwrap(); + assert_eq!(value["chunk_algorithms"][0], "fastcdc-v1"); + let path = format!( + "/project/demo.git/info/lfs/libra/media/v1/manifests/by-media/{}", + "a".repeat(64) + ); + let response = app + .oneshot( + Request::builder() + .uri(path) + .header("Authorization", "Bearer test-bob") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn media_http_errors_preserve_response_contract() { + let too_large = body(Request::new(Body::from("ab")), 1).await.unwrap_err(); + for (error, status, content_type, message) in [ + ( + MediaHttpError::from(MediaError::Invalid("bad manifest".into())), + StatusCode::BAD_REQUEST, + LFS_CONTENT_TYPE, + "Invalid media request: bad manifest", + ), + ( + MediaHttpError::from(MediaError::NotFound), + StatusCode::NOT_FOUND, + LFS_CONTENT_TYPE, + "Media object not found", + ), + ( + MediaHttpError::from(MediaError::Conflict), + StatusCode::CONFLICT, + LFS_CONTENT_TYPE, + "Media manifest conflicts with the finalized object", + ), + ( + MediaHttpError::from(MediaError::Storage("private backend diagnostics".into())), + StatusCode::INTERNAL_SERVER_ERROR, + LFS_CONTENT_TYPE, + "media storage operation failed", + ), + ( + too_large, + StatusCode::PAYLOAD_TOO_LARGE, + "application/json", + "media body exceeds limit or cannot be read", + ), + ] { + let response = error.into_response(); + assert_eq!(response.status(), status); + assert_eq!(response.headers()["Content-Type"], content_type); + let value: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), 4096).await.unwrap()).unwrap(); + assert_eq!(value, serde_json::json!({"message": message})); + } +} + +#[tokio::test] +async fn media_router_rejects_invalid_manifest_bodies() { + let (_dir, app) = fixture().await; + let unreadable = Body::from_stream(futures::stream::once(async { + Err::, _>(std::io::Error::other("test body failure")) + })); + for (body, status, content_type, message) in [ + ( + Body::from("{"), + StatusCode::BAD_REQUEST, + LFS_CONTENT_TYPE, + "Invalid media request: malformed manifest JSON", + ), + ( + unreadable, + StatusCode::PAYLOAD_TOO_LARGE, + "application/json", + "media body exceeds limit or cannot be read", + ), + ] { + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/project/demo.git/info/lfs/libra/media/v1/manifests") + .header("Authorization", "Bearer test-alice") + .body(body) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), status); + assert_eq!(response.headers()["Content-Type"], content_type); + let value: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), 4096).await.unwrap()).unwrap(); + assert_eq!(value, serde_json::json!({"message": message})); + } +} + +/// Serves the actual production media router plus token validation against an +/// isolated SQLite database for Libra's ignored cross-repository HTTP test. +#[tokio::test] +#[ignore = "run with MEGA_FASTCDC_READY_FILE for Libra/Mega interop"] +async fn serve_libra_interop() { + let ready = + std::env::var("MEGA_FASTCDC_READY_FILE").expect("MEGA_FASTCDC_READY_FILE is required"); + let (_dir, app) = fixture().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!( + "http://{}/project/demo.git/info/lfs/", + listener.local_addr().unwrap() + ); + std::fs::write( + ready, + serde_json::to_vec(&serde_json::json!({"lfs_url":url,"token":"test-alice"})).unwrap(), + ) + .unwrap(); + let (stop, stopped) = tokio::sync::oneshot::channel(); + let stop = Arc::new(tokio::sync::Mutex::new(Some(stop))); + let app = app.route( + "/__test/stop", + axum::routing::post(move || { + let stop = stop.clone(); + async move { + if let Some(stop) = stop.lock().await.take() { + let _ = stop.send(()); + } + StatusCode::NO_CONTENT + } + }), + ); + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = tokio::time::timeout(std::time::Duration::from_secs(600), stopped).await; + }) + .await + .unwrap(); +} diff --git a/mono/src/server/http_server.rs b/mono/src/server/http_server.rs index 10f5f60b8..e781ec387 100644 --- a/mono/src/server/http_server.rs +++ b/mono/src/server/http_server.rs @@ -497,10 +497,11 @@ pub async fn app(ctx: AppContext, host: String, port: u16) -> Router { .merge(SwaggerUi::new("/swagger-ui").url("/api/openapi.json", api)) } -fn rewrite_lfs_request_uri(mut req: Request) -> Request { +pub(crate) fn rewrite_lfs_request_uri(mut req: Request) -> Request { let full_path = req.uri().path(); if let Some(pos) = full_path.rfind("/info/lfs/") { + let repository = full_path[..pos].to_owned(); let lfs_subpath = &full_path[pos..]; let new_path_and_query = if let Some(query) = req.uri().query() { @@ -509,6 +510,10 @@ fn rewrite_lfs_request_uri(mut req: Request) -> Request { lfs_subpath.to_owned() }; + if !repository.is_empty() { + req.extensions_mut() + .insert(lfs_router::LfsRepository(repository)); + } let new_uri = match Uri::builder().path_and_query(&new_path_and_query).build() { Ok(uri) => uri, Err(e) => {