diff --git a/Cargo.toml b/Cargo.toml index b5ba6ac7..8d6f0819 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ tokio-stream = { version = "0.1.18", features = ["fs"] } bytes = "1.11.1" tempfile = "3.25.0" tower = "0.5.3" -tower-http = { version = "0.6.8", features = ["fs", "compression-gzip", "compression-br", "trace", "cors", "add-extension", "request-id", "set-header"] } +tower-http = { version = "0.6.8", features = ["fs", "compression-gzip", "compression-br", "trace", "cors", "add-extension", "request-id", "set-header", "limit"] } flate2 = "1.1.9" zip = "=6.0.0" tracing = "0.1.44" diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index ba629e65..e452123f 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -70,6 +70,20 @@ pub trait FileUploadUseCase: Send + Sync + 'static { /// Updates the content of an existing file (for WebDAV) async fn update_file(&self, path: &str, content: &[u8]) -> Result<(), DomainError>; + + /// Streaming update — spools body to a temp file with incremental hash, + /// then atomically replaces the file content via dedup store. + /// + /// Peak RAM: ~256 KB regardless of file size. + /// Used by WebDAV PUT for large files. + async fn update_file_streaming( + &self, + path: &str, + temp_path: &Path, + size: u64, + content_type: &str, + pre_computed_hash: Option, + ) -> Result<(), DomainError>; } // ───────────────────────────────────────────────────── diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 989d54e9..ea2b3ac3 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -164,6 +164,20 @@ pub trait FileWritePort: Send + Sync + 'static { async fn update_file_content(&self, file_id: &str, content: Vec) -> Result<(), DomainError>; + /// Streaming update — replaces file content from a temp file on disk. + /// + /// When `pre_computed_hash` is provided, the dedup service skips the + /// hash re-read — zero extra I/O beyond the initial spool. + /// Peak RAM: ~256 KB regardless of file size. + async fn update_file_content_from_temp( + &self, + file_id: &str, + temp_path: &std::path::Path, + size: u64, + content_type: Option, + pre_computed_hash: Option, + ) -> Result<(), DomainError>; + /// Registers file metadata WITHOUT writing content to disk (write-behind). /// /// Returns `(File, PathBuf)` where `PathBuf` is the destination path for the diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index ef9b7d6f..146ee358 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -237,4 +237,66 @@ impl FileUploadUseCase for FileUploadService { .await?; Ok(()) } + + /// Streaming update — replaces file content from a temp file on disk. + /// + /// Uses `update_file_content_from_temp` which passes the pre-computed hash + /// to dedup, avoiding a second full read of the file. + /// For new files (not found at `path`), falls back to `upload_file_streaming`. + /// + /// Peak RAM: ~256 KB regardless of file size. + async fn update_file_streaming( + &self, + path: &str, + temp_path: &Path, + size: u64, + content_type: &str, + pre_computed_hash: Option, + ) -> Result<(), DomainError> { + // Try to find the existing file first + if let Some(file_read) = &self.file_read + && let Some(file) = file_read.find_file_by_path(path).await? + { + self.file_write + .update_file_content_from_temp( + file.id(), + temp_path, + size, + Some(content_type.to_string()), + pre_computed_hash, + ) + .await?; + return Ok(()); + } + + // File doesn't exist — create it via streaming upload + let path_normalized = path.trim_start_matches('/').trim_end_matches('/'); + let (parent_path, filename) = if let Some(idx) = path_normalized.rfind('/') { + (&path_normalized[..idx], &path_normalized[idx + 1..]) + } else { + ("", path_normalized) + }; + + let parent_id = if !parent_path.is_empty() { + if let Some(file_read) = &self.file_read { + file_read.get_parent_folder_id(parent_path).await.ok() + } else { + None + } + } else { + None + }; + + self.file_write + .save_file_from_temp( + filename.to_string(), + parent_id, + content_type.to_string(), + temp_path, + size, + pre_computed_hash, + ) + .await?; + Ok(()) + } } diff --git a/src/common/config.rs b/src/common/config.rs index fb065bda..84c8d08d 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -200,6 +200,9 @@ pub struct StorageConfig { pub parallel_threshold: usize, /// Retention days for files in the trash pub trash_retention_days: u32, + /// Maximum upload file size in bytes (default: 10 GB). + /// Applied as a hard limit to WebDAV PUT and streaming uploads. + pub max_upload_size: usize, } impl Default for StorageConfig { @@ -209,6 +212,7 @@ impl Default for StorageConfig { chunk_size: 1024 * 1024, // 1 MB parallel_threshold: 100 * 1024 * 1024, // 100 MB trash_retention_days: 30, // 30 days + max_upload_size: 10 * 1024 * 1024 * 1024, // 10 GB } } } @@ -566,6 +570,14 @@ impl AppConfig { config.features.enable_search = val; } + // Storage limits + if let Ok(max_upload) = env::var("OXICLOUD_MAX_UPLOAD_SIZE") + .map(|v| v.parse::()) + && let Ok(val) = max_upload + { + config.storage.max_upload_size = val; + } + // OIDC configuration if let Ok(v) = env::var("OXICLOUD_OIDC_ENABLED") { config.oidc.enabled = v.parse::().unwrap_or(false); diff --git a/src/common/stubs.rs b/src/common/stubs.rs index 6ad24f35..6cd37d31 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -193,6 +193,17 @@ impl FileWritePort for StubFileWritePort { Ok(()) } + async fn update_file_content_from_temp( + &self, + _file_id: &str, + _temp_path: &Path, + _size: u64, + _content_type: Option, + _pre_computed_hash: Option, + ) -> Result<(), DomainError> { + Ok(()) + } + async fn register_file_deferred( &self, _name: String, @@ -485,6 +496,17 @@ impl FileUploadUseCase for StubFileUploadUseCase { async fn update_file(&self, _path: &str, _content: &[u8]) -> Result<(), DomainError> { Ok(()) } + + async fn update_file_streaming( + &self, + _path: &str, + _temp_path: &Path, + _size: u64, + _content_type: &str, + _pre_computed_hash: Option, + ) -> Result<(), DomainError> { + Ok(()) + } } // --------------------------------------------------------------------------- diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 230367f5..bc0adac6 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -110,6 +110,74 @@ impl FileBlobWriteRepository { )), } } + + /// Atomically swap the blob hash of a file. + /// + /// Uses a CTE to capture the old hash before updating so the old blob + /// reference can be decremented afterwards. Compensates on failure by + /// removing the new blob reference. + async fn swap_blob_hash( + &self, + file_id: &str, + new_hash: &str, + new_size: i64, + ) -> Result<(), DomainError> { + // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU. + let old_hash = match sqlx::query_scalar::<_, String>( + r#" + WITH old AS ( + SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE + ) + UPDATE storage.files f + SET blob_hash = $1, size = $2, updated_at = NOW() + FROM old + WHERE f.id = old.id + RETURNING old.blob_hash + "#, + ) + .bind(new_hash) + .bind(new_size) + .bind(file_id) + .fetch_optional(self.pool.as_ref()) + .await + { + Ok(Some(old)) => old, + Ok(None) => { + // File not found — compensate: remove the new blob ref + if let Err(e) = self.dedup.remove_reference(new_hash).await { + tracing::error!("Blob orphaned after missing file: {}", e); + } + return Err(DomainError::not_found("File", file_id)); + } + Err(e) => { + // UPDATE failed — compensate: remove the new blob ref + if let Err(rollback_err) = self.dedup.remove_reference(new_hash).await { + tracing::error!( + "Blob orphaned after failed UPDATE — hash: {}, err: {}", + &new_hash[..12], + rollback_err + ); + } + return Err(DomainError::internal_error( + "FileBlobWrite", + format!("update: {e}"), + )); + } + }; + + // Decrement old blob ref (only if hash changed, best-effort) + if old_hash != new_hash { + if let Err(e) = self.dedup.remove_reference(&old_hash).await { + tracing::warn!( + "Failed to decrement old blob ref {}: {}", + &old_hash[..12], + e + ); + } + } + + Ok(()) + } } #[async_trait] @@ -476,62 +544,25 @@ impl FileWritePort for FileBlobWriteRepository { let dedup_result = self.dedup.store_bytes(&content, None).await?; let new_hash = dedup_result.hash().to_string(); - // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU. - // The `old` CTE locks + reads the row *before* the update touches it. - let old_hash = match sqlx::query_scalar::<_, String>( - r#" - WITH old AS ( - SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE - ) - UPDATE storage.files f - SET blob_hash = $1, size = $2, updated_at = NOW() - FROM old - WHERE f.id = old.id - RETURNING old.blob_hash - "#, - ) - .bind(&new_hash) - .bind(new_size) - .bind(file_id) - .fetch_optional(self.pool.as_ref()) - .await - { - Ok(Some(old)) => old, - Ok(None) => { - // File not found — compensate: remove the new blob ref - if let Err(e) = self.dedup.remove_reference(&new_hash).await { - tracing::error!("Blob orphaned after missing file: {}", e); - } - return Err(DomainError::not_found("File", file_id)); - } - Err(e) => { - // UPDATE failed — compensate: remove the new blob ref - if let Err(rollback_err) = self.dedup.remove_reference(&new_hash).await { - tracing::error!( - "Blob orphaned after failed UPDATE — hash: {}, err: {}", - &new_hash[..12], - rollback_err - ); - } - return Err(DomainError::internal_error( - "FileBlobWrite", - format!("update: {e}"), - )); - } - }; + self.swap_blob_hash(file_id, &new_hash, new_size).await + } - // Decrement old blob ref (only if hash changed, best-effort) - if old_hash != new_hash - && let Err(e) = self.dedup.remove_reference(&old_hash).await - { - tracing::warn!( - "Failed to decrement old blob ref {}: {}", - &old_hash[..12], - e - ); - } + async fn update_file_content_from_temp( + &self, + file_id: &str, + temp_path: &std::path::Path, + size: u64, + content_type: Option, + pre_computed_hash: Option, + ) -> Result<(), DomainError> { + // Streaming: pass pre-computed hash so dedup skips re-reading the file. + let dedup_result = self + .dedup + .store_from_file(temp_path, content_type, pre_computed_hash) + .await?; + let new_hash = dedup_result.hash().to_string(); - Ok(()) + self.swap_blob_hash(file_id, &new_hash, size as i64).await } async fn register_file_deferred( diff --git a/src/infrastructure/services/oidc_service.rs b/src/infrastructure/services/oidc_service.rs index 1b0ff911..f4b810d3 100644 --- a/src/infrastructure/services/oidc_service.rs +++ b/src/infrastructure/services/oidc_service.rs @@ -8,11 +8,16 @@ use async_trait::async_trait; use serde::Deserialize; use std::sync::RwLock; +use std::time::{Duration, Instant}; use crate::application::ports::auth_ports::{OidcIdClaims, OidcServicePort, OidcTokenSet}; use crate::common::config::OidcConfig; use crate::common::errors::{DomainError, ErrorKind}; +/// How long discovery/JWKS documents stay cached before re-fetching. +/// 1 hour balances freshness against unnecessary network requests. +const OIDC_CACHE_TTL: Duration = Duration::from_secs(3600); + // ============================================================================ // OIDC Discovery Document // ============================================================================ @@ -92,13 +97,33 @@ struct UserInfoResponse { // OIDC Service // ============================================================================ +/// A cached value with a fetch timestamp for TTL-based expiry. +#[derive(Clone)] +struct Cached { + value: T, + fetched_at: Instant, +} + +impl Cached { + fn new(value: T) -> Self { + Self { + value, + fetched_at: Instant::now(), + } + } + + fn is_expired(&self) -> bool { + self.fetched_at.elapsed() > OIDC_CACHE_TTL + } +} + pub struct OidcService { config: OidcConfig, http_client: reqwest::Client, - /// Cached discovery document - discovery: RwLock>, - /// Cached JWKS (typed JWK keys) - jwks: RwLock>, + /// Cached discovery document (expires after OIDC_CACHE_TTL) + discovery: RwLock>>, + /// Cached JWKS (expires after OIDC_CACHE_TTL) + jwks: RwLock>>, } impl OidcService { @@ -116,16 +141,19 @@ impl OidcService { } } - /// Fetch and cache the OIDC discovery document + /// Fetch and cache the OIDC discovery document (TTL: 1 hour) async fn get_discovery(&self) -> Result { - // Check cache first + // Check cache first (return cached value only if not expired) { let cache = self .discovery .read() .map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?; - if let Some(ref disc) = *cache { - return Ok(disc.clone()); + if let Some(ref cached) = *cache { + if !cached.is_expired() { + return Ok(cached.value.clone()); + } + tracing::debug!("OIDC discovery cache expired, re-fetching"); } } @@ -164,28 +192,31 @@ impl OidcService { ) })?; - // Cache it + // Cache it with timestamp { let mut cache = self .discovery .write() .map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?; - *cache = Some(discovery.clone()); + *cache = Some(Cached::new(discovery.clone())); } Ok(discovery) } - /// Fetch and cache JWKS document for ID token validation + /// Fetch and cache JWKS document for ID token validation (TTL: 1 hour) async fn get_jwks(&self) -> Result { - // Check cache first + // Check cache first (return cached value only if not expired) { let cache = self .jwks .read() .map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?; - if let Some(ref jwks) = *cache { - return Ok(jwks.clone()); + if let Some(ref cached) = *cache { + if !cached.is_expired() { + return Ok(cached.value.clone()); + } + tracing::debug!("OIDC JWKS cache expired, re-fetching"); } } @@ -214,13 +245,13 @@ impl OidcService { ) })?; - // Cache it + // Cache it with timestamp { let mut cache = self .jwks .write() .map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?; - *cache = Some(jwks.clone()); + *cache = Some(Cached::new(jwks.clone())); } Ok(jwks) diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index b2adf31e..bb3bcb8c 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -1,8 +1,6 @@ use async_trait::async_trait; use bytes::Bytes; use image::{ImageFormat, imageops::FilterType}; -use lru::LruCache; -use std::num::NonZeroUsize; /** * Thumbnail Generation Service * @@ -12,13 +10,12 @@ use std::num::NonZeroUsize; * - Background thumbnail generation after upload * - Multiple sizes (icon 150x150, preview 800x600) * - WebP output for smaller file sizes - * - LRU cache for hot thumbnails + * - Lock-free moka cache with weight-based eviction * - Lazy generation on first request if not pre-generated */ use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::fs; -use tokio::sync::RwLock; use crate::application::ports::thumbnail_ports::{ ThumbnailPort, ThumbnailSize as PortThumbnailSize, ThumbnailStatsDto, @@ -76,12 +73,10 @@ struct ThumbnailCacheKey { pub struct ThumbnailService { /// Root path for thumbnail storage thumbnails_root: PathBuf, - /// In-memory LRU cache for hot thumbnails - cache: Arc>>, - /// Maximum cache size in bytes - max_cache_bytes: usize, - /// Current cache size in bytes - current_cache_bytes: Arc>, + /// Lock-free concurrent cache (moka) with weight-based eviction + cache: moka::future::Cache, + /// Configured maximum cache weight (for stats reporting) + max_cache_bytes: u64, } impl ThumbnailService { @@ -89,18 +84,27 @@ impl ThumbnailService { /// /// # Arguments /// * `storage_root` - Root path of file storage - /// * `max_cache_entries` - Maximum number of thumbnails to cache in memory + /// * `max_cache_entries` - (ignored — moka uses weight-based eviction) /// * `max_cache_bytes` - Maximum total bytes to cache pub fn new(storage_root: &Path, max_cache_entries: usize, max_cache_bytes: usize) -> Self { let thumbnails_root = storage_root.join(".thumbnails"); + // Ignore max_cache_entries — weight-based eviction is more accurate + // for variable-size thumbnails than entry-count limits. + let _ = max_cache_entries; + + let cache = moka::future::Cache::builder() + .max_capacity(max_cache_bytes as u64) + .weigher(|_key: &ThumbnailCacheKey, value: &Bytes| -> u32 { + value.len().min(u32::MAX as usize) as u32 + }) + .time_to_live(std::time::Duration::from_secs(600)) + .build(); + Self { thumbnails_root, - cache: Arc::new(RwLock::new(LruCache::new( - NonZeroUsize::new(max_cache_entries).unwrap_or(NonZeroUsize::new(1000).unwrap()), - ))), - max_cache_bytes, - current_cache_bytes: Arc::new(RwLock::new(0)), + cache, + max_cache_bytes: max_cache_bytes as u64, } } @@ -158,13 +162,10 @@ impl ThumbnailService { size, }; - // Check in-memory cache first - { - let cache = self.cache.read().await; - if let Some(data) = cache.peek(&cache_key) { - tracing::debug!("🔥 Thumbnail cache HIT: {} {:?}", file_id, size); - return Ok(data.clone()); - } + // Check lock-free cache first + if let Some(data) = self.cache.get(&cache_key).await { + tracing::debug!("🔥 Thumbnail cache HIT: {} {:?}", file_id, size); + return Ok(data); } // Check if thumbnail exists on disk @@ -177,8 +178,8 @@ impl ThumbnailService { .map_err(|e| ThumbnailError::IoError(e.to_string()))?; let bytes = Bytes::from(data); - // Add to cache - self.add_to_cache(cache_key, bytes.clone()).await; + // Add to cache (lock-free insert — moka handles eviction) + self.cache.insert(cache_key, bytes.clone()).await; tracing::debug!("💾 Thumbnail loaded from disk: {} {:?}", file_id, size); return Ok(bytes); @@ -198,8 +199,8 @@ impl ThumbnailService { .await .map_err(|e| ThumbnailError::IoError(e.to_string()))?; - // Add to cache - self.add_to_cache(cache_key, bytes.clone()).await; + // Add to cache (lock-free insert) + self.cache.insert(cache_key, bytes.clone()).await; Ok(bytes) } @@ -245,31 +246,6 @@ impl ThumbnailService { result.map(Bytes::from) } - /// Add a thumbnail to the in-memory cache - async fn add_to_cache(&self, key: ThumbnailCacheKey, data: Bytes) { - let data_size = data.len(); - - // Check if adding this would exceed max cache size - let mut current_size = self.current_cache_bytes.write().await; - - // Evict items if needed to make room - if *current_size + data_size > self.max_cache_bytes { - let mut cache = self.cache.write().await; - while *current_size + data_size > self.max_cache_bytes && !cache.is_empty() { - if let Some((_, evicted)) = cache.pop_lru() { - *current_size = current_size.saturating_sub(evicted.len()); - } - } - } - - // Add to cache - let mut cache = self.cache.write().await; - if let Some(old) = cache.put(key, data) { - *current_size = current_size.saturating_sub(old.len()); - } - *current_size += data_size; - } - /// Generate all thumbnail sizes for a file in the background /// /// This is called after file upload to pre-generate thumbnails @@ -316,16 +292,12 @@ impl ThumbnailService { .map_err(|e| ThumbnailError::IoError(e.to_string()))?; } - // Remove from cache + // Remove from cache (lock-free invalidation) let cache_key = ThumbnailCacheKey { file_id: file_id.to_string(), size: *size, }; - let mut cache = self.cache.write().await; - if let Some(removed) = cache.pop(&cache_key) { - let mut current_size = self.current_cache_bytes.write().await; - *current_size = current_size.saturating_sub(removed.len()); - } + self.cache.invalidate(&cache_key).await; } tracing::debug!("🗑️ Deleted thumbnails for: {}", file_id); @@ -334,13 +306,10 @@ impl ThumbnailService { /// Get cache statistics pub async fn get_stats(&self) -> ThumbnailStats { - let cache = self.cache.read().await; - let current_size = *self.current_cache_bytes.read().await; - ThumbnailStats { - cached_thumbnails: cache.len(), - cache_size_bytes: current_size, - max_cache_bytes: self.max_cache_bytes, + cached_thumbnails: self.cache.entry_count() as usize, + cache_size_bytes: self.cache.weighted_size() as usize, + max_cache_bytes: self.max_cache_bytes as usize, } } } diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index a75b3e54..b10da8e5 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -29,6 +29,13 @@ const HEADER_DAV: HeaderName = HeaderName::from_static("dav"); const HEADER_LOCK_TOKEN: HeaderName = HeaderName::from_static("lock-token"); // const HEADER_IF: HeaderName = HeaderName::from_static("if"); +/// Maximum body size for XML-based WebDAV requests (PROPFIND, PROPPATCH, LOCK). +/// 1 MB is generous — a typical PROPFIND body is < 1 KB. +const MAX_XML_BODY: usize = 1_048_576; + +/// Maximum body size for MKCOL requests (RFC 4918: body must be empty). +const MAX_MKCOL_BODY: usize = 4096; + /** * Creates and returns the WebDAV router with all required endpoints. * @@ -167,8 +174,8 @@ async fn handle_propfind( // Convert the request into a body let body = req.into_body(); - // Read request body - body::to_bytes(body, usize::MAX) + // Read request body (PROPFIND is XML, 1 MB is more than enough) + body::to_bytes(body, MAX_XML_BODY) .await .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))? }; @@ -340,12 +347,10 @@ async fn handle_proppatch( .get::() .ok_or_else(|| AppError::unauthorized("Authentication required"))?; - // Read request body - let body_bytes = body::to_bytes(req.into_body(), usize::MAX) + // Read request body (XML — bounded to 1 MB) + let body_bytes = body::to_bytes(req.into_body(), MAX_XML_BODY) .await - .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))?; - - // Parse PROPPATCH request + .map_err(|e| AppError::payload_too_large(format!("PROPPATCH body too large or unreadable: {}", e)))?; let (props_to_set, props_to_remove) = WebDavAdapter::parse_proppatch(body_bytes.reader()) .map_err(|e| AppError::bad_request(format!("Failed to parse PROPPATCH request: {}", e)))?; @@ -486,10 +491,12 @@ async fn handle_head( /** * Handles PUT requests to create or update files. * - * This handler creates a new file or updates an existing file at the specified path. + * **Streaming implementation**: the request body is spooled to a temp file + * with incremental SHA-256 hashing. Peak RAM usage is ~256 KB regardless + * of file size. The temp file is then atomically moved into blob storage + * via `update_file_streaming`. * * @param state The application state containing service dependencies - * @param user The authenticated user information * @param path The requested resource path * @param req The HTTP request containing the file contents * @return HTTP response indicating success @@ -499,46 +506,90 @@ async fn handle_put( req: Request, path: String, ) -> Result, AppError> { + use http_body_util::BodyStream; + use sha2::{Digest, Sha256}; + use tokio::io::AsyncWriteExt; + use tokio_stream::StreamExt; + // Get file service from state let file_upload_service = &state.applications.file_upload_service; // Check if path is empty (root folder) - if path.is_empty() || path == "/" { - return Err(AppError::bad_request("Cannot PUT to root folder")); + if path.is_empty() || path == \"/\" { + return Err(AppError::bad_request(\"Cannot PUT to root folder\")); } + // Hard upload size limit from config + let max_upload = state.core.config.storage.max_upload_size; + // Extract content type before consuming the request - let _content_type = req + let content_type = req .headers() .get(header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) - .unwrap_or("application/octet-stream") + .unwrap_or(\"application/octet-stream\") .to_string(); - // Read request body - let body_bytes = { - // Convert the request into a body - let body = req.into_body(); + // ── Streaming spool: body → temp file + incremental hash ── + let temp_file = tempfile::NamedTempFile::new() + .map_err(|e| AppError::internal_error(format!(\"Failed to create temp file: {}\", e)))?; + let temp_path = temp_file.path().to_path_buf(); - // Read request body - body::to_bytes(body, usize::MAX) - .await - .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))? - }; + let mut file = tokio::fs::File::create(&temp_path) + .await + .map_err(|e| AppError::internal_error(format!(\"Failed to open temp file: {}\", e)))?; - // Check if file exists - let file_exists = file_upload_service.update_file(&path, &body_bytes).await; + let mut hasher = Sha256::new(); + let mut total_bytes: usize = 0; + let mut stream = BodyStream::new(req.into_body()); - match file_exists { - Ok(_) => { - // update_file handles both update and create-if-not-found - Ok(Response::builder() - .status(StatusCode::NO_CONTENT) - .body(Body::empty()) - .unwrap()) + while let Some(frame_result) = stream.next().await { + let frame = frame_result + .map_err(|e| AppError::bad_request(format!(\"Failed to read request body: {}\", e)))?; + if let Some(chunk) = frame.data_ref() { + total_bytes += chunk.len(); + if total_bytes > max_upload { + // Abort early — stop reading, delete temp file + drop(file); + let _ = tokio::fs::remove_file(&temp_path).await; + return Err(AppError::payload_too_large(format!( + \"Upload exceeds maximum size of {} bytes\", + max_upload + ))); + } + hasher.update(chunk); + file.write_all(chunk) + .await + .map_err(|e| AppError::internal_error(format!(\"Failed to write to temp file: {}\", e)))?; } + } + file.flush().await + .map_err(|e| AppError::internal_error(format!(\"Failed to flush temp file: {}\", e)))?; + drop(file); + + let hash = hex::encode(hasher.finalize()); + + // ── Atomic store: temp file → dedup blob + DB metadata update ── + let result = file_upload_service + .update_file_streaming( + &path, + &temp_path, + total_bytes as u64, + &content_type, + Some(hash), + ) + .await; + + // Clean up temp file (may already be moved by dedup, ignore error) + let _ = tokio::fs::remove_file(&temp_path).await; + + match result { + Ok(_) => Ok(Response::builder() + .status(StatusCode::NO_CONTENT) + .body(Body::empty()) + .unwrap()), Err(e) => Err(AppError::internal_error(format!( - "Failed to put file: {}", + \"Failed to put file: {}\", e ))), } @@ -572,10 +623,10 @@ async fn handle_mkcol( // Convert the request into a body let body = req.into_body(); - // Read request body - body::to_bytes(body, usize::MAX) + // Read request body (MKCOL — must be empty per RFC 4918) + body::to_bytes(body, MAX_MKCOL_BODY) .await - .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))? + .map_err(|e| AppError::payload_too_large(format!("MKCOL body too large: {}", e)))? }; if !body_bytes.is_empty() { @@ -1055,8 +1106,8 @@ async fn handle_lock( // Convert the request into a body let body = req.into_body(); - // Read request body - body::to_bytes(body, usize::MAX) + // Read request body (LOCK is XML, 1 MB is more than enough) + body::to_bytes(body, MAX_XML_BODY) .await .map_err(|e| AppError::bad_request(format!("Failed to read request body: {}", e)))? }; diff --git a/src/interfaces/errors.rs b/src/interfaces/errors.rs index 6dd05261..71747b6d 100644 --- a/src/interfaces/errors.rs +++ b/src/interfaces/errors.rs @@ -95,6 +95,15 @@ impl AppError { "PreconditionFailed", ) } + + /// Create a 413 Payload Too Large error. + pub fn payload_too_large(message: impl Into) -> Self { + Self::new( + StatusCode::PAYLOAD_TOO_LARGE, + message, + "PayloadTooLarge", + ) + } } impl From for AppError { diff --git a/src/main.rs b/src/main.rs index 855b87c1..57b7cb86 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use axum::Router; use axum::extract::DefaultBodyLimit; use tower_http::trace::TraceLayer; +use tower_http::limit::RequestBodyLimitLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; /// OxiCloud - Cloud Storage Platform @@ -106,6 +107,13 @@ async fn main() -> Result<(), Box> { let carddav_router = carddav_handler::carddav_routes(); let webdav_router = webdav_handler::webdav_routes(); + // CalDAV/CardDAV only carry XML payloads — cap at 1 MB at the transport + // level so `body::to_bytes()` cannot be abused to OOM the server. + // WebDAV is excluded: its streaming PUT handler enforces its own per-upload + // limit from StorageConfig::max_upload_size. + let caldav_router = caldav_router.layer(RequestBodyLimitLayer::new(1_048_576)); + let carddav_router = carddav_router.layer(RequestBodyLimitLayer::new(1_048_576)); + // Build WOPI routes if enabled use oxicloud::interfaces::api::handlers::wopi_handler; let wopi_routes = if config.wopi.enabled {