fix(webdav): close PATCH gaps found in review (quota, authz, cap, races)
Fixes to the RFC 5789 PATCH implementation found by review of the rfc-5789-http-patch branch: - nextcloud/webdav_handler.rs::handle_patch now enforces storage quota before committing, matching the plain WebDAV surface (was a quota bypass via the NextCloud endpoint). - The plain surface's If-Match/If-None-Match comparison reused a hand-rolled single-value strong compare that mishandled weak (W/) validators and multi-value lists. Moved the correct RFC 7232 helpers (already used by the NC surface) into the shared handler file so both surfaces use one conformant implementation. - NC handle_patch resolved the target file via get_file_by_path, which performs no authorization check, before any permission-gated call — for a full-file-range patch this could leak size/ETag via 412/416 responses to a caller without Read on that file. Added the same explicit authz.require(Read, ...) the plain surface already has. - ingest_range_patch_to_cas capped the whole spliced stream (prefix + edit + suffix) against direct_put_max_bytes, so PATCH became permanently unusable on any file at or above that size regardless of edit size. The cap now only bounds the edit itself. - NC handle_patch had no active-lock guard, so a LOCK taken via /webdav/ didn't protect the same file reached through /remote.php/dav/. Now shares enforce_native_lock with the plain surface. - Added a re-check of the file's ETag immediately before the write on both surfaces, narrowing (not eliminating — that would need compare-and-swap support in the write path) the window in which two concurrent PATCHes to disjoint ranges could silently clobber each other. - NC handle_patch returned 404 for a PATCH on a directory instead of 409 like the plain surface; now checks folder existence first. - The Content-Length-vs-X-Update-Range span check only fired when Content-Length was present, so a chunked-transfer body could silently diverge from the declared span. ingest_range_patch_to_cas now counts actual body bytes and validates against the declared span regardless, discarding the ingested blob on mismatch.
This commit is contained in:
@@ -1656,7 +1656,7 @@ fn evaluate_if_header(
|
||||
///
|
||||
/// Shared by `handle_put`, `handle_delete`, `handle_move`,
|
||||
/// `handle_copy`, and `handle_proppatch`.
|
||||
fn enforce_native_lock(
|
||||
pub(crate) fn enforce_native_lock(
|
||||
lock_store: &crate::infrastructure::services::webdav_lock_service::WebDavLockStore,
|
||||
if_header: Option<&str>,
|
||||
path: &str,
|
||||
@@ -2042,6 +2042,59 @@ pub(crate) fn parse_update_range(header: &str, size: u64) -> Result<(u64, Option
|
||||
Ok((start, Some(end)))
|
||||
}
|
||||
|
||||
/// Strip the optional `W/` weak prefix and surrounding double-quotes
|
||||
/// from one ETag value in an `If-Match` / `If-None-Match` list. Returns
|
||||
/// `(is_weak, inner)`.
|
||||
fn parse_etag_value(raw: &str) -> (bool, &str) {
|
||||
let trimmed = raw.trim();
|
||||
if let Some(rest) = trimmed.strip_prefix("W/") {
|
||||
(true, rest.trim().trim_matches('"'))
|
||||
} else {
|
||||
(false, trimmed.trim_matches('"'))
|
||||
}
|
||||
}
|
||||
|
||||
/// RFC 7232 §3.2 — `If-None-Match` fails when:
|
||||
/// - the header value is `*` and a current representation exists, OR
|
||||
/// - any listed ETag matches the current representation (weak comparison
|
||||
/// — weak validators in the request are equivalent to strong for the
|
||||
/// match itself, only If-Match is required to be strong).
|
||||
///
|
||||
/// `pub(crate)` so both the plain and NextCloud-surface WebDAV handlers
|
||||
/// share one RFC 7232-conformant implementation instead of each
|
||||
/// reimplementing ETag comparison (multi-value lists, `W/` weak prefix).
|
||||
pub(crate) fn if_none_match_precondition_fails(header: &str, current_etag: Option<&str>) -> bool {
|
||||
let v = header.trim();
|
||||
if v == "*" {
|
||||
return current_etag.is_some();
|
||||
}
|
||||
let Some(current) = current_etag else {
|
||||
return false;
|
||||
};
|
||||
v.split(',').any(|tag| {
|
||||
let (_, parsed) = parse_etag_value(tag);
|
||||
!parsed.is_empty() && parsed == current
|
||||
})
|
||||
}
|
||||
|
||||
/// RFC 7232 §3.1 — `If-Match` fails when:
|
||||
/// - the resource doesn't currently exist (no strong validator to match), OR
|
||||
/// - the header isn't `*` and no listed ETag strong-matches the current one
|
||||
/// (weak validators in the request never satisfy a strong-match).
|
||||
pub(crate) fn if_match_precondition_fails(header: &str, current_etag: Option<&str>) -> bool {
|
||||
let v = header.trim();
|
||||
let Some(current) = current_etag else {
|
||||
return true;
|
||||
};
|
||||
if v == "*" {
|
||||
return false;
|
||||
}
|
||||
!v.split(',').any(|tag| {
|
||||
let (is_weak, parsed) = parse_etag_value(tag);
|
||||
!is_weak && !parsed.is_empty() && parsed == current
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles PATCH requests (RFC 5789) for partial byte-range content updates.
|
||||
*
|
||||
@@ -2164,22 +2217,18 @@ async fn handle_patch(
|
||||
}
|
||||
|
||||
// ── RFC 7232 conditional preconditions ────────────────────────────
|
||||
if let Some(ref inm) = if_none_match {
|
||||
let server_tag = file.etag.trim_matches('"');
|
||||
if inm == "*" || inm.trim_matches('"') == server_tag {
|
||||
return Err(AppError::precondition_failed(
|
||||
"If-None-Match — resource already exists with that ETag",
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(ref im) = if_match
|
||||
&& im != "*"
|
||||
let current_etag = Some(file.etag.as_str());
|
||||
if let Some(ref value) = if_none_match
|
||||
&& if_none_match_precondition_fails(value, current_etag)
|
||||
{
|
||||
let client_tag = im.trim_matches('"');
|
||||
let server_tag = file.etag.trim_matches('"');
|
||||
if client_tag != server_tag {
|
||||
return Err(AppError::precondition_failed("If-Match — ETag mismatch"));
|
||||
}
|
||||
return Err(AppError::precondition_failed(
|
||||
"If-None-Match — resource already exists with that ETag",
|
||||
));
|
||||
}
|
||||
if let Some(ref value) = if_match
|
||||
&& if_match_precondition_fails(value, current_etag)
|
||||
{
|
||||
return Err(AppError::precondition_failed("If-Match — ETag mismatch"));
|
||||
}
|
||||
|
||||
// ── Range parsing + validation ─────────────────────────────────────
|
||||
@@ -2205,6 +2254,10 @@ async fn handle_patch(
|
||||
.map_err(AppError::from)?,
|
||||
)
|
||||
};
|
||||
let suffix_len = match end {
|
||||
Some(end) if end + 1 < file.size => file.size - (end + 1),
|
||||
_ => 0,
|
||||
};
|
||||
let suffix_stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>> = match end
|
||||
{
|
||||
Some(end) if end + 1 < file.size => Box::into_pin(
|
||||
@@ -2215,16 +2268,18 @@ async fn handle_patch(
|
||||
),
|
||||
_ => Box::pin(stream::empty()),
|
||||
};
|
||||
|
||||
let filename = crate::common::mime_detect::filename_from_path(&path).to_string();
|
||||
let ingested = upload_ingest::ingest_range_patch_to_cas(
|
||||
prefix_stream,
|
||||
(prefix_stream, start),
|
||||
req.into_body(),
|
||||
suffix_stream,
|
||||
(suffix_stream, suffix_len),
|
||||
&state.core.dedup_service,
|
||||
&filename,
|
||||
&content_type,
|
||||
max_upload,
|
||||
upload_ingest::PatchIngestBudget {
|
||||
max_bytes: max_upload,
|
||||
expected_body_len: end.map(|end| end - start + 1),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -2248,6 +2303,24 @@ 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 ──────────────────────────────────────────────────
|
||||
let new_size = ingested.size;
|
||||
let content_type = ingested.content_type.clone();
|
||||
|
||||
@@ -24,6 +24,7 @@ use crate::application::ports::file_ports::{
|
||||
FileManagementUseCase, FileRetrievalUseCase, FileUploadUseCase,
|
||||
};
|
||||
use crate::application::ports::folder_ports::FolderUseCase;
|
||||
use crate::application::ports::storage_ports::StorageUsagePort;
|
||||
use crate::application::ports::trash_ports::TrashUseCase;
|
||||
use crate::common::di::AppState;
|
||||
use crate::common::mime_detect::filename_from_path;
|
||||
@@ -31,12 +32,15 @@ use crate::domain::services::authorization::{Permission, Resource, Subject};
|
||||
use crate::infrastructure::services::path_resolver_service::ResolvedResource;
|
||||
use crate::infrastructure::services::webdav_dead_property_store::ResourceRef;
|
||||
use crate::interfaces::api::handlers::webdav_handler::{
|
||||
PROPFIND_BATCH_SIZE, dead_props_for, file_dead_props, files_dead_props_map, folder_dead_props, parse_update_range,
|
||||
PROPFIND_BATCH_SIZE, dead_props_for, enforce_native_lock, file_dead_props, files_dead_props_map, folder_dead_props,
|
||||
if_match_precondition_fails, if_none_match_precondition_fails, parse_update_range,
|
||||
folders_dead_props_map, streamed_file_dead_props,
|
||||
};
|
||||
use crate::interfaces::errors::AppError;
|
||||
use crate::interfaces::range_requests::{not_modified_response, range_response};
|
||||
use crate::interfaces::upload_ingest::{ingest_body_to_cas, ingest_range_patch_to_cas};
|
||||
use crate::interfaces::upload_ingest::{
|
||||
PatchIngestBudget, discard_ingested, ingest_body_to_cas, ingest_range_patch_to_cas,
|
||||
};
|
||||
|
||||
/// Extension trait to map XML write errors to `String` concisely.
|
||||
trait XmlResultExt<T> {
|
||||
@@ -793,55 +797,6 @@ async fn handle_proppatch(
|
||||
|
||||
// ──────────────────── PUT ────────────────────
|
||||
|
||||
/// Strip the optional `W/` weak prefix and surrounding double-quotes
|
||||
/// from one ETag value in an `If-Match` / `If-None-Match` list. Returns
|
||||
/// `(is_weak, inner)`.
|
||||
fn parse_etag_value(raw: &str) -> (bool, &str) {
|
||||
let trimmed = raw.trim();
|
||||
if let Some(rest) = trimmed.strip_prefix("W/") {
|
||||
(true, rest.trim().trim_matches('"'))
|
||||
} else {
|
||||
(false, trimmed.trim_matches('"'))
|
||||
}
|
||||
}
|
||||
|
||||
/// RFC 7232 §3.2 — `If-None-Match` fails for PUT when:
|
||||
/// - the header value is `*` and a current representation exists, OR
|
||||
/// - any listed ETag matches the current representation (weak comparison
|
||||
/// — weak validators in the request are equivalent to strong for the
|
||||
/// match itself, only If-Match is required to be strong).
|
||||
fn if_none_match_precondition_fails(header: &str, current_etag: Option<&str>) -> bool {
|
||||
let v = header.trim();
|
||||
if v == "*" {
|
||||
return current_etag.is_some();
|
||||
}
|
||||
let Some(current) = current_etag else {
|
||||
return false;
|
||||
};
|
||||
v.split(',').any(|tag| {
|
||||
let (_, parsed) = parse_etag_value(tag);
|
||||
!parsed.is_empty() && parsed == current
|
||||
})
|
||||
}
|
||||
|
||||
/// RFC 7232 §3.1 — `If-Match` fails for PUT when:
|
||||
/// - the resource doesn't currently exist (no strong validator to match), OR
|
||||
/// - the header isn't `*` and no listed ETag strong-matches the current one
|
||||
/// (weak validators in the request never satisfy a strong-match).
|
||||
fn if_match_precondition_fails(header: &str, current_etag: Option<&str>) -> bool {
|
||||
let v = header.trim();
|
||||
let Some(current) = current_etag else {
|
||||
return true;
|
||||
};
|
||||
if v == "*" {
|
||||
return false;
|
||||
}
|
||||
!v.split(',').any(|tag| {
|
||||
let (is_weak, parsed) = parse_etag_value(tag);
|
||||
!is_weak && !parsed.is_empty() && parsed == current
|
||||
})
|
||||
}
|
||||
|
||||
fn precondition_failed_response() -> Response<Body> {
|
||||
Response::builder()
|
||||
.status(StatusCode::PRECONDITION_FAILED)
|
||||
@@ -1034,15 +989,67 @@ async fn handle_patch(
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("application/octet-stream")
|
||||
.to_string();
|
||||
let if_header = req
|
||||
.headers()
|
||||
.get("If")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
let max_upload = state.core.config.storage.direct_put_max_bytes;
|
||||
|
||||
// ── Existence check ───────────────────────────────────────────────
|
||||
// Unlike PUT, PATCH requires an existing file — a partial update of
|
||||
// nothing isn't meaningful.
|
||||
let file = file_service
|
||||
// nothing isn't meaningful. On lookup failure, distinguish "it's a
|
||||
// directory" (409, matching the plain WebDAV surface) from "it
|
||||
// doesn't exist at all" (404) instead of collapsing both to 404.
|
||||
let file = match file_service
|
||||
.get_file_by_path(&internal_path, chroot.drive_id)
|
||||
.await
|
||||
{
|
||||
Ok(file) => file,
|
||||
Err(_) => {
|
||||
if state
|
||||
.applications
|
||||
.folder_service
|
||||
.get_folder_by_path(&internal_path, chroot.drive_id)
|
||||
.await
|
||||
.is_ok()
|
||||
{
|
||||
return Err(AppError::conflict("Cannot PATCH a directory"));
|
||||
}
|
||||
return Err(AppError::not_found(format!(
|
||||
"File not found: {}",
|
||||
internal_path
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
// `get_file_by_path` performs no authorization check (see its own
|
||||
// doc comment) — mirrors the plain WebDAV surface's explicit
|
||||
// defense-in-depth Read check right after resolving the file, so a
|
||||
// caller without Read on this specific file can't learn its size or
|
||||
// ETag via the precondition/range-bounds responses below.
|
||||
let file_uuid = Uuid::parse_str(&file.id)
|
||||
.map_err(|_| AppError::not_found(format!("File not found: {}", internal_path)))?;
|
||||
state
|
||||
.authorization
|
||||
.require(
|
||||
Subject::User(session.user.id),
|
||||
Permission::Read,
|
||||
Resource::File(file_uuid),
|
||||
)
|
||||
.await?;
|
||||
|
||||
// ── Active-lock guard (RFC 4918 §10.4 If: evaluation) ─────────────
|
||||
// Shared with the plain WebDAV surface so a LOCK taken via /webdav/
|
||||
// also protects the same file reached through /remote.php/dav/.
|
||||
if let Some(resp) = enforce_native_lock(
|
||||
&state.webdav_lock_store,
|
||||
if_header.as_deref(),
|
||||
&internal_path,
|
||||
Some(&file.etag),
|
||||
) {
|
||||
return Ok(resp);
|
||||
}
|
||||
|
||||
// ── RFC 7232 conditional preconditions ────────────────────────────
|
||||
let current_etag = Some(file.etag.as_str());
|
||||
@@ -1080,6 +1087,10 @@ async fn handle_patch(
|
||||
.map_err(AppError::from)?,
|
||||
)
|
||||
};
|
||||
let suffix_len = match end {
|
||||
Some(end) if end + 1 < file.size => file.size - (end + 1),
|
||||
_ => 0,
|
||||
};
|
||||
let suffix_stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>> = match end
|
||||
{
|
||||
Some(end) if end + 1 < file.size => Box::into_pin(
|
||||
@@ -1090,19 +1101,59 @@ async fn handle_patch(
|
||||
),
|
||||
_ => Box::pin(stream::empty()),
|
||||
};
|
||||
|
||||
let filename = filename_from_path(subpath).to_string();
|
||||
let ingested = ingest_range_patch_to_cas(
|
||||
prefix_stream,
|
||||
(prefix_stream, start),
|
||||
req.into_body(),
|
||||
suffix_stream,
|
||||
(suffix_stream, suffix_len),
|
||||
&state.core.dedup_service,
|
||||
&filename,
|
||||
&claimed_type,
|
||||
max_upload,
|
||||
PatchIngestBudget {
|
||||
max_bytes: max_upload,
|
||||
expected_body_len: end.map(|end| end - start + 1),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
// ── Quota enforcement ─────────────────────────────────────────────
|
||||
if let Some(storage_svc) = state.storage_usage_service.as_ref()
|
||||
&& let Err(err) = storage_svc
|
||||
.check_storage_quota(session.user.id, ingested.size)
|
||||
.await
|
||||
{
|
||||
discard_ingested(&state.core.dedup_service, &ingested).await;
|
||||
tracing::warn!(
|
||||
"⛔ NC WEBDAV PATCH REJECTED (quota): user={}, file={}, size={}",
|
||||
session.user.id,
|
||||
internal_path,
|
||||
ingested.size
|
||||
);
|
||||
return Err(AppError::new(
|
||||
StatusCode::INSUFFICIENT_STORAGE,
|
||||
err.message,
|
||||
"QuotaExceeded",
|
||||
));
|
||||
}
|
||||
|
||||
// ── 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 ──────────────────────────────────────────────────
|
||||
let new_size = ingested.size;
|
||||
let content_type = ingested.content_type.clone();
|
||||
|
||||
@@ -16,7 +16,7 @@ use std::path::{Path, PathBuf};
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex as StdMutex;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
|
||||
use axum::body::Body;
|
||||
use bytes::Bytes;
|
||||
@@ -81,6 +81,24 @@ pub async fn discard_ingested(dedup: &DedupService, blob: &IngestedBlob) {
|
||||
/// ingest pass (REST chunked uploads) — no post-store re-read needed.
|
||||
pub type ChecksumTee = Arc<StdMutex<Option<IncrementalHasher>>>;
|
||||
|
||||
/// A byte-range stream paired with its known length, used by
|
||||
/// [`ingest_range_patch_to_cas`] for the untouched prefix/suffix either
|
||||
/// side of a PATCH edit.
|
||||
pub type RangeSegment = (
|
||||
Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
|
||||
u64,
|
||||
);
|
||||
|
||||
/// Size cap and expected-length validation for [`ingest_range_patch_to_cas`].
|
||||
pub struct PatchIngestBudget {
|
||||
/// Caps the size of the *edit* (the request body), not the whole
|
||||
/// spliced file — see the field's use in [`ingest_range_patch_to_cas`].
|
||||
pub max_bytes: usize,
|
||||
/// Declared `X-Update-Range` span, `None` for `append`. Validated
|
||||
/// against the actual streamed body length, not `Content-Length`.
|
||||
pub expected_body_len: Option<u64>,
|
||||
}
|
||||
|
||||
/// Create a checksum tee for [`ingest_stream_to_cas`].
|
||||
pub fn checksum_tee(alg: ChecksumAlg) -> ChecksumTee {
|
||||
Arc::new(StdMutex::new(Some(IncrementalHasher::new(alg))))
|
||||
@@ -260,24 +278,83 @@ pub async fn ingest_body_to_cas(
|
||||
/// Because FastCDC chunking is content-defined rather than offset-defined,
|
||||
/// unedited chunks on either side of the edit typically dedup for free.
|
||||
///
|
||||
/// Each of `prefix`/`suffix` is paired with its known byte length (from the
|
||||
/// file's size and the requested range — callers compute it, not this
|
||||
/// function). `budget.max_bytes` bounds the size of the *edit* (the request
|
||||
/// body) — it is widened by the prefix/suffix lengths before being applied
|
||||
/// to the combined stream, so that already-stored, untouched bytes being
|
||||
/// re-ingested unchanged don't count against the cap. Without this, any
|
||||
/// PATCH against a file at or above `max_bytes` would be rejected
|
||||
/// regardless of how small the edit itself is.
|
||||
///
|
||||
/// `budget.expected_body_len`, when `Some`, is validated against the
|
||||
/// *actual* number of body bytes streamed — not the client-supplied
|
||||
/// `Content-Length` header, which chunked-transfer-encoded requests may
|
||||
/// omit entirely. Counting the real bytes means a request that declares
|
||||
/// `X-Update-Range: bytes=5-9` (a 5-byte span) but streams a
|
||||
/// differently-sized body is caught regardless of whether `Content-Length`
|
||||
/// was present. On mismatch the already-ingested blob is discarded and
|
||||
/// never reaches the atomic store, so a rejected PATCH can't leave stray
|
||||
/// unreferenced content.
|
||||
///
|
||||
/// [RFC 5789]: https://www.rfc-editor.org/rfc/rfc5789
|
||||
pub async fn ingest_range_patch_to_cas(
|
||||
prefix: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
|
||||
prefix: RangeSegment,
|
||||
body: Body,
|
||||
suffix: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
|
||||
suffix: RangeSegment,
|
||||
dedup: &Arc<DedupService>,
|
||||
filename: &str,
|
||||
claimed_type: &str,
|
||||
max_bytes: usize,
|
||||
budget: PatchIngestBudget,
|
||||
) -> Result<IngestedBlob, AppError> {
|
||||
let body_stream = BodyStream::new(body).filter_map(|item| async move {
|
||||
match item {
|
||||
Ok(frame) => frame.into_data().ok().map(Ok),
|
||||
Err(e) => Some(Err(std::io::Error::other(e.to_string()))),
|
||||
let PatchIngestBudget {
|
||||
max_bytes,
|
||||
expected_body_len,
|
||||
} = budget;
|
||||
let (prefix_stream, prefix_len) = prefix;
|
||||
let (suffix_stream, suffix_len) = suffix;
|
||||
let body_len = Arc::new(AtomicU64::new(0));
|
||||
let counter = body_len.clone();
|
||||
let body_stream = BodyStream::new(body).filter_map(move |item| {
|
||||
let counter = counter.clone();
|
||||
async move {
|
||||
match item {
|
||||
Ok(frame) => {
|
||||
let bytes = frame.into_data().ok()?;
|
||||
counter.fetch_add(bytes.len() as u64, Ordering::Relaxed);
|
||||
Some(Ok(bytes))
|
||||
}
|
||||
Err(e) => Some(Err(std::io::Error::other(e.to_string()))),
|
||||
}
|
||||
}
|
||||
});
|
||||
let combined = prefix.chain(body_stream).chain(suffix);
|
||||
ingest_stream_to_cas(combined, dedup, filename, claimed_type, max_bytes, None).await
|
||||
let effective_max = max_bytes.saturating_add((prefix_len + suffix_len) as usize);
|
||||
let combined = prefix_stream.chain(body_stream).chain(suffix_stream);
|
||||
let ingested =
|
||||
ingest_stream_to_cas(combined, dedup, filename, claimed_type, effective_max, None)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
if e.error_type == "PayloadTooLarge" {
|
||||
AppError::payload_too_large(format!(
|
||||
"PATCH body exceeds the direct-PATCH edit-size cap ({max_bytes} bytes). \
|
||||
Use the chunked-upload protocol for edits larger than this."
|
||||
))
|
||||
} else {
|
||||
e
|
||||
}
|
||||
})?;
|
||||
|
||||
if let Some(expected) = expected_body_len {
|
||||
let actual = body_len.load(Ordering::Relaxed);
|
||||
if actual != expected {
|
||||
discard_ingested(dedup, &ingested).await;
|
||||
return Err(AppError::bad_request(format!(
|
||||
"PATCH body ({actual} bytes) does not match the X-Update-Range span ({expected} bytes)"
|
||||
)));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(ingested)
|
||||
}
|
||||
|
||||
/// Adapt a multipart field into a byte stream for [`ingest_stream_to_cas`].
|
||||
|
||||
Reference in New Issue
Block a user