mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-04 04:17:44 +00:00
Zhengchao An
b2a376c2d2
* fix(admin): bound IAM import archive expansion MAX_IAM_IMPORT_SIZE caps the compressed upload at 10 MB, but every member of the archive was then read with read_to_end into an unbounded Vec. Deflate ratios well above 100:1 are easy to construct, so a small authorized upload could expand without limit across the seven members ImportIam reads. Add a shared expansion budget (MAX_IAM_IMPORT_EXPANDED_SIZE, 10x the compressed cap) drawn down by every member, and route all seven reads through one helper that reads a byte past the remaining budget to detect overrun. Sharing the budget bounds the archive as a whole rather than letting each member spend the full limit independently. Covers R03-CAN-024 through R03-CAN-030 plus R04-CAN-077 (backlog #1471) — one fix rather than seven, since all seven call sites were byte-identical. * fix(kms): confine local key paths and refuse silent key replacement Local KMS key identifiers arrive from request input — the `name` tag on CreateKey, the `keyId` body field or query parameter on DeleteKey — and were joined onto `key_dir` with no validation. An identifier such as `../../tmp/evil` escaped the configured directory, making key creation a constrained arbitrary-file write and `DeleteKey` with `force_immediate` a cross-directory delete. Validate in `master_key_path` and make it fallible, so every filesystem path in this backend inherits the guard: decode_stored_key, load_master_key, save_master_key, create_key and delete_key all derive their paths there. The rule is containment rather than a character allowlist, so identifiers already in use keep resolving; only separators, NUL, absolute paths and non-single-component forms are refused. Note `.` and `..` are contained rather than refused — the `.key` suffix turns them into the ordinary filenames `..key` and `...key`. Separately, `LocalKmsBackend::create_key` had no existence check, while the sibling `KmsClient::create_key` has always had one. Since `save_master_key` renames over its destination, creating a key under an existing name silently replaced its material and destroyed the ability to decrypt everything wrapped under it — and the backend path is the one the admin API uses. It now returns KeyAlreadyExists, matching StaticKmsBackend. Covers R03-CAN-072, R03-CAN-073 and R07-CAN-103 (backlog #1475). R03-CAN-073 needed no separate change: delete_key routes both its load and its remove_file through master_key_path. * fix(swift): bound SLO manifest reads to the 2 MiB manifest limit The three Swift SLO handlers that load a stored manifest (handle_slo_get, handle_slo_get_manifest, handle_slo_delete) read the `<object>.slo-manifest` object to EOF with AsyncReadExt::read_to_end. That key is predictable and writable through the ordinary object PUT path, so a tenant can replace the manifest with an arbitrarily large object and then make the server allocate its full size on every SLO GET, multipart-manifest=get, or multipart-manifest=delete request - a memory amplification bounded only by the stored object size (CWE-400 / CWE-770). The 2 MiB manifest limit that handle_slo_put enforces was not applied on the read side. Introduce MAX_SLO_MANIFEST_SIZE (the existing 2 MiB PUT limit, now a named constant) and a shared read_manifest_bytes helper that reads through a `take(limit + 1)` and rejects anything larger, so an oversized manifest is refused instead of being buffered first. All three call sites go through the helper. handle_slo_put now checks the size before parsing the JSON. Regression tests: test_read_manifest_bytes_rejects_oversized_manifest and test_read_manifest_bytes_stops_reading_oversized_manifest (which asserts the reader is not consumed past the limit), plus a boundary test that a manifest at exactly 2 MiB is still accepted. * fix(protocols): authorize every object in FTPS/WebDAV recursive deletes The FTPS and WebDAV gateways authorized only the container before a recursive delete and then destroyed everything inside it without a further check: - FTPS RMD (and DELE on a bucket path ending in '/') cleared s3:DeleteBucket, then delete_bucket_recursively listed the bucket and deleted every object. - WebDAV DELETE on a bucket did the same via its own delete_bucket_recursively. - WebDAV DELETE on a directory cleared s3:DeleteObject for the directory marker key ("dir/") only, then listed that prefix and deleted every child under it. A principal holding s3:DeleteBucket (or s3:DeleteObject on a single marker key) could therefore erase objects it had no s3:DeleteObject permission for, and the operation reported success. Deletion stays recursive - that is the expected behaviour for these protocols - but each object now clears s3:DeleteObject on its own key before it is removed, and the enumeration clears s3:ListBucket. A denial aborts the whole operation with access denied rather than being skipped, so the caller can never be told the delete succeeded while objects were left behind or removed without authorization. The test double gained shared-state cloning, delete_object/delete_bucket call logs, and list/delete queue helpers so the regression tests can observe that nothing is deleted once a deny lands. * fix(server,ecstore): bound TLS handshakes and remote volume RPC waits Three call sites let an unauthenticated client or a misbehaving peer hold server resources with no deadline. TLS listener (R03-CAN-035): process_connection awaited `acceptor.accept(socket)` with no bound. A client that opens a TCP connection and never finishes the handshake parks a Tokio task and a socket forever, and the connection cap (RUSTFS_API_MAX_CONNECTIONS) is unlimited by default, so nothing else sheds it. The handshake now runs under accept_tls_with_deadline(), reusing the existing HTTP/1 header-read budget — the established slow-client bound for the pre-request phase — and the expiry is recorded through the same log/metric path as a handshake error, under a new TIMEOUT failure kind. Remote disk RPCs (R03-CAN-049, R03-CAN-050): list_volumes and delete_volume passed Duration::ZERO, which execute_with_timeout treats as "no deadline", so a peer that accepts the request and never answers stalls the coordinator (and, for delete_volume, the bucket-deletion workflow). Both now pass get_max_timeout_duration(), matching every sibling method in the file. Regression tests: a silent TLS peer must be shed by the handshake deadline; list_volumes/delete_volume against a peer that completes the TCP connect and then goes silent must fail with DiskError::Timeout instead of hanging. * fix(security): stop leaking signed headers and bound OIDC/KMS credentials Three independent hygiene fixes found by the security review. R03-CAN-018 (crates/signer): try_get_canonical_headers and get_signed_headers logged the complete header map at DEBUG before signing. Runtime callers pass session credentials and SSE-C key material through these headers, so anyone able to raise the log level (or read DEBUG logs) recovered X-Amz-Security-Token and SSE-C keys verbatim. The statements were debugging leftovers with no operational value and are deleted rather than redacted. R03-CAN-014 (crates/iam): the OIDC HTTP adapter buffered provider responses with an unbounded Response::bytes(), so a configured, compromised or attacker-pointed IdP endpoint could stream an arbitrarily large or endless body into memory (the ValidateOidcConfig admin handler lets a ServerInfo caller choose the endpoint). Responses are now read incrementally and fail closed past MAX_OIDC_RESPONSE_SIZE, and the already SSRF-hardened client builder gains request and connect timeouts so a stalled provider cannot pin the calling task indefinitely. R07-CAN-105 (helm): the Vault KMS token was serialized into the chart ConfigMap, exposing it to every subject allowed to get ConfigMaps in the namespace. It now renders into a dedicated Secret that the Deployment and StatefulSet consume via envFrom; the Secret is separate from the main credentials Secret so it also works when secret.existingSecret is set. Regression tests: - rustfs-signer: signing_never_logs_signed_header_material - rustfs-iam: oidc_response_body_past_the_limit_is_rejected, oidc_response_body_at_the_limit_is_accepted - scripts/test_helm_templates.sh: KMS token must never render in plaintext * fix(webdav): enforce body limit, request timeout and connection cap The configured WebDAV maximum body size was enforced from Content-Length, so a chunked request declared no length and bypassed it entirely. The configured request timeout was never applied to the connection at all, and the accept loop spawned a task per connection with no bound, so an unauthenticated client could hold resources indefinitely and in unbounded number. Enforce the limit on bytes actually read rather than the declared length, apply the configured timeout to the request, and bound accepted connections with a new RUSTFS_WEBDAV_MAX_CONNECTIONS (default 1024) surfaced in the config report. Covers R03-CAN-051, R03-CAN-052, R03-CAN-067, R04-CAN-089, R05-CAN-094 and R05-CAN-097 (backlog #1471, #1474). * fix(security): stop STS credentials from crossing the parent trust boundary Two related credential-boundary holes let a short-lived STS credential act with the full, unrestricted authority of the long-term user it was minted from. AddUser (R03-CAN-021, CWE-269/863): should_check_deny_only relaxes the admin policy check to deny-only when a Console/STS session targets the IAM user it represents. Nothing then stopped that session from calling AddUser with its own parent's access key, so the handler wrote an attacker-chosen secret key and status over the parent's stored Credentials via create_user -> save_user_identity. A session that expires in minutes became permanent control of the account. AddUser now rejects any temp or service-account requester whose resolved parent equals the target access key, resolving the parent the same way should_check_deny_only does (parent_user field, else the JWT `parent` claim, since some stores persist the parent only in the token). FTPS/SFTP/WebDAV password auth (R04-CAN-086, CWE-287/862): these protocols looked the access key up with check_key, which falls back to the STS account cache, and then compared only the stored secret. An STS access key plus secret therefore authenticated with no session token presented and no session-policy claims applied - the holder got the parent's full permissions. Password authentication now rejects temporary credentials before the secret comparison. The discriminator is is_temp() && !is_service_account(), the same one IamCache::update_user_with_claims uses to route an identity into the STS cache, so service accounts - which resolve policy from stored IAM state rather than a client-presented token - keep working over these protocols. Regression tests cover both predicates and pin the guards to their call sites so neither can be dropped without a test failure.
2400 lines
88 KiB
Rust
2400 lines
88 KiB
Rust
// Copyright 2024 RustFS Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
use crate::common::client::s3::StorageBackend as S3StorageBackend;
|
|
use crate::common::gateway::{S3Action, authorize_operation};
|
|
use crate::common::session::SessionContext;
|
|
use bytes::Bytes;
|
|
use dav_server::davpath::DavPath;
|
|
use dav_server::fs::{
|
|
DavDirEntry, DavFile, DavFileSystem, DavMetaData, FsError, FsFuture, FsResult, FsStream, OpenOptions, ReadDirMeta,
|
|
};
|
|
use futures_util::{FutureExt, StreamExt, stream};
|
|
use percent_encoding::percent_decode_str;
|
|
use rustfs_utils::MaskedAccessKey;
|
|
use rustfs_utils::path;
|
|
use s3s::dto::*;
|
|
use std::fmt::Debug;
|
|
use std::io::SeekFrom;
|
|
use std::sync::Arc;
|
|
use std::time::SystemTime;
|
|
use tokio::sync::RwLock;
|
|
use tracing::{debug, error};
|
|
|
|
const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
|
|
const LOG_SUBSYSTEM_WEBDAV_DRIVER: &str = "webdav_driver";
|
|
const EVENT_WEBDAV_OBJECT_METADATA_FAILED: &str = "webdav_object_metadata_failed";
|
|
const EVENT_WEBDAV_STREAM_READ_FAILED: &str = "webdav_stream_read_failed";
|
|
const EVENT_WEBDAV_OBJECT_READ_FAILED: &str = "webdav_object_read_failed";
|
|
const EVENT_WEBDAV_OBJECT_WRITE_STATE: &str = "webdav_object_write_state";
|
|
const EVENT_WEBDAV_LIST_FAILED: &str = "webdav_list_failed";
|
|
const EVENT_WEBDAV_COPY_FAILED: &str = "webdav_copy_failed";
|
|
const EVENT_WEBDAV_DELETE_FAILED: &str = "webdav_delete_failed";
|
|
const EVENT_WEBDAV_PROBE_FAILED: &str = "webdav_probe_failed";
|
|
const EVENT_WEBDAV_BUCKET_LIST_FAILED: &str = "webdav_bucket_list_failed";
|
|
const EVENT_WEBDAV_BUCKET_METADATA_STATE: &str = "webdav_bucket_metadata_state";
|
|
const EVENT_WEBDAV_DIRECTORY_STATE: &str = "webdav_directory_state";
|
|
const EVENT_WEBDAV_OBJECT_DELETE_STATE: &str = "webdav_object_delete_state";
|
|
const EVENT_WEBDAV_RENAME_STATE: &str = "webdav_rename_state";
|
|
|
|
/// Convert s3s ETag enum to string
|
|
fn etag_to_string(etag: &ETag) -> String {
|
|
match etag {
|
|
ETag::Strong(s) => s.clone(),
|
|
ETag::Weak(s) => s.clone(),
|
|
}
|
|
}
|
|
|
|
/// WebDAV metadata implementation
|
|
#[derive(Debug, Clone)]
|
|
pub struct WebDavMetaData {
|
|
/// File size in bytes
|
|
pub size: u64,
|
|
/// Modification time
|
|
pub modified: SystemTime,
|
|
/// Creation time
|
|
pub created: SystemTime,
|
|
/// Whether this is a directory
|
|
pub is_dir: bool,
|
|
/// ETag (optional)
|
|
pub etag: Option<String>,
|
|
/// Content type (optional)
|
|
pub content_type: Option<String>,
|
|
}
|
|
|
|
impl DavMetaData for WebDavMetaData {
|
|
fn len(&self) -> u64 {
|
|
self.size
|
|
}
|
|
|
|
fn modified(&self) -> FsResult<SystemTime> {
|
|
Ok(self.modified)
|
|
}
|
|
|
|
fn is_dir(&self) -> bool {
|
|
self.is_dir
|
|
}
|
|
|
|
fn created(&self) -> FsResult<SystemTime> {
|
|
Ok(self.created)
|
|
}
|
|
|
|
fn etag(&self) -> Option<String> {
|
|
self.etag.clone()
|
|
}
|
|
}
|
|
|
|
/// WebDAV directory entry implementation
|
|
#[derive(Debug, Clone)]
|
|
pub struct WebDavDirEntry {
|
|
/// Entry name
|
|
pub name: String,
|
|
/// Entry metadata
|
|
pub metadata: WebDavMetaData,
|
|
}
|
|
|
|
impl DavDirEntry for WebDavDirEntry {
|
|
fn name(&self) -> Vec<u8> {
|
|
self.name.as_bytes().to_vec()
|
|
}
|
|
|
|
fn metadata(&self) -> FsFuture<'_, Box<dyn DavMetaData>> {
|
|
let meta = self.metadata.clone();
|
|
async move { Ok(Box::new(meta) as Box<dyn DavMetaData>) }.boxed()
|
|
}
|
|
}
|
|
|
|
/// WebDAV file implementation for reading/writing
|
|
pub struct WebDavFile<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
/// Storage backend
|
|
storage: S,
|
|
/// Session context for authorization
|
|
session_context: Arc<SessionContext>,
|
|
/// Bucket name
|
|
bucket: String,
|
|
/// Object key
|
|
key: String,
|
|
/// Current position in file (using RwLock for interior mutability in async)
|
|
position: Arc<RwLock<u64>>,
|
|
/// File size (known after metadata fetch)
|
|
size: Option<u64>,
|
|
/// Write buffer for accumulating data before upload
|
|
write_buffer: Arc<RwLock<Vec<u8>>>,
|
|
/// Whether we're in write mode
|
|
is_write: bool,
|
|
/// Maximum body size for chunked transfers
|
|
max_body_size: u64,
|
|
}
|
|
|
|
impl<S> WebDavFile<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
/// Default maximum body size (5GB)
|
|
pub const DEFAULT_MAX_BODY_SIZE: u64 = 5 * 1024 * 1024 * 1024;
|
|
|
|
pub fn new(storage: S, session_context: Arc<SessionContext>, bucket: String, key: String, is_write: bool) -> Self {
|
|
Self::with_max_body_size(storage, session_context, bucket, key, is_write, Self::DEFAULT_MAX_BODY_SIZE)
|
|
}
|
|
|
|
pub fn with_max_body_size(
|
|
storage: S,
|
|
session_context: Arc<SessionContext>,
|
|
bucket: String,
|
|
key: String,
|
|
is_write: bool,
|
|
max_body_size: u64,
|
|
) -> Self {
|
|
Self {
|
|
storage,
|
|
session_context,
|
|
bucket,
|
|
key,
|
|
position: Arc::new(RwLock::new(0)),
|
|
size: None,
|
|
write_buffer: Arc::new(RwLock::new(Vec::new())),
|
|
is_write,
|
|
max_body_size,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<S> Debug for WebDavFile<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("WebDavFile")
|
|
.field("bucket", &self.bucket)
|
|
.field("key", &self.key)
|
|
.field("position", &"<locked>")
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl<S> DavFile for WebDavFile<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
fn metadata(&mut self) -> FsFuture<'_, Box<dyn DavMetaData>> {
|
|
let storage = self.storage.clone();
|
|
let session_context = self.session_context.clone();
|
|
let bucket = self.bucket.clone();
|
|
let key = self.key.clone();
|
|
|
|
async move {
|
|
match storage
|
|
.head_object(
|
|
&bucket,
|
|
&key,
|
|
&session_context.principal.user_identity.credentials.access_key,
|
|
&session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(output) => {
|
|
let size = output.content_length.unwrap_or(0) as u64;
|
|
let modified = output
|
|
.last_modified
|
|
.map(|dt| {
|
|
let offset_dt: time::OffsetDateTime = dt.into();
|
|
SystemTime::from(offset_dt)
|
|
})
|
|
.unwrap_or_else(SystemTime::now);
|
|
|
|
Ok(Box::new(WebDavMetaData {
|
|
size,
|
|
modified,
|
|
created: modified,
|
|
is_dir: false,
|
|
etag: output.e_tag.as_ref().map(etag_to_string),
|
|
content_type: output.content_type.map(|c| c.to_string()),
|
|
}) as Box<dyn DavMetaData>)
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_OBJECT_METADATA_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
object = %key,
|
|
error = %e,
|
|
"webdav object metadata failed"
|
|
);
|
|
Err(FsError::NotFound)
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn write_buf(&mut self, mut buf: Box<dyn bytes::Buf + Send>) -> FsFuture<'_, ()> {
|
|
let write_buffer = self.write_buffer.clone();
|
|
let max_body_size = self.max_body_size;
|
|
async move {
|
|
let mut buffer = write_buffer.write().await;
|
|
// Consume all chunks from the buffer, not just the first one
|
|
while buf.has_remaining() {
|
|
let chunk = buf.chunk();
|
|
// Check size limit before extending
|
|
if buffer.len() as u64 + chunk.len() as u64 > max_body_size {
|
|
return Err(FsError::TooLarge);
|
|
}
|
|
buffer.extend_from_slice(chunk);
|
|
buf.advance(chunk.len());
|
|
}
|
|
Ok(())
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn write_bytes(&mut self, buf: Bytes) -> FsFuture<'_, ()> {
|
|
let write_buffer = self.write_buffer.clone();
|
|
let max_body_size = self.max_body_size;
|
|
async move {
|
|
let mut buffer = write_buffer.write().await;
|
|
// Check size limit before extending
|
|
if buffer.len() as u64 + buf.len() as u64 > max_body_size {
|
|
return Err(FsError::TooLarge);
|
|
}
|
|
buffer.extend_from_slice(&buf);
|
|
Ok(())
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn read_bytes(&mut self, count: usize) -> FsFuture<'_, Bytes> {
|
|
let storage = self.storage.clone();
|
|
let session_context = self.session_context.clone();
|
|
let bucket = self.bucket.clone();
|
|
let key = self.key.clone();
|
|
let position = self.position.clone();
|
|
|
|
async move {
|
|
let start_pos = *position.read().await;
|
|
match storage
|
|
.get_object_range(
|
|
&bucket,
|
|
&key,
|
|
&session_context.principal.user_identity.credentials.access_key,
|
|
&session_context.principal.user_identity.credentials.secret_key,
|
|
start_pos,
|
|
count as u64,
|
|
)
|
|
.await
|
|
{
|
|
Ok(output) => {
|
|
if let Some(body) = output.body {
|
|
let mut data = Vec::new();
|
|
let mut stream = body;
|
|
while let Some(chunk_result) = stream.next().await {
|
|
match chunk_result {
|
|
Ok(bytes) => data.extend_from_slice(&bytes),
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_STREAM_READ_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
object = %key,
|
|
error = %e,
|
|
"webdav stream read failed"
|
|
);
|
|
return Err(FsError::GeneralFailure);
|
|
}
|
|
}
|
|
}
|
|
// Update position after successful read
|
|
let bytes_read = data.len() as u64;
|
|
*position.write().await = start_pos + bytes_read;
|
|
Ok(Bytes::from(data))
|
|
} else {
|
|
Ok(Bytes::new())
|
|
}
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_OBJECT_READ_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
object = %key,
|
|
start_pos,
|
|
read_len = count,
|
|
error = %e,
|
|
"webdav object read failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn seek(&mut self, pos: SeekFrom) -> FsFuture<'_, u64> {
|
|
let position = self.position.clone();
|
|
let size = self.size;
|
|
|
|
async move {
|
|
let current_pos = *position.read().await;
|
|
let new_pos = match pos {
|
|
SeekFrom::Start(offset) => offset,
|
|
SeekFrom::End(offset) => {
|
|
let file_size = size.unwrap_or(0);
|
|
if offset < 0 {
|
|
file_size.saturating_sub((-offset) as u64)
|
|
} else {
|
|
file_size + offset as u64
|
|
}
|
|
}
|
|
SeekFrom::Current(offset) => {
|
|
if offset < 0 {
|
|
current_pos.saturating_sub((-offset) as u64)
|
|
} else {
|
|
current_pos + offset as u64
|
|
}
|
|
}
|
|
};
|
|
// Persist the new position
|
|
*position.write().await = new_pos;
|
|
Ok(new_pos)
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn flush(&mut self) -> FsFuture<'_, ()> {
|
|
let storage = self.storage.clone();
|
|
let session_context = self.session_context.clone();
|
|
let bucket = self.bucket.clone();
|
|
let key = self.key.clone();
|
|
let write_buffer = self.write_buffer.clone();
|
|
let is_write = self.is_write;
|
|
|
|
async move {
|
|
if !is_write {
|
|
return Ok(());
|
|
}
|
|
|
|
// Use write lock and std::mem::take to avoid cloning the buffer
|
|
let mut buffer = write_buffer.write().await;
|
|
let file_size = buffer.len();
|
|
let data_bytes = Bytes::from(std::mem::take(&mut *buffer));
|
|
drop(buffer);
|
|
|
|
let stream = stream::once(async move { Ok::<Bytes, std::io::Error>(data_bytes) });
|
|
let streaming_blob = StreamingBlob::wrap(stream);
|
|
|
|
let put_input = PutObjectInput::builder()
|
|
.bucket(bucket.clone())
|
|
.key(key.clone())
|
|
.content_length(Some(file_size as i64))
|
|
.body(Some(streaming_blob))
|
|
.build()
|
|
.map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
match storage
|
|
.put_object(
|
|
put_input,
|
|
&session_context.principal.user_identity.credentials.access_key,
|
|
&session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
debug!(
|
|
event = EVENT_WEBDAV_OBJECT_WRITE_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "flushed",
|
|
bucket = %bucket,
|
|
object = %key,
|
|
file_size,
|
|
"WebDAV object flush completed"
|
|
);
|
|
// Buffer already cleared by std::mem::take above
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_OBJECT_WRITE_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "flush_failed",
|
|
bucket = %bucket,
|
|
object = %key,
|
|
file_size,
|
|
error = %e,
|
|
"WebDAV object flush failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
}
|
|
|
|
/// WebDAV filesystem driver implementation
|
|
pub struct WebDavDriver<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
/// Storage backend for S3 operations
|
|
storage: S,
|
|
/// Session context for authorization
|
|
session_context: Arc<SessionContext>,
|
|
}
|
|
|
|
enum ResolvedPath {
|
|
File(Box<HeadObjectOutput>),
|
|
Directory {
|
|
prefix: String,
|
|
metadata: Option<Box<HeadObjectOutput>>,
|
|
},
|
|
}
|
|
|
|
enum HeadObjectProbe {
|
|
Forbidden,
|
|
Missing,
|
|
Found(Box<HeadObjectOutput>),
|
|
}
|
|
|
|
impl<S> Debug for WebDavDriver<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("WebDavDriver").field("storage", &"StorageBackend").finish()
|
|
}
|
|
}
|
|
|
|
impl<S> Clone for WebDavDriver<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
storage: self.storage.clone(),
|
|
session_context: self.session_context.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<S> WebDavDriver<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
/// Create a new WebDAV driver with the given storage backend and session context
|
|
pub fn new(storage: S, session_context: Arc<SessionContext>) -> Self {
|
|
Self {
|
|
storage,
|
|
session_context,
|
|
}
|
|
}
|
|
|
|
fn credentials(&self) -> (&str, &str) {
|
|
(
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
}
|
|
|
|
fn is_missing_head_object_error(error: &str) -> bool {
|
|
let lower = error.to_ascii_lowercase();
|
|
lower.contains("nosuchkey")
|
|
|| lower.contains("notfound")
|
|
|| lower.contains("not found")
|
|
|| lower.contains("status code: 404")
|
|
}
|
|
|
|
async fn prefix_has_entries(&self, bucket: &str, prefix: &str) -> FsResult<bool> {
|
|
let (access_key, secret_key) = self.credentials();
|
|
let list_input = ListObjectsV2Input::builder()
|
|
.bucket(bucket.to_string())
|
|
.prefix(Some(prefix.to_string()))
|
|
.max_keys(Some(1))
|
|
.build()
|
|
.map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
let output = self
|
|
.storage
|
|
.list_objects_v2(list_input, access_key, secret_key)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_LIST_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
prefix = %prefix,
|
|
error = %e,
|
|
"webdav list failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
Ok(output.contents.map(|c| !c.is_empty()).unwrap_or(false)
|
|
|| output.common_prefixes.map(|c| !c.is_empty()).unwrap_or(false))
|
|
}
|
|
|
|
async fn copy_object_streaming(&self, src_bucket: &str, src_key: &str, dst_bucket: &str, dst_key: &str) -> FsResult<()> {
|
|
let (access_key, secret_key) = self.credentials();
|
|
let get_output = self
|
|
.storage
|
|
.get_object(src_bucket, src_key, access_key, secret_key, None)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_COPY_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "source_read_failed",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
error = %e,
|
|
"webdav copy failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
let GetObjectOutput {
|
|
body,
|
|
content_length,
|
|
content_type,
|
|
..
|
|
} = get_output;
|
|
let body = body.ok_or_else(|| {
|
|
error!(
|
|
event = EVENT_WEBDAV_COPY_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "source_body_missing",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
"webdav copy failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
let mut put_builder = PutObjectInput::builder()
|
|
.bucket(dst_bucket.to_string())
|
|
.key(dst_key.to_string())
|
|
.body(Some(body));
|
|
|
|
if let Some(content_length) = content_length {
|
|
put_builder = put_builder.content_length(Some(content_length));
|
|
}
|
|
|
|
if let Some(content_type) = content_type {
|
|
put_builder = put_builder.content_type(Some(content_type));
|
|
}
|
|
|
|
let put_input = put_builder.build().map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
self.storage
|
|
.put_object(put_input, access_key, secret_key)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_COPY_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "destination_write_failed",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
error = %e,
|
|
"webdav copy failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn execute_directory_rename_pairs(
|
|
&self,
|
|
src_bucket: &str,
|
|
dst_bucket: &str,
|
|
rename_pairs: &[(String, String)],
|
|
) -> FsResult<()> {
|
|
let (access_key, secret_key) = self.credentials();
|
|
|
|
for (src_obj_key, dst_obj_key) in rename_pairs {
|
|
self.copy_object_streaming(src_bucket, src_obj_key, dst_bucket, dst_obj_key)
|
|
.await?;
|
|
}
|
|
|
|
for (src_obj_key, _) in rename_pairs {
|
|
self.storage
|
|
.delete_object(src_bucket, src_obj_key, access_key, secret_key)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_DELETE_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "rename_cleanup_failed",
|
|
bucket = %src_bucket,
|
|
object = %src_obj_key,
|
|
error = %e,
|
|
"webdav delete failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn probe_head_object(&self, bucket: &str, key: &str) -> FsResult<HeadObjectProbe> {
|
|
let (access_key, secret_key) = self.credentials();
|
|
|
|
if authorize_operation(&self.session_context, &S3Action::HeadObject, bucket, Some(key))
|
|
.await
|
|
.is_err()
|
|
{
|
|
return Ok(HeadObjectProbe::Forbidden);
|
|
}
|
|
|
|
match self.storage.head_object(bucket, key, access_key, secret_key).await {
|
|
Ok(output) => Ok(HeadObjectProbe::Found(Box::new(output))),
|
|
Err(e) => {
|
|
let err_msg = e.to_string();
|
|
if Self::is_missing_head_object_error(&err_msg) {
|
|
Ok(HeadObjectProbe::Missing)
|
|
} else {
|
|
error!(
|
|
event = EVENT_WEBDAV_PROBE_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
object = %key,
|
|
error = %err_msg,
|
|
"webdav probe failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn resolve_path(&self, bucket: &str, key: &str) -> FsResult<ResolvedPath> {
|
|
let prefix = format!("{}/", key);
|
|
let mut had_visibility = false;
|
|
|
|
match self.probe_head_object(bucket, key).await? {
|
|
HeadObjectProbe::Found(output) => {
|
|
let size = output.content_length.unwrap_or(0) as u64;
|
|
let is_dir_marker = output.content_type.as_deref() == Some("application/x-directory");
|
|
|
|
if is_dir_marker {
|
|
return Ok(ResolvedPath::Directory {
|
|
prefix,
|
|
metadata: Some(output),
|
|
});
|
|
}
|
|
|
|
if size == 0
|
|
&& authorize_operation(&self.session_context, &S3Action::ListBucket, bucket, Some(&prefix))
|
|
.await
|
|
.is_ok()
|
|
&& self.prefix_has_entries(bucket, &prefix).await?
|
|
{
|
|
return Ok(ResolvedPath::Directory {
|
|
prefix,
|
|
metadata: Some(output),
|
|
});
|
|
}
|
|
|
|
return Ok(ResolvedPath::File(output));
|
|
}
|
|
HeadObjectProbe::Missing => {
|
|
had_visibility = true;
|
|
}
|
|
HeadObjectProbe::Forbidden => {}
|
|
}
|
|
|
|
match self.probe_head_object(bucket, &prefix).await? {
|
|
HeadObjectProbe::Found(output) => {
|
|
return Ok(ResolvedPath::Directory {
|
|
prefix,
|
|
metadata: Some(output),
|
|
});
|
|
}
|
|
HeadObjectProbe::Missing => {
|
|
had_visibility = true;
|
|
}
|
|
HeadObjectProbe::Forbidden => {}
|
|
}
|
|
|
|
if authorize_operation(&self.session_context, &S3Action::ListBucket, bucket, Some(&prefix))
|
|
.await
|
|
.is_ok()
|
|
{
|
|
had_visibility = true;
|
|
if self.prefix_has_entries(bucket, &prefix).await? {
|
|
return Ok(ResolvedPath::Directory { prefix, metadata: None });
|
|
}
|
|
}
|
|
|
|
if had_visibility {
|
|
Err(FsError::NotFound)
|
|
} else {
|
|
Err(FsError::Forbidden)
|
|
}
|
|
}
|
|
|
|
/// Parse WebDAV path to bucket and object key
|
|
fn parse_path(&self, path: &DavPath) -> Result<(String, Option<String>), FsError> {
|
|
let path_str = path.as_url_string();
|
|
let decoded_path = percent_decode_str(&path_str)
|
|
.decode_utf8()
|
|
.map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
if decoded_path.chars().any(char::is_control) {
|
|
return Err(FsError::GeneralFailure);
|
|
}
|
|
|
|
let cleaned_path = path::clean(&decoded_path);
|
|
let (bucket, object) = path::path_to_bucket_object(&cleaned_path);
|
|
|
|
if bucket.is_empty() {
|
|
return Ok((String::new(), None));
|
|
}
|
|
|
|
if object.contains(path::GLOBAL_DIR_SUFFIX) {
|
|
return Err(FsError::GeneralFailure);
|
|
}
|
|
|
|
let key = if object.is_empty() { None } else { Some(object) };
|
|
Ok((bucket, key))
|
|
}
|
|
|
|
/// Check if path is root
|
|
fn is_root(&self, path: &DavPath) -> bool {
|
|
let path_str = path.as_url_string();
|
|
path_str == "/" || path_str.is_empty()
|
|
}
|
|
|
|
/// List all buckets (for root path)
|
|
async fn list_buckets(&self) -> FsResult<Vec<WebDavDirEntry>> {
|
|
match authorize_operation(&self.session_context, &S3Action::ListBuckets, "", None).await {
|
|
Ok(_) => {}
|
|
Err(_e) => {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
}
|
|
|
|
match self
|
|
.storage
|
|
.list_buckets(
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(output) => {
|
|
let mut entries = Vec::new();
|
|
if let Some(buckets) = output.buckets {
|
|
for bucket in buckets {
|
|
if let Some(ref bucket_name) = bucket.name {
|
|
let modified = bucket
|
|
.creation_date
|
|
.map(|dt| {
|
|
let offset_dt: time::OffsetDateTime = dt.into();
|
|
SystemTime::from(offset_dt)
|
|
})
|
|
.unwrap_or_else(SystemTime::now);
|
|
|
|
entries.push(WebDavDirEntry {
|
|
name: bucket_name.clone(),
|
|
metadata: WebDavMetaData {
|
|
size: 0,
|
|
modified,
|
|
created: modified,
|
|
is_dir: true,
|
|
etag: None,
|
|
content_type: None,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Ok(entries)
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_BUCKET_LIST_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
error = %e,
|
|
access_key = %MaskedAccessKey(&self.session_context.principal.user_identity.credentials.access_key),
|
|
"webdav bucket list failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// List objects in a bucket
|
|
async fn list_objects(&self, bucket: &str, prefix: Option<&str>) -> FsResult<Vec<WebDavDirEntry>> {
|
|
// Authorize the operation
|
|
authorize_operation(&self.session_context, &S3Action::ListBucket, bucket, prefix)
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
let prefix_with_slash = prefix.map(|p| if p.ends_with('/') { p.to_string() } else { format!("{}/", p) });
|
|
|
|
let list_input = ListObjectsV2Input::builder()
|
|
.bucket(bucket.to_string())
|
|
.prefix(prefix_with_slash.clone())
|
|
.delimiter(Some("/".to_string()))
|
|
.build()
|
|
.map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
match self
|
|
.storage
|
|
.list_objects_v2(
|
|
list_input,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(output) => {
|
|
let mut entries = Vec::new();
|
|
|
|
// Collect common prefix base names for filtering
|
|
let common_prefix_names: std::collections::HashSet<String> = output
|
|
.common_prefixes
|
|
.as_ref()
|
|
.map(|prefixes| {
|
|
prefixes
|
|
.iter()
|
|
.filter_map(|p| p.prefix.as_ref())
|
|
.map(|p| {
|
|
std::path::PathBuf::from(p.trim_end_matches('/'))
|
|
.file_name()
|
|
.map(|n| n.to_string_lossy().to_string())
|
|
.unwrap_or_else(|| p.clone())
|
|
})
|
|
.collect()
|
|
})
|
|
.unwrap_or_default();
|
|
|
|
// Add files (objects)
|
|
if let Some(objects) = output.contents {
|
|
for obj in objects {
|
|
if let Some(key) = obj.key {
|
|
// Filter: only show files directly in current directory
|
|
let should_show = if prefix.is_none() {
|
|
!key.contains('/')
|
|
} else {
|
|
key.starts_with(&prefix_with_slash.clone().unwrap_or_default())
|
|
};
|
|
|
|
if !should_show {
|
|
continue;
|
|
}
|
|
|
|
let filename = std::path::PathBuf::from(key.as_str())
|
|
.file_name()
|
|
.map(|n| n.to_string_lossy().to_string())
|
|
.unwrap_or_else(|| key.clone());
|
|
|
|
let size = obj.size.unwrap_or(0) as u64;
|
|
|
|
// Skip directory markers (keys ending with /)
|
|
if key.ends_with('/') {
|
|
continue;
|
|
}
|
|
|
|
// Skip 0-byte objects that match a directory name (Windows WebDAV duplicates)
|
|
if size == 0 && common_prefix_names.contains(&filename) {
|
|
continue;
|
|
}
|
|
|
|
let modified = obj
|
|
.last_modified
|
|
.map(|dt| {
|
|
let offset_dt: time::OffsetDateTime = dt.into();
|
|
SystemTime::from(offset_dt)
|
|
})
|
|
.unwrap_or_else(SystemTime::now);
|
|
|
|
entries.push(WebDavDirEntry {
|
|
name: filename,
|
|
metadata: WebDavMetaData {
|
|
size,
|
|
modified,
|
|
created: modified,
|
|
is_dir: false,
|
|
etag: obj.e_tag.as_ref().map(etag_to_string),
|
|
content_type: None,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
// Add directories (common prefixes)
|
|
if let Some(common_prefixes) = output.common_prefixes {
|
|
for prefix in common_prefixes {
|
|
if let Some(prefix_str) = prefix.prefix {
|
|
let dir_name = std::path::PathBuf::from(prefix_str.as_str().trim_end_matches('/'))
|
|
.file_name()
|
|
.map(|n| n.to_string_lossy().to_string())
|
|
.unwrap_or_else(|| prefix_str.clone());
|
|
|
|
entries.push(WebDavDirEntry {
|
|
name: dir_name,
|
|
metadata: WebDavMetaData {
|
|
size: 0,
|
|
modified: SystemTime::now(),
|
|
created: SystemTime::now(),
|
|
is_dir: true,
|
|
etag: None,
|
|
content_type: None,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(entries)
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_LIST_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
bucket = %bucket,
|
|
prefix = %prefix_with_slash.unwrap_or_default(),
|
|
error = %e,
|
|
"webdav list failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Recursively delete all objects in a bucket, then delete the bucket itself
|
|
async fn delete_bucket_recursively(&self, bucket: &str) -> FsResult<()> {
|
|
// SECURITY: s3:DeleteBucket does not imply the right to destroy the
|
|
// bucket contents. Enumerating and deleting each object are separate
|
|
// authorization boundaries and must be cleared on their own.
|
|
authorize_operation(&self.session_context, &S3Action::ListBucket, bucket, None)
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
// First, delete all objects in the bucket (with pagination)
|
|
let mut continuation_token = None;
|
|
loop {
|
|
let mut list_input = ListObjectsV2Input::builder().bucket(bucket.to_string());
|
|
|
|
if let Some(token) = continuation_token {
|
|
list_input = list_input.continuation_token(token);
|
|
}
|
|
|
|
let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
if let Ok(output) = self
|
|
.storage
|
|
.list_objects_v2(
|
|
list_input,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
// Delete all objects in this page
|
|
if let Some(objects) = output.contents {
|
|
for obj in objects {
|
|
if let Some(obj_key) = obj.key {
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, bucket, Some(&obj_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
let _ = self
|
|
.storage
|
|
.delete_object(
|
|
bucket,
|
|
&obj_key,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check if there are more objects
|
|
if !output.is_truncated.unwrap_or(false) {
|
|
break;
|
|
}
|
|
continuation_token = Some(output.next_continuation_token);
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Then delete the bucket
|
|
match self
|
|
.storage
|
|
.delete_bucket(
|
|
bucket,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => Ok(()),
|
|
Err(e) if e.to_string().contains("NoSuchBucket") => Ok(()),
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_DELETE_FAILED,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "bucket_delete_failed",
|
|
bucket = %bucket,
|
|
error = %e,
|
|
"webdav delete failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<S> DavFileSystem for WebDavDriver<S>
|
|
where
|
|
S: S3StorageBackend + Debug + Clone + Send + Sync + 'static,
|
|
{
|
|
fn open<'a>(&'a self, path: &'a DavPath, options: OpenOptions) -> FsFuture<'a, Box<dyn DavFile>> {
|
|
let storage = self.storage.clone();
|
|
let session_context = self.session_context.clone();
|
|
|
|
async move {
|
|
let (bucket, key) = self.parse_path(path)?;
|
|
|
|
if bucket.is_empty() {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
|
|
let key = key.ok_or(FsError::Forbidden)?; // Cannot open a bucket as a file
|
|
|
|
// Check authorization based on operation type
|
|
if options.write || options.create || options.create_new || options.append {
|
|
authorize_operation(&session_context, &S3Action::PutObject, &bucket, Some(&key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
} else {
|
|
authorize_operation(&session_context, &S3Action::GetObject, &bucket, Some(&key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
}
|
|
|
|
let is_write = options.write || options.create || options.create_new || options.append;
|
|
let file = WebDavFile::new(storage, session_context, bucket, key, is_write);
|
|
|
|
Ok(Box::new(file) as Box<dyn DavFile>)
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn read_dir<'a>(&'a self, path: &'a DavPath, _meta: ReadDirMeta) -> FsFuture<'a, FsStream<Box<dyn DavDirEntry>>> {
|
|
async move {
|
|
let entries = if self.is_root(path) {
|
|
self.list_buckets().await?
|
|
} else {
|
|
let (bucket, prefix) = self.parse_path(path)?;
|
|
if bucket.is_empty() {
|
|
self.list_buckets().await?
|
|
} else {
|
|
self.list_objects(&bucket, prefix.as_deref()).await?
|
|
}
|
|
};
|
|
|
|
let stream = stream::iter(entries.into_iter().map(|e| Ok(Box::new(e) as Box<dyn DavDirEntry>)));
|
|
Ok(Box::pin(stream) as FsStream<Box<dyn DavDirEntry>>)
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn metadata<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, Box<dyn DavMetaData>> {
|
|
async move {
|
|
if self.is_root(path) {
|
|
return Ok(Box::new(WebDavMetaData {
|
|
size: 0,
|
|
modified: SystemTime::now(),
|
|
created: SystemTime::now(),
|
|
is_dir: true,
|
|
etag: None,
|
|
content_type: None,
|
|
}) as Box<dyn DavMetaData>);
|
|
}
|
|
|
|
let (bucket, key) = self.parse_path(path)?;
|
|
|
|
if bucket.is_empty() {
|
|
return Ok(Box::new(WebDavMetaData {
|
|
size: 0,
|
|
modified: SystemTime::now(),
|
|
created: SystemTime::now(),
|
|
is_dir: true,
|
|
etag: None,
|
|
content_type: None,
|
|
}) as Box<dyn DavMetaData>);
|
|
}
|
|
|
|
if let Some(key) = key {
|
|
return match self.resolve_path(&bucket, &key).await? {
|
|
ResolvedPath::File(output) => {
|
|
let size = output.content_length.unwrap_or(0) as u64;
|
|
let modified = output
|
|
.last_modified
|
|
.map(|dt| {
|
|
let offset_dt: time::OffsetDateTime = dt.into();
|
|
SystemTime::from(offset_dt)
|
|
})
|
|
.unwrap_or_else(SystemTime::now);
|
|
|
|
Ok(Box::new(WebDavMetaData {
|
|
size,
|
|
modified,
|
|
created: modified,
|
|
is_dir: false,
|
|
etag: output.e_tag.as_ref().map(etag_to_string),
|
|
content_type: output.content_type.map(|c| c.to_string()),
|
|
}) as Box<dyn DavMetaData>)
|
|
}
|
|
ResolvedPath::Directory { metadata, .. } => {
|
|
let modified = metadata
|
|
.as_ref()
|
|
.and_then(|output| output.last_modified.as_ref())
|
|
.map(|dt| {
|
|
let offset_dt: time::OffsetDateTime = dt.clone().into();
|
|
SystemTime::from(offset_dt)
|
|
})
|
|
.unwrap_or_else(SystemTime::now);
|
|
|
|
Ok(Box::new(WebDavMetaData {
|
|
size: 0,
|
|
modified,
|
|
created: modified,
|
|
is_dir: true,
|
|
etag: metadata.as_ref().and_then(|output| output.e_tag.as_ref().map(etag_to_string)),
|
|
content_type: metadata.and_then(|output| output.content_type.map(|c| c.to_string())),
|
|
}) as Box<dyn DavMetaData>)
|
|
}
|
|
};
|
|
} else {
|
|
// Get bucket metadata
|
|
authorize_operation(&self.session_context, &S3Action::HeadBucket, &bucket, None)
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
match self
|
|
.storage
|
|
.head_bucket(
|
|
&bucket,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => Ok(Box::new(WebDavMetaData {
|
|
size: 0,
|
|
modified: SystemTime::now(),
|
|
created: SystemTime::now(),
|
|
is_dir: true,
|
|
etag: None,
|
|
content_type: None,
|
|
}) as Box<dyn DavMetaData>),
|
|
Err(e) => {
|
|
debug!(
|
|
event = EVENT_WEBDAV_BUCKET_METADATA_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
result = "not_found",
|
|
bucket = %bucket,
|
|
error = %e,
|
|
"WebDAV bucket metadata listed"
|
|
);
|
|
Err(FsError::NotFound)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn create_dir<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
|
|
async move {
|
|
let (bucket, key) = self.parse_path(path)?;
|
|
|
|
if bucket.is_empty() {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
|
|
if let Some(key_str) = key {
|
|
// Creating a "directory" in S3 by creating a zero-byte object with trailing slash
|
|
let dir_key = if key_str.ends_with('/') {
|
|
key_str.to_string()
|
|
} else {
|
|
format!("{}/", key_str)
|
|
};
|
|
|
|
authorize_operation(&self.session_context, &S3Action::PutObject, &bucket, Some(&dir_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
// Create empty streaming blob for directory marker
|
|
let stream = futures_util::stream::once(async { Ok::<Bytes, std::io::Error>(Bytes::new()) });
|
|
let streaming_blob = s3s::dto::StreamingBlob::wrap(stream);
|
|
|
|
let put_input = s3s::dto::PutObjectInput::builder()
|
|
.bucket(bucket.clone())
|
|
.key(dir_key.clone())
|
|
.content_length(Some(0))
|
|
.content_type(Some("application/x-directory".to_string()))
|
|
.body(Some(streaming_blob))
|
|
.build()
|
|
.map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
match self
|
|
.storage
|
|
.put_object(
|
|
put_input,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
debug!(
|
|
event = EVENT_WEBDAV_DIRECTORY_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "created",
|
|
bucket = %bucket,
|
|
object = %dir_key,
|
|
"WebDAV directory marker created"
|
|
);
|
|
return Ok(());
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_DIRECTORY_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "create_failed",
|
|
bucket = %bucket,
|
|
object = %dir_key,
|
|
error = %e,
|
|
"WebDAV directory marker create failed"
|
|
);
|
|
return Err(FsError::GeneralFailure);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Create bucket
|
|
authorize_operation(&self.session_context, &S3Action::CreateBucket, &bucket, None)
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
match self
|
|
.storage
|
|
.create_bucket(
|
|
&bucket,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
debug!(
|
|
event = EVENT_WEBDAV_DIRECTORY_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "bucket_created",
|
|
bucket = %bucket,
|
|
"WebDAV bucket created"
|
|
);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_DIRECTORY_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "bucket_create_failed",
|
|
bucket = %bucket,
|
|
error = %e,
|
|
"WebDAV bucket create failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn remove_dir<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
|
|
async move {
|
|
let (bucket, key) = self.parse_path(path)?;
|
|
|
|
if bucket.is_empty() {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
|
|
if let Some(prefix) = key {
|
|
// Delete all objects with this prefix (subdirectory)
|
|
let prefix_with_slash = if prefix.ends_with('/') {
|
|
prefix.to_string()
|
|
} else {
|
|
format!("{}/", prefix)
|
|
};
|
|
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &bucket, Some(&prefix_with_slash))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
// SECURITY: clearing s3:DeleteObject on the directory marker key
|
|
// says nothing about the children stored under it. Enumerating the
|
|
// prefix and deleting each child are separate authorization
|
|
// boundaries and must be cleared on their own.
|
|
authorize_operation(&self.session_context, &S3Action::ListBucket, &bucket, Some(&prefix_with_slash))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
// List and delete all objects with this prefix
|
|
let mut continuation_token = None;
|
|
loop {
|
|
let mut list_input = ListObjectsV2Input::builder()
|
|
.bucket(bucket.clone())
|
|
.prefix(Some(prefix_with_slash.clone()));
|
|
|
|
if let Some(token) = continuation_token {
|
|
list_input = list_input.continuation_token(token);
|
|
}
|
|
|
|
let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?;
|
|
|
|
if let Ok(output) = self
|
|
.storage
|
|
.list_objects_v2(
|
|
list_input,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
if let Some(objects) = output.contents {
|
|
for obj in objects {
|
|
if let Some(obj_key) = obj.key {
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &bucket, Some(&obj_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
let _ = self
|
|
.storage
|
|
.delete_object(
|
|
&bucket,
|
|
&obj_key,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
if !output.is_truncated.unwrap_or(false) {
|
|
break;
|
|
}
|
|
continuation_token = Some(output.next_continuation_token);
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Also delete the directory marker itself
|
|
let _ = self
|
|
.storage
|
|
.delete_object(
|
|
&bucket,
|
|
&prefix_with_slash,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await;
|
|
|
|
return Ok(());
|
|
}
|
|
|
|
// Delete bucket
|
|
authorize_operation(&self.session_context, &S3Action::DeleteBucket, &bucket, None)
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
self.delete_bucket_recursively(&bucket).await
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn remove_file<'a>(&'a self, path: &'a DavPath) -> FsFuture<'a, ()> {
|
|
async move {
|
|
let (bucket, key) = self.parse_path(path)?;
|
|
|
|
if bucket.is_empty() {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
|
|
let key = key.ok_or(FsError::Forbidden)?;
|
|
|
|
// Authorize delete object
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &bucket, Some(&key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
match self
|
|
.storage
|
|
.delete_object(
|
|
&bucket,
|
|
&key,
|
|
&self.session_context.principal.user_identity.credentials.access_key,
|
|
&self.session_context.principal.user_identity.credentials.secret_key,
|
|
)
|
|
.await
|
|
{
|
|
Ok(_) => {
|
|
debug!(
|
|
event = EVENT_WEBDAV_OBJECT_DELETE_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "deleted",
|
|
bucket = %bucket,
|
|
object = %key,
|
|
"WebDAV object deleted"
|
|
);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
error!(
|
|
event = EVENT_WEBDAV_OBJECT_DELETE_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "delete_failed",
|
|
bucket = %bucket,
|
|
object = %key,
|
|
error = %e,
|
|
"WebDAV object delete failed"
|
|
);
|
|
Err(FsError::GeneralFailure)
|
|
}
|
|
}
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn rename<'a>(&'a self, from: &'a DavPath, to: &'a DavPath) -> FsFuture<'a, ()> {
|
|
async move {
|
|
let (src_bucket, src_key) = self.parse_path(from)?;
|
|
let (dst_bucket, dst_key) = self.parse_path(to)?;
|
|
|
|
if src_bucket.is_empty() || dst_bucket.is_empty() {
|
|
return Err(FsError::Forbidden);
|
|
}
|
|
|
|
let src_key = src_key.ok_or(FsError::Forbidden)?;
|
|
let dst_key = dst_key.ok_or(FsError::Forbidden)?;
|
|
let (access_key, secret_key) = self.credentials();
|
|
let resolved_src = self.resolve_path(&src_bucket, &src_key).await?;
|
|
let (src_prefix, include_src_marker) = match resolved_src {
|
|
ResolvedPath::File(_) => {
|
|
authorize_operation(&self.session_context, &S3Action::GetObject, &src_bucket, Some(&src_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::PutObject, &dst_bucket, Some(&dst_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &src_bucket, Some(&src_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
self.copy_object_streaming(&src_bucket, &src_key, &dst_bucket, &dst_key)
|
|
.await?;
|
|
|
|
self.storage
|
|
.delete_object(&src_bucket, &src_key, access_key, secret_key)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_RENAME_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "source_delete_failed",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
error = %e,
|
|
"WebDAV rename source delete failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
debug!(
|
|
event = EVENT_WEBDAV_RENAME_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "file_renamed",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
"WebDAV file renamed"
|
|
);
|
|
return Ok(());
|
|
}
|
|
ResolvedPath::Directory { prefix, .. } => {
|
|
let include_src_marker =
|
|
matches!(self.probe_head_object(&src_bucket, &src_key).await?, HeadObjectProbe::Found(_));
|
|
(prefix, include_src_marker)
|
|
}
|
|
};
|
|
let dst_prefix = format!("{}/", dst_key);
|
|
|
|
authorize_operation(&self.session_context, &S3Action::ListBucket, &src_bucket, Some(&src_prefix))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
let mut continuation_token: Option<String> = None;
|
|
let mut renamed_any = false;
|
|
|
|
if include_src_marker {
|
|
authorize_operation(&self.session_context, &S3Action::GetObject, &src_bucket, Some(&src_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::PutObject, &dst_bucket, Some(&dst_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &src_bucket, Some(&src_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
|
|
self.execute_directory_rename_pairs(&src_bucket, &dst_bucket, &[(src_key.clone(), dst_key.clone())])
|
|
.await?;
|
|
renamed_any = true;
|
|
}
|
|
|
|
loop {
|
|
let mut list_builder = ListObjectsV2Input::builder()
|
|
.bucket(src_bucket.clone())
|
|
.prefix(Some(src_prefix.clone()));
|
|
|
|
if let Some(ref token) = continuation_token {
|
|
list_builder = list_builder.continuation_token(Some(token.clone()));
|
|
}
|
|
|
|
let list_input = list_builder.build().map_err(|_| FsError::GeneralFailure)?;
|
|
let output = self
|
|
.storage
|
|
.list_objects_v2(list_input, access_key, secret_key)
|
|
.await
|
|
.map_err(|e| {
|
|
error!(
|
|
event = EVENT_WEBDAV_RENAME_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "directory_list_failed",
|
|
src_bucket = %src_bucket,
|
|
src_prefix = %src_prefix,
|
|
dst_bucket = %dst_bucket,
|
|
dst_prefix = %dst_prefix,
|
|
error = %e,
|
|
"WebDAV rename directory listing failed"
|
|
);
|
|
FsError::GeneralFailure
|
|
})?;
|
|
|
|
let mut page_pairs: Vec<(String, String)> = Vec::new();
|
|
if let Some(objects) = output.contents {
|
|
for obj in objects {
|
|
if let Some(obj_key) = obj.key {
|
|
let new_key = obj_key.replacen(&src_prefix, &dst_prefix, 1);
|
|
page_pairs.push((obj_key, new_key));
|
|
}
|
|
}
|
|
}
|
|
|
|
if !page_pairs.is_empty() {
|
|
for (src_obj_key, dst_obj_key) in &page_pairs {
|
|
authorize_operation(&self.session_context, &S3Action::GetObject, &src_bucket, Some(src_obj_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::PutObject, &dst_bucket, Some(dst_obj_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
authorize_operation(&self.session_context, &S3Action::DeleteObject, &src_bucket, Some(src_obj_key))
|
|
.await
|
|
.map_err(|_| FsError::Forbidden)?;
|
|
}
|
|
|
|
self.execute_directory_rename_pairs(&src_bucket, &dst_bucket, &page_pairs)
|
|
.await?;
|
|
renamed_any = true;
|
|
}
|
|
|
|
if !output.is_truncated.unwrap_or(false) {
|
|
break;
|
|
}
|
|
continuation_token = output.next_continuation_token;
|
|
}
|
|
|
|
if !renamed_any {
|
|
debug!(
|
|
event = EVENT_WEBDAV_RENAME_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
result = "source_not_found",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
"WebDAV rename source not found"
|
|
);
|
|
return Err(FsError::NotFound);
|
|
}
|
|
|
|
debug!(
|
|
event = EVENT_WEBDAV_RENAME_STATE,
|
|
component = LOG_COMPONENT_PROTOCOLS,
|
|
subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
|
|
state = "directory_renamed",
|
|
src_bucket = %src_bucket,
|
|
src_object = %src_key,
|
|
dst_bucket = %dst_bucket,
|
|
dst_object = %dst_key,
|
|
"WebDAV directory renamed"
|
|
);
|
|
Ok(())
|
|
}
|
|
.boxed()
|
|
}
|
|
|
|
fn copy<'a>(&'a self, _from: &'a DavPath, _to: &'a DavPath) -> FsFuture<'a, ()> {
|
|
// Could implement using S3 CopyObject, but not required for basic WebDAV
|
|
async move { Err(FsError::NotImplemented) }.boxed()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::WebDavDriver;
|
|
use crate::common::client::s3::StorageBackend as S3StorageBackend;
|
|
use crate::common::gateway::{S3Action, with_test_auth_override};
|
|
use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
|
|
use async_trait::async_trait;
|
|
use bytes::Bytes;
|
|
use dav_server::davpath::DavPath;
|
|
use dav_server::fs::{DavFileSystem, FsError};
|
|
use futures_util::StreamExt;
|
|
use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
|
|
use rustfs_credentials::Credentials;
|
|
use rustfs_policy::auth::UserIdentity;
|
|
use s3s::dto::*;
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::fmt::{Debug, Formatter};
|
|
use std::net::{IpAddr, Ipv4Addr};
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
#[derive(Clone)]
|
|
struct DummyStorage;
|
|
|
|
impl Debug for DummyStorage {
|
|
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
|
f.write_str("DummyStorage")
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl S3StorageBackend for DummyStorage {
|
|
type Error = std::io::Error;
|
|
|
|
async fn get_object(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
_start_pos: Option<u64>,
|
|
) -> Result<GetObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn get_object_range(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
_start_pos: u64,
|
|
_length: u64,
|
|
) -> Result<GetObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn put_object(
|
|
&self,
|
|
_input: PutObjectInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<PutObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn delete_object(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<DeleteObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn head_object(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<HeadObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn head_bucket(
|
|
&self,
|
|
_bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<HeadBucketOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn list_objects_v2(
|
|
&self,
|
|
_input: ListObjectsV2Input,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<ListObjectsV2Output, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn create_bucket(
|
|
&self,
|
|
_bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CreateBucketOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn delete_bucket(
|
|
&self,
|
|
_bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<DeleteBucketOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn copy_object(
|
|
&self,
|
|
_input: CopyObjectInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CopyObjectOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn create_multipart_upload(
|
|
&self,
|
|
_input: CreateMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CreateMultipartUploadOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn upload_part(
|
|
&self,
|
|
_input: UploadPartInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<UploadPartOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn complete_multipart_upload(
|
|
&self,
|
|
_input: CompleteMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CompleteMultipartUploadOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn abort_multipart_upload(
|
|
&self,
|
|
_input: AbortMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<AbortMultipartUploadOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
|
|
async fn upload_part_copy(
|
|
&self,
|
|
_input: UploadPartCopyInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<UploadPartCopyOutput, Self::Error> {
|
|
unreachable!("parse_path tests should not hit storage")
|
|
}
|
|
}
|
|
|
|
fn driver() -> WebDavDriver<DummyStorage> {
|
|
let identity = UserIdentity::new(Credentials {
|
|
access_key: "ak".to_string(),
|
|
secret_key: "sk".to_string(),
|
|
..Default::default()
|
|
});
|
|
let session_context = SessionContext::new(
|
|
ProtocolPrincipal::new(Arc::new(identity)),
|
|
Protocol::WebDav,
|
|
IpAddr::V4(Ipv4Addr::LOCALHOST),
|
|
);
|
|
|
|
WebDavDriver::new(DummyStorage, Arc::new(session_context))
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct RecordingStorageState {
|
|
objects: HashMap<(String, String), Vec<u8>>,
|
|
put_keys: Vec<String>,
|
|
delete_keys: Vec<String>,
|
|
deleted_buckets: Vec<String>,
|
|
fail_delete_keys: HashSet<String>,
|
|
}
|
|
|
|
#[derive(Clone, Default)]
|
|
struct RecordingStorage {
|
|
state: Arc<Mutex<RecordingStorageState>>,
|
|
}
|
|
|
|
impl Debug for RecordingStorage {
|
|
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
|
|
f.write_str("RecordingStorage")
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl S3StorageBackend for RecordingStorage {
|
|
type Error = std::io::Error;
|
|
|
|
async fn get_object(
|
|
&self,
|
|
bucket: &str,
|
|
key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
_start_pos: Option<u64>,
|
|
) -> Result<GetObjectOutput, Self::Error> {
|
|
let data = self
|
|
.state
|
|
.lock()
|
|
.expect("recording storage lock poisoned")
|
|
.objects
|
|
.get(&(bucket.to_string(), key.to_string()))
|
|
.cloned()
|
|
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::NotFound, "missing object"))?;
|
|
|
|
let content_length = data.len() as i64;
|
|
let body =
|
|
StreamingBlob::wrap(futures_util::stream::once(async move { Ok::<Bytes, std::io::Error>(Bytes::from(data)) }));
|
|
|
|
Ok(GetObjectOutput {
|
|
body: Some(body),
|
|
content_length: Some(content_length),
|
|
..Default::default()
|
|
})
|
|
}
|
|
|
|
async fn get_object_range(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
_start_pos: u64,
|
|
_length: u64,
|
|
) -> Result<GetObjectOutput, Self::Error> {
|
|
unreachable!("range reads are not used in rename regression tests")
|
|
}
|
|
|
|
async fn put_object(
|
|
&self,
|
|
mut input: PutObjectInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<PutObjectOutput, Self::Error> {
|
|
let bucket = input.bucket.clone();
|
|
let key = input.key.clone();
|
|
let mut bytes = Vec::new();
|
|
|
|
if let Some(mut body) = input.body.take() {
|
|
while let Some(chunk) = body.next().await {
|
|
let chunk = chunk.map_err(|e| std::io::Error::other(e.to_string()))?;
|
|
bytes.extend_from_slice(&chunk);
|
|
}
|
|
}
|
|
|
|
let mut state = self.state.lock().expect("recording storage lock poisoned");
|
|
state.put_keys.push(key.clone());
|
|
state.objects.insert((bucket, key), bytes);
|
|
|
|
Ok(PutObjectOutput::default())
|
|
}
|
|
|
|
async fn delete_object(
|
|
&self,
|
|
bucket: &str,
|
|
key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<DeleteObjectOutput, Self::Error> {
|
|
let mut state = self.state.lock().expect("recording storage lock poisoned");
|
|
state.delete_keys.push(key.to_string());
|
|
if state.fail_delete_keys.contains(key) {
|
|
return Err(std::io::Error::other("injected delete failure"));
|
|
}
|
|
state.objects.remove(&(bucket.to_string(), key.to_string()));
|
|
Ok(DeleteObjectOutput::default())
|
|
}
|
|
|
|
async fn head_object(
|
|
&self,
|
|
_bucket: &str,
|
|
_key: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<HeadObjectOutput, Self::Error> {
|
|
unreachable!("head_object is not used in rename regression tests")
|
|
}
|
|
|
|
async fn head_bucket(
|
|
&self,
|
|
_bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<HeadBucketOutput, Self::Error> {
|
|
unreachable!("head_bucket is not used in rename regression tests")
|
|
}
|
|
|
|
async fn list_objects_v2(
|
|
&self,
|
|
input: ListObjectsV2Input,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<ListObjectsV2Output, Self::Error> {
|
|
let prefix = input.prefix.map(|p| p.to_string()).unwrap_or_default();
|
|
let mut keys: Vec<String> = self
|
|
.state
|
|
.lock()
|
|
.expect("recording storage lock poisoned")
|
|
.objects
|
|
.keys()
|
|
.filter(|(bucket, key)| *bucket == input.bucket && key.starts_with(&prefix))
|
|
.map(|(_, key)| key.clone())
|
|
.collect();
|
|
keys.sort();
|
|
|
|
Ok(ListObjectsV2Output {
|
|
contents: Some(
|
|
keys.into_iter()
|
|
.map(|key| Object {
|
|
key: Some(ObjectKey::from(key)),
|
|
..Default::default()
|
|
})
|
|
.collect(),
|
|
),
|
|
is_truncated: Some(false),
|
|
..Default::default()
|
|
})
|
|
}
|
|
|
|
async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
|
|
unreachable!("list_buckets is not used in rename regression tests")
|
|
}
|
|
|
|
async fn create_bucket(
|
|
&self,
|
|
_bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CreateBucketOutput, Self::Error> {
|
|
unreachable!("create_bucket is not used in rename regression tests")
|
|
}
|
|
|
|
async fn delete_bucket(
|
|
&self,
|
|
bucket: &str,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<DeleteBucketOutput, Self::Error> {
|
|
self.state
|
|
.lock()
|
|
.expect("recording storage lock poisoned")
|
|
.deleted_buckets
|
|
.push(bucket.to_string());
|
|
Ok(DeleteBucketOutput::default())
|
|
}
|
|
|
|
async fn copy_object(
|
|
&self,
|
|
_input: CopyObjectInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CopyObjectOutput, Self::Error> {
|
|
unreachable!("copy_object is not used in rename regression tests")
|
|
}
|
|
|
|
async fn create_multipart_upload(
|
|
&self,
|
|
_input: CreateMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CreateMultipartUploadOutput, Self::Error> {
|
|
unreachable!("create_multipart_upload is not used in rename regression tests")
|
|
}
|
|
|
|
async fn upload_part(
|
|
&self,
|
|
_input: UploadPartInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<UploadPartOutput, Self::Error> {
|
|
unreachable!("upload_part is not used in rename regression tests")
|
|
}
|
|
|
|
async fn complete_multipart_upload(
|
|
&self,
|
|
_input: CompleteMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<CompleteMultipartUploadOutput, Self::Error> {
|
|
unreachable!("complete_multipart_upload is not used in rename regression tests")
|
|
}
|
|
|
|
async fn abort_multipart_upload(
|
|
&self,
|
|
_input: AbortMultipartUploadInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<AbortMultipartUploadOutput, Self::Error> {
|
|
unreachable!("abort_multipart_upload is not used in rename regression tests")
|
|
}
|
|
|
|
async fn upload_part_copy(
|
|
&self,
|
|
_input: UploadPartCopyInput,
|
|
_access_key: &str,
|
|
_secret_key: &str,
|
|
) -> Result<UploadPartCopyOutput, Self::Error> {
|
|
unreachable!("upload_part_copy is not used in rename regression tests")
|
|
}
|
|
}
|
|
|
|
fn recording_driver(
|
|
initial_objects: &[(&str, &str, &[u8])],
|
|
fail_delete_keys: &[&str],
|
|
) -> (WebDavDriver<RecordingStorage>, RecordingStorage) {
|
|
let storage = RecordingStorage::default();
|
|
let identity = UserIdentity::new(Credentials {
|
|
access_key: "ak".to_string(),
|
|
secret_key: "sk".to_string(),
|
|
..Default::default()
|
|
});
|
|
let session_context = SessionContext::new(
|
|
ProtocolPrincipal::new(Arc::new(identity)),
|
|
Protocol::WebDav,
|
|
IpAddr::V4(Ipv4Addr::LOCALHOST),
|
|
);
|
|
|
|
let state = RecordingStorageState {
|
|
objects: initial_objects
|
|
.iter()
|
|
.map(|(bucket, key, body)| (((*bucket).to_string(), (*key).to_string()), body.to_vec()))
|
|
.collect(),
|
|
fail_delete_keys: fail_delete_keys.iter().map(|key| (*key).to_string()).collect(),
|
|
..Default::default()
|
|
};
|
|
|
|
*storage.state.lock().expect("recording storage lock poisoned") = state;
|
|
|
|
(WebDavDriver::new(storage.clone(), Arc::new(session_context)), storage)
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_decodes_url_encoded_object_names() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/%E6%96%87%E4%BB%B6%20name.txt").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert_eq!(key.as_deref(), Some("文件 name.txt"));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_rejects_invalid_utf8_percent_encoding() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/%FFreport.txt").expect("path should parse");
|
|
|
|
let err = driver.parse_path(&path).expect_err("invalid utf8 should be rejected");
|
|
|
|
assert_eq!(err, FsError::GeneralFailure);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_handles_directory_paths_with_trailing_slash() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/folder/").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert_eq!(key.as_deref(), Some("folder"));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_handles_chinese_directory_names() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/%E6%96%B0%E5%BB%BA%E6%96%87%E4%BB%B6%E5%A4%B9%20(4)").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert_eq!(key.as_deref(), Some("新建文件夹 (4)"));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_handles_nested_paths() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/dir/subdir/file.txt").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert_eq!(key.as_deref(), Some("dir/subdir/file.txt"));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_returns_none_key_for_bucket_root() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert!(key.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_handles_url_encoded_spaces_in_object_name() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/file%20with%20spaces.txt").expect("path should parse");
|
|
|
|
let (bucket, key) = driver.parse_path(&path).expect("path should decode");
|
|
|
|
assert_eq!(bucket, "bucket");
|
|
assert_eq!(key.as_deref(), Some("file with spaces.txt"));
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_rejects_control_bytes_after_decode() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/report%0Aname.txt").expect("path should parse");
|
|
|
|
let err = driver.parse_path(&path).expect_err("control bytes should be rejected");
|
|
|
|
assert_eq!(err, FsError::GeneralFailure);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_path_rejects_internal_directory_marker() {
|
|
let driver = driver();
|
|
let path = DavPath::new("/bucket/__XLDIR__").expect("path should parse");
|
|
|
|
let err = driver
|
|
.parse_path(&path)
|
|
.expect_err("internal directory marker should be rejected");
|
|
|
|
assert_eq!(err, FsError::GeneralFailure);
|
|
}
|
|
|
|
proptest::proptest! {
|
|
#[test]
|
|
fn parse_path_never_leaks_control_bytes_or_traversal_in_ok_output(
|
|
input in proptest::prelude::any::<String>(),
|
|
) {
|
|
let driver = driver();
|
|
let encoded = utf8_percent_encode(&input, NON_ALPHANUMERIC).to_string();
|
|
let Ok(path) = DavPath::new(&format!("/{encoded}")) else {
|
|
return Ok(());
|
|
};
|
|
|
|
match driver.parse_path(&path) {
|
|
Err(err) => {
|
|
proptest::prop_assert_eq!(err, FsError::GeneralFailure);
|
|
}
|
|
Ok((bucket, key)) => {
|
|
proptest::prop_assert!(!bucket.contains('/'));
|
|
proptest::prop_assert!(!bucket.chars().any(char::is_control));
|
|
|
|
if let Some(k) = key.as_deref() {
|
|
proptest::prop_assert!(!k.chars().any(char::is_control));
|
|
proptest::prop_assert!(!k.starts_with('/'));
|
|
proptest::prop_assert!(!k.split('/').any(|segment| segment == ".."));
|
|
proptest::prop_assert!(!k.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A bucket DELETE wipes every object in the bucket, so `s3:DeleteBucket`
|
|
/// alone must not be enough: each object needs its own `s3:DeleteObject`
|
|
/// boundary. The backend would happily complete the whole recursive delete
|
|
/// if the per-object check were removed again.
|
|
#[tokio::test]
|
|
async fn bucket_delete_denied_per_object_leaves_contents_and_bucket_intact() {
|
|
let (driver, storage) = recording_driver(&[("bucket", "secret.txt", b"secret")], &[]);
|
|
let path = DavPath::new("/bucket/").expect("path should parse");
|
|
|
|
let err = with_test_auth_override(
|
|
|action, _bucket, _object| !matches!(action, S3Action::DeleteObject),
|
|
driver.remove_dir(&path),
|
|
)
|
|
.await
|
|
.expect_err("bucket DELETE must fail closed when s3:DeleteObject is denied for a bucket member");
|
|
|
|
assert_eq!(err, FsError::Forbidden);
|
|
|
|
let state = storage.state.lock().expect("recording storage lock poisoned");
|
|
assert!(state.delete_keys.is_empty(), "no object may be deleted once the deny lands");
|
|
assert!(
|
|
state.deleted_buckets.is_empty(),
|
|
"the bucket must survive when its contents could not be authorized for deletion"
|
|
);
|
|
assert!(state.objects.contains_key(&("bucket".to_string(), "secret.txt".to_string())));
|
|
}
|
|
|
|
/// A directory DELETE authorizes the `dir/` marker key, but the children
|
|
/// stored under that prefix are separate resources and each needs its own
|
|
/// `s3:DeleteObject` boundary.
|
|
#[tokio::test]
|
|
async fn directory_delete_denied_for_child_leaves_child_intact() {
|
|
let (driver, storage) = recording_driver(&[("bucket", "dir/child.txt", b"child")], &[]);
|
|
let path = DavPath::new("/bucket/dir/").expect("path should parse");
|
|
|
|
let err = with_test_auth_override(
|
|
|action, _bucket, object| !matches!((action, object), (S3Action::DeleteObject, Some("dir/child.txt"))),
|
|
driver.remove_dir(&path),
|
|
)
|
|
.await
|
|
.expect_err("directory DELETE must fail closed when a child object denies s3:DeleteObject");
|
|
|
|
assert_eq!(err, FsError::Forbidden);
|
|
|
|
let state = storage.state.lock().expect("recording storage lock poisoned");
|
|
assert!(
|
|
state.delete_keys.is_empty(),
|
|
"the denied child must not be deleted, got {:?}",
|
|
state.delete_keys
|
|
);
|
|
assert!(
|
|
state
|
|
.objects
|
|
.contains_key(&("bucket".to_string(), "dir/child.txt".to_string()))
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn directory_rename_returns_error_when_delete_fails_after_successful_copy() {
|
|
let (driver, storage) = recording_driver(
|
|
&[
|
|
("bucket", "src/file-a.txt", b"file-a"),
|
|
("bucket", "src/file-b.txt", b"file-b"),
|
|
],
|
|
&["src/file-a.txt"],
|
|
);
|
|
|
|
let err = driver
|
|
.execute_directory_rename_pairs(
|
|
"bucket",
|
|
"bucket",
|
|
&[
|
|
("src/file-a.txt".to_string(), "dst/file-a.txt".to_string()),
|
|
("src/file-b.txt".to_string(), "dst/file-b.txt".to_string()),
|
|
],
|
|
)
|
|
.await
|
|
.expect_err("delete failure should be surfaced");
|
|
|
|
assert_eq!(err, FsError::GeneralFailure);
|
|
|
|
let state = storage.state.lock().expect("recording storage lock poisoned");
|
|
assert_eq!(state.put_keys, vec!["dst/file-a.txt".to_string(), "dst/file-b.txt".to_string()]);
|
|
assert_eq!(state.delete_keys, vec!["src/file-a.txt".to_string()]);
|
|
assert!(
|
|
state
|
|
.objects
|
|
.contains_key(&("bucket".to_string(), "dst/file-a.txt".to_string()))
|
|
);
|
|
assert!(
|
|
state
|
|
.objects
|
|
.contains_key(&("bucket".to_string(), "dst/file-b.txt".to_string()))
|
|
);
|
|
}
|
|
}
|