From af74c940288a24532f68b0a8c46e5606276c9945 Mon Sep 17 00:00:00 2001 From: "M.Schmidt" Date: Wed, 15 Jul 2026 10:50:48 +0200 Subject: [PATCH] fix(webdav): make PATCH's concurrency guard a real compare-and-swap The app-level ETag re-check before the write still left a gap between the check and the actual UPDATE for a concurrent writer to land in. Push the check into the write path itself: swap_blob_hash now takes an expected_hash and only applies the SET under the same FOR UPDATE row lock it already held, closing the race instead of just narrowing it. Adds ErrorKind::PreconditionFailed (412) for the CAS-miss path; PUT/WOPI/chunked-upload keep blind-overwrite semantics by passing None --- src/application/ports/file_ports.rs | 11 ++ src/application/ports/storage_ports.rs | 9 ++ .../services/file_upload_service.rs | 5 +- .../services/trash_service_test.rs | 1 + src/common/stubs.rs | 3 + src/domain/errors.rs | 18 +++ .../pg/file_blob_write_repository.rs | 121 ++++++++++++------ src/interfaces/api/handlers/webdav_handler.rs | 30 ++--- src/interfaces/api/handlers/wopi_handler.rs | 1 + src/interfaces/errors.rs | 1 + src/interfaces/nextcloud/uploads_handler.rs | 5 + src/interfaces/nextcloud/webdav_handler.rs | 30 ++--- 12 files changed, 157 insertions(+), 78 deletions(-) diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index 86539185..98b2e180 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -99,6 +99,16 @@ pub trait FileUploadUseCase: Send + Sync + 'static { /// on the overwrite branch and `authz.require(caller, Create, /// Folder|Drive(id))` on the new-file branch. Handlers just plumb /// `caller_id` through — no protocol-layer authz. + /// + /// `expected_hash`: forwarded to + /// `FileWritePort::update_file_content_with_blob` on the overwrite + /// branch for compare-and-swap; ignored on the new-file branch + /// (nothing to compare against). Pass `None` for plain PUT/WOPI/ + /// chunked-upload last-write-wins semantics; pass the pre-write + /// snapshot's content hash for PATCH, where a concurrent write + /// during the (potentially slow) splice must be rejected rather + /// than silently clobbered. + #[allow(clippy::too_many_arguments)] async fn update_file_streaming_with_perms( &self, path: &str, @@ -107,6 +117,7 @@ pub trait FileUploadUseCase: Send + Sync + 'static { content_type: &str, modified_at: Option, caller_id: Uuid, + expected_hash: Option<&str>, ) -> Result; } diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 549aa6da..b2d4dcbc 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -299,6 +299,14 @@ pub trait FileWritePort: Send + Sync + 'static { /// /// `caller_id` is stamped into `updated_by` alongside the /// `updated_at` bump (§14 provenance). + /// + /// `expected_hash`: when `Some`, makes this a true compare-and-swap — + /// the write only takes effect if the row's current `blob_hash` + /// still equals it, checked and applied atomically under the same + /// row lock (no gap between check and write for a concurrent writer + /// to land in). A mismatch returns `ErrorKind::PreconditionFailed` + /// and leaves the row untouched. `None` keeps the previous + /// blind-overwrite behaviour (PUT/WOPI/chunked-upload finalize). async fn update_file_content_with_blob( &self, file_id: &str, @@ -306,6 +314,7 @@ pub trait FileWritePort: Send + Sync + 'static { size: u64, modified_at: Option, caller_id: Uuid, + expected_hash: Option<&str>, ) -> Result<(String, i64), DomainError>; /// Registers file metadata WITHOUT writing content to disk (write-behind). diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index f73d1b76..9b26acb7 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -305,7 +305,7 @@ impl FileUploadService { let file = file_read.get_file(file_id).await?; let (new_hash, updated_at) = self .file_write - .update_file_content_with_blob(file_id, &blob.hash, blob.size, None, caller_id) + .update_file_content_with_blob(file_id, &blob.hash, blob.size, None, caller_id, None) .await?; // The file maps to a different blob now — stale cached content must // never be served for the rest of its TTI window. @@ -506,6 +506,7 @@ impl FileUploadUseCase for FileUploadService { /// member and cross-tenant PUT. See /// `docs/plan/authz_audit/nextcloud.md` and the sibling native /// `/webdav/*` handler. + #[allow(clippy::too_many_arguments)] async fn update_file_streaming_with_perms( &self, path: &str, @@ -514,6 +515,7 @@ impl FileUploadUseCase for FileUploadService { content_type: &str, modified_at: Option, caller_id: Uuid, + expected_hash: Option<&str>, ) -> Result { let Some(authz) = &self.authorization else { return Err(DomainError::internal_error( @@ -552,6 +554,7 @@ impl FileUploadUseCase for FileUploadService { blob.size, modified_at, caller_id, + expected_hash, ) .await?; // Invalidate content cache — file content has changed. diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index 16e8a40a..f7572434 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -589,6 +589,7 @@ impl FileWritePort for MockFileRepository { _size: u64, _modified_at: Option, _caller_id: Uuid, + _expected_hash: Option<&str>, ) -> std::result::Result<(String, i64), DomainError> { Ok((String::new(), 0)) } diff --git a/src/common/stubs.rs b/src/common/stubs.rs index 8743fca8..4aa44b57 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -196,6 +196,7 @@ impl FileWritePort for StubFileWritePort { _size: u64, _modified_at: Option, _caller_id: Uuid, + _expected_hash: Option<&str>, ) -> Result<(String, i64), DomainError> { Ok((String::new(), 0)) } @@ -491,6 +492,7 @@ impl FileUploadUseCase for StubFileUploadUseCase { Ok(FileDto::default()) } + #[allow(clippy::too_many_arguments)] async fn update_file_streaming_with_perms( &self, _path: &str, @@ -499,6 +501,7 @@ impl FileUploadUseCase for StubFileUploadUseCase { _content_type: &str, _modified_at: Option, _caller_id: Uuid, + _expected_hash: Option<&str>, ) -> Result { Ok(FileDto::default()) } diff --git a/src/domain/errors.rs b/src/domain/errors.rs index ede2a9f1..0d6f2f44 100644 --- a/src/domain/errors.rs +++ b/src/domain/errors.rs @@ -39,6 +39,12 @@ pub enum ErrorKind { /// `AlreadyExists` (which is a uniqueness violation) so audit /// readers can tell them apart. Conflict, + /// RFC 7232 precondition failure — a caller-supplied conditional + /// (If-Match, or an internal compare-and-swap standing in for one) + /// did not hold against the resource's current state. Maps to + /// HTTP 412. Distinct from `Conflict` (409): this is specifically + /// "the state you thought you were writing against has moved." + PreconditionFailed, } impl ErrorKind { @@ -58,6 +64,7 @@ impl ErrorKind { ErrorKind::DatabaseError => "Database Error", ErrorKind::QuotaExceeded => "Quota Exceeded", ErrorKind::Conflict => "Conflict", + ErrorKind::PreconditionFailed => "Precondition Failed", } } } @@ -196,6 +203,17 @@ impl DomainError { } } + /// Creates a precondition-failed error (RFC 7232 / CAS mismatch) + pub fn precondition_failed>(entity_type: &'static str, message: S) -> Self { + Self { + kind: ErrorKind::PreconditionFailed, + entity_type, + entity_id: None, + message: message.into(), + source: None, + } + } + /// Creates a validation error pub fn validation_error>(message: S) -> Self { Self { diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index f4673fc1..b84a5378 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -153,6 +153,15 @@ impl FileBlobWriteRepository { /// row — not the row's owner. D2 shared drives let non-owners /// overwrite content; the previous `updated_by = f.user_id` would /// have silently recorded the wrong principal. + /// `expected_hash`, when `Some`, turns this into a real + /// compare-and-swap: the SET clause only takes effect if the row's + /// `blob_hash` still matches at the moment the `FOR UPDATE` lock is + /// held (same statement, same transaction — no gap a concurrent + /// writer can land in). A mismatch leaves the row untouched and is + /// reported back via the `matched` flag rather than silently + /// overwriting a sibling PATCH's content. `None` preserves the old + /// blind-overwrite behaviour for PUT/WOPI/chunked-upload finalize, + /// where last-write-wins is the intended HTTP semantics. async fn swap_blob_hash( &self, file_id: &str, @@ -160,57 +169,83 @@ impl FileBlobWriteRepository { new_size: i64, modified_at: Option, caller_id: Uuid, + expected_hash: Option<&str>, ) -> Result<(String, i64), DomainError> { - // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU. + // Atomic CTE: capture old hash then conditionally update in one + // round-trip, no TOCTOU. The CASE arms make the SET a no-op when + // `expected_hash` is given and doesn't match `old.blob_hash` — + // the row is still returned (with its unchanged values) so the + // caller can tell "mismatch" apart from "file not found". // Deadlock victims (40P01) retry before the compensation below runs — // a successful retry must keep the new blob reference alive. - let (old_hash, updated_at) = match retry_on_deadlock("files.swap_blob_hash", || { - sqlx::query_as::<_, (String, i64)>( - r#" + let (old_hash, updated_at, matched) = + match retry_on_deadlock("files.swap_blob_hash", || { + sqlx::query_as::<_, (String, i64, bool)>( + 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 = COALESCE(to_timestamp($4), NOW()), - updated_by = $5 + SET blob_hash = CASE WHEN $6::text IS NULL OR old.blob_hash = $6 + THEN $1 ELSE f.blob_hash END, + size = CASE WHEN $6::text IS NULL OR old.blob_hash = $6 + THEN $2 ELSE f.size END, + updated_at = CASE WHEN $6::text IS NULL OR old.blob_hash = $6 + THEN COALESCE(to_timestamp($4), NOW()) ELSE f.updated_at END, + updated_by = CASE WHEN $6::text IS NULL OR old.blob_hash = $6 + THEN $5 ELSE f.updated_by END FROM old WHERE f.id = old.id - RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint + RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint, + ($6::text IS NULL OR old.blob_hash = $6) "#, - ) - .bind(new_hash) - .bind(new_size) - .bind(file_id) - .bind(modified_at.map(|t| t as f64)) - .bind(caller_id) - .fetch_optional(self.pool.as_ref()) - }) - .await - { - Ok(Some(row)) => row, - 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); + ) + .bind(new_hash) + .bind(new_size) + .bind(file_id) + .bind(modified_at.map(|t| t as f64)) + .bind(caller_id) + .bind(expected_hash) + .fetch_optional(self.pool.as_ref()) + }) + .await + { + Ok(Some(row)) => row, + 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)); } - 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 - ); + 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}"), + )); } - return Err(DomainError::internal_error( - "FileBlobWrite", - format!("update: {e}"), - )); + }; + + if !matched { + // CAS lost the race — some other writer's content is now the + // row's truth. Release the blob we ingested for nothing; + // nothing was written. + if let Err(e) = self.dedup.remove_reference(new_hash).await { + tracing::error!("Blob orphaned after CAS mismatch: {}", e); } - }; + return Err(DomainError::precondition_failed( + "File", + "content was modified concurrently", + )); + } // Decrement old blob ref (only if hash changed, best-effort) if old_hash != new_hash @@ -790,12 +825,20 @@ impl FileWritePort for FileBlobWriteRepository { size: u64, modified_at: Option, caller_id: Uuid, + expected_hash: Option<&str>, ) -> Result<(String, i64), DomainError> { // The content was already ingested into the chunk store by the // upload-ingest layer; swap_blob_hash consumes its reference and // releases it on failure. let swapped = self - .swap_blob_hash(file_id, blob_hash, size as i64, modified_at, caller_id) + .swap_blob_hash( + file_id, + blob_hash, + size as i64, + modified_at, + caller_id, + expected_hash, + ) .await?; // The file now maps to a different blob — drop the read-side cache // entry so streaming downloads cannot serve the previous content diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index e628b2e5..3d10fcf3 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -1972,6 +1972,7 @@ async fn handle_put( &content_type, None, user.id, + None, ) .await; @@ -2303,25 +2304,15 @@ async fn handle_patch( )); } - // ── Optimistic-concurrency re-check ─────────────────────────────── - // `file.etag` was snapshotted before the (potentially slow) splice + - // CAS-ingest above. Re-verify nothing else wrote to this file in the - // meantime, narrowing the window in which two concurrent PATCHes to - // disjoint ranges — each individually passing its own If-Match check - // against the same stale snapshot — could otherwise silently clobber - // each other on the blind-overwrite write path below. - if let Ok(current) = file_retrieval_service - .get_file_by_path(&path, drive_id) - .await - && current.etag != file.etag - { - upload_ingest::discard_ingested(&state.core.dedup_service, &ingested).await; - return Err(AppError::precondition_failed( - "File was modified concurrently — retry the PATCH", - )); - } - - // ── Atomic store ────────────────────────────────────────────────── + // ── Atomic store, compare-and-swap on the pre-splice content hash ── + // `file.content_hash` was snapshotted before the (potentially slow) + // splice + CAS-ingest above. Passing it as `expected_hash` makes the + // write itself a compare-and-swap: the repository checks and applies + // under the same row lock, so nothing else can write to this file + // between the check and the write. This is what actually closes the + // race two concurrent PATCHes to disjoint ranges could otherwise hit + // — each individually passing its own If-Match check against the + // same stale snapshot, then blindly overwriting each other. let new_size = ingested.size; let content_type = ingested.content_type.clone(); let result = file_upload_service @@ -2332,6 +2323,7 @@ async fn handle_patch( &content_type, None, user.id, + Some(&file.content_hash), ) .await; diff --git a/src/interfaces/api/handlers/wopi_handler.rs b/src/interfaces/api/handlers/wopi_handler.rs index f3a2c1f2..36929b9f 100644 --- a/src/interfaces/api/handlers/wopi_handler.rs +++ b/src/interfaces/api/handlers/wopi_handler.rs @@ -413,6 +413,7 @@ async fn put_file( &content_type, None, claims_sub_uuid, + None, ) .await; diff --git a/src/interfaces/errors.rs b/src/interfaces/errors.rs index 148ad70c..a859e261 100644 --- a/src/interfaces/errors.rs +++ b/src/interfaces/errors.rs @@ -131,6 +131,7 @@ impl From for AppError { ErrorKind::DatabaseError => StatusCode::INTERNAL_SERVER_ERROR, ErrorKind::QuotaExceeded => StatusCode::INSUFFICIENT_STORAGE, ErrorKind::Conflict => StatusCode::CONFLICT, + ErrorKind::PreconditionFailed => StatusCode::PRECONDITION_FAILED, }; Self { diff --git a/src/interfaces/nextcloud/uploads_handler.rs b/src/interfaces/nextcloud/uploads_handler.rs index ccd3e66a..47c05c56 100644 --- a/src/interfaces/nextcloud/uploads_handler.rs +++ b/src/interfaces/nextcloud/uploads_handler.rs @@ -466,6 +466,10 @@ async fn handle_assemble( // AuthZ audit #2 (2026-07-12): route DomainError through // `AppError::from` so authz denials keep the graduated 403/404 // shape instead of collapsing into 500. + // + // No client-supplied ETag to enforce here (NC chunked MOVE has no + // If-Match semantics) — `expected_hash: None`, same as every other + // plain-write callsite; only PATCH's CAS passes `Some(&hash)`. let dto = match upload_service .update_file_streaming_with_perms( &internal_path, @@ -474,6 +478,7 @@ async fn handle_assemble( &content_type, oc_mtime, user.id, + None, ) .await { diff --git a/src/interfaces/nextcloud/webdav_handler.rs b/src/interfaces/nextcloud/webdav_handler.rs index b571cf6f..6f6b90c3 100644 --- a/src/interfaces/nextcloud/webdav_handler.rs +++ b/src/interfaces/nextcloud/webdav_handler.rs @@ -912,6 +912,7 @@ async fn handle_put( &content_type, oc_mtime, session.user.id, + None, ) .await .map_err(AppError::from)?; @@ -1145,25 +1146,15 @@ async fn handle_patch( )); } - // ── Optimistic-concurrency re-check ─────────────────────────────── - // `file.etag` was snapshotted before the (potentially slow) splice + - // CAS-ingest above. Re-verify nothing else wrote to this file in the - // meantime, narrowing the window in which two concurrent PATCHes to - // disjoint ranges — each individually passing its own If-Match check - // against the same stale snapshot — could otherwise silently clobber - // each other on the blind-overwrite write path below. - if let Ok(current) = file_service - .get_file_by_path(&internal_path, chroot.drive_id) - .await - && current.etag != file.etag - { - discard_ingested(&state.core.dedup_service, &ingested).await; - return Err(AppError::precondition_failed( - "File was modified concurrently — retry the PATCH", - )); - } - - // ── Atomic store ────────────────────────────────────────────────── + // ── Atomic store, compare-and-swap on the pre-splice content hash ── + // `file.content_hash` was snapshotted before the (potentially slow) + // splice + CAS-ingest above. Passing it as `expected_hash` makes the + // write itself a compare-and-swap: the repository checks and applies + // under the same row lock, so nothing else can write to this file + // between the check and the write. This is what actually closes the + // race two concurrent PATCHes to disjoint ranges could otherwise hit + // — each individually passing its own If-Match check against the + // same stale snapshot, then blindly overwriting each other. let new_size = ingested.size; let content_type = ingested.content_type.clone(); let stored = upload_service @@ -1174,6 +1165,7 @@ async fn handle_patch( &content_type, None, session.user.id, + Some(&file.content_hash), ) .await .map_err(AppError::from)?;