kqjy/MyFSIO
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity

Fix S3 versioning (live-object VersionId, DM PUT/DELETE), harden DeleteObjects/ListObjects conformance, and run hot paths on blocking threads

Browse Source
This commit is contained in:
kqjy
2026-04-23 22:40:38 +08:00
parent bd405cc2fe
commit f2df64479c
9 changed files with 994 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/myfsio-common/src/error.rs
View File

@@ -5,6 +5,7 @@ pub enum S3ErrorCode {
AccessDenied,
BadDigest,
BucketAlreadyExists,
BucketAlreadyOwnedByYou,
BucketNotEmpty,
EntityTooLarge,
EntityTooSmall,
@@ -43,6 +44,7 @@ impl S3ErrorCode {
Self::AccessDenied => 403,
Self::BadDigest => 400,
Self::BucketAlreadyExists => 409,
Self::BucketAlreadyOwnedByYou => 409,
Self::BucketNotEmpty => 409,
Self::EntityTooLarge => 413,
Self::EntityTooSmall => 400,
@@ -72,7 +74,7 @@ impl S3ErrorCode {
Self::RequestTimeTooSkewed => 403,
Self::ServerSideEncryptionConfigurationNotFoundError => 404,
Self::SignatureDoesNotMatch => 403,
Self::SlowDown => 429,
Self::SlowDown => 503,
}
}
@@ -81,6 +83,7 @@ impl S3ErrorCode {
Self::AccessDenied => "AccessDenied",
Self::BadDigest => "BadDigest",
Self::BucketAlreadyExists => "BucketAlreadyExists",
Self::BucketAlreadyOwnedByYou => "BucketAlreadyOwnedByYou",
Self::BucketNotEmpty => "BucketNotEmpty",
Self::EntityTooLarge => "EntityTooLarge",
Self::EntityTooSmall => "EntityTooSmall",
@@ -121,6 +124,7 @@ impl S3ErrorCode {
Self::AccessDenied => "Access Denied",
Self::BadDigest => "The Content-MD5 or checksum value you specified did not match what we received",
Self::BucketAlreadyExists => "The requested bucket name is not available",
Self::BucketAlreadyOwnedByYou => "Your previous request to create the named bucket succeeded and you already own it",
Self::BucketNotEmpty => "The bucket you tried to delete is not empty",
Self::EntityTooLarge => "Your proposed upload exceeds the maximum allowed size",
Self::EntityTooSmall => "Your proposed upload is smaller than the minimum allowed object size",

4
crates/myfsio-server/src/handlers/mod.rs
View File

@@ -148,6 +148,7 @@ async fn ensure_object_lock_allows_write(
Ok(())
}
Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => Ok(()),
Err(myfsio_storage::error::StorageError::DeleteMarker { .. }) => Ok(()),
Err(err) => Err(storage_err_response(err)),
}
}
@@ -2666,7 +2667,8 @@ async fn evaluate_put_preconditions(
}
None
}
Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => {
Err(myfsio_storage::error::StorageError::ObjectNotFound { .. })
| Err(myfsio_storage::error::StorageError::DeleteMarker { .. }) => {
if has_if_match {
Some(s3_error_response(S3Error::from_code(
S3ErrorCode::PreconditionFailed,

23
crates/myfsio-server/src/middleware/ratelimit.rs
View File

@@ -162,20 +162,31 @@ pub async fn rate_limit_layer(
let limiter = state.select_limiter(&req);
match limiter.check(&key) {
Ok(()) => next.run(req).await,
Err(retry_after) => too_many_requests(retry_after),
Err(retry_after) => {
let resource = req.uri().path().to_string();
too_many_requests(retry_after, &resource)
}
}
}
fn too_many_requests(retry_after: u64) -> Response {
(
StatusCode::TOO_MANY_REQUESTS,
fn too_many_requests(retry_after: u64, resource: &str) -> Response {
let request_id = uuid::Uuid::new_v4().simple().to_string();
let body = myfsio_xml::response::rate_limit_exceeded_xml(resource, &request_id);
let mut response = (
StatusCode::SERVICE_UNAVAILABLE,
[
(header::CONTENT_TYPE, "application/xml".to_string()),
(header::RETRY_AFTER, retry_after.to_string()),
],
myfsio_xml::response::rate_limit_exceeded_xml(),
body,
)
.into_response()
.into_response();
if let Ok(value) = request_id.parse() {
response
.headers_mut()
.insert("x-amz-request-id", value);
}
response
}
fn rate_limit_key(req: &Request, num_trusted_proxies: usize) -> String {

372
crates/myfsio-server/tests/integration.rs
View File

@@ -163,7 +163,7 @@ async fn rate_limit_default_and_admin_are_independent() {
)
.await
.unwrap();
assert_eq!(second.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE);
assert!(second.headers().contains_key("retry-after"));
let admin_first = app
@@ -199,7 +199,7 @@ async fn rate_limit_default_and_admin_are_independent() {
)
.await
.unwrap();
assert_eq!(admin_third.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(admin_third.status(), StatusCode::SERVICE_UNAVAILABLE);
}
fn test_ui_state() -> (myfsio_server::state::AppState, tempfile::TempDir) {
@@ -2311,9 +2311,16 @@ async fn test_versioned_object_can_be_read_and_deleted_by_version_id() {
)
.unwrap();
let archived_version_id = list_body
.split("<Version>")
.skip(1)
.find(|block| block.contains("<IsLatest>false</IsLatest>"))
.and_then(|block| {
block
.split("<VersionId>")
.filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
.find(|id| *id != "null")
.nth(1)
.and_then(|s| s.split_once("</VersionId>").map(|(id, _)| id))
})
.filter(|id| *id != "null")
.expect("archived version id")
.to_string();
@@ -2506,6 +2513,352 @@ async fn test_versioned_put_and_delete_emit_version_headers_and_delete_markers()
);
}
#[tokio::test]
async fn test_consecutive_slashes_in_key_round_trip() {
let (app, _tmp) = test_app();
app.clone()
.oneshot(signed_request(Method::PUT, "/slashes-bucket", Body::empty()))
.await
.unwrap();
let put_ab = app
.clone()
.oneshot(signed_request(
Method::PUT,
"/slashes-bucket/a/b",
Body::from("single"),
))
.await
.unwrap();
assert_eq!(put_ab.status(), StatusCode::OK);
let put_double = app
.clone()
.oneshot(signed_request(
Method::PUT,
"/slashes-bucket/a//b",
Body::from("double"),
))
.await
.unwrap();
assert_eq!(put_double.status(), StatusCode::OK);
let put_triple = app
.clone()
.oneshot(signed_request(
Method::PUT,
"/slashes-bucket/a///b",
Body::from("triple"),
))
.await
.unwrap();
assert_eq!(put_triple.status(), StatusCode::OK);
let get_ab = app
.clone()
.oneshot(signed_request(
Method::GET,
"/slashes-bucket/a/b",
Body::empty(),
))
.await
.unwrap();
assert_eq!(get_ab.status(), StatusCode::OK);
let body_ab = get_ab.into_body().collect().await.unwrap().to_bytes();
assert_eq!(&body_ab[..], b"single");
let get_triple = app
.clone()
.oneshot(signed_request(
Method::GET,
"/slashes-bucket/a///b",
Body::empty(),
))
.await
.unwrap();
assert_eq!(get_triple.status(), StatusCode::OK);
let body_triple = get_triple.into_body().collect().await.unwrap().to_bytes();
assert_eq!(&body_triple[..], b"triple");
let list_resp = app
.oneshot(signed_request(
Method::GET,
"/slashes-bucket?list-type=2",
Body::empty(),
))
.await
.unwrap();
let list_body = String::from_utf8(
list_resp
.into_body()
.collect()
.await
.unwrap()
.to_bytes()
.to_vec(),
)
.unwrap();
assert!(
list_body.contains("<Key>a/b</Key>"),
"expected a/b in listing: {}",
list_body
);
assert!(
list_body.contains("<Key>a//b</Key>"),
"expected a//b in listing: {}",
list_body
);
assert!(
list_body.contains("<Key>a///b</Key>"),
"expected a///b in listing: {}",
list_body
);
}
#[tokio::test]
async fn test_delete_live_version_restores_previous_to_live_slot() {
let (app, _tmp) = test_app();
app.clone()
.oneshot(signed_request(Method::PUT, "/restore-bucket", Body::empty()))
.await
.unwrap();
app.clone()
.oneshot(
Request::builder()
.method(Method::PUT)
.uri("/restore-bucket?versioning")
.header("x-access-key", TEST_ACCESS_KEY)
.header("x-secret-key", TEST_SECRET_KEY)
.body(Body::from(
"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
))
.unwrap(),
)
.await
.unwrap();
let v1_resp = app
.clone()
.oneshot(signed_request(
Method::PUT,
"/restore-bucket/k",
Body::from("one"),
))
.await
.unwrap();
let v1 = v1_resp
.headers()
.get("x-amz-version-id")
.unwrap()
.to_str()
.unwrap()
.to_string();
let v2_resp = app
.clone()
.oneshot(signed_request(
Method::PUT,
"/restore-bucket/k",
Body::from("two"),
))
.await
.unwrap();
let v2 = v2_resp
.headers()
.get("x-amz-version-id")
.unwrap()
.to_str()
.unwrap()
.to_string();
assert_ne!(v1, v2);
let del = app
.clone()
.oneshot(signed_request(
Method::DELETE,
&format!("/restore-bucket/k?versionId={}", v2),
Body::empty(),
))
.await
.unwrap();
assert_eq!(del.status(), StatusCode::NO_CONTENT);
let get_live = app
.clone()
.oneshot(signed_request(
Method::GET,
"/restore-bucket/k",
Body::empty(),
))
.await
.unwrap();
assert_eq!(get_live.status(), StatusCode::OK);
let body = get_live.into_body().collect().await.unwrap().to_bytes();
assert_eq!(&body[..], b"one");
let get_v1 = app
.oneshot(signed_request(
Method::GET,
&format!("/restore-bucket/k?versionId={}", v1),
Body::empty(),
))
.await
.unwrap();
assert_eq!(get_v1.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_delete_active_delete_marker_restores_previous_to_live_slot() {
let (app, _tmp) = test_app();
app.clone()
.oneshot(signed_request(Method::PUT, "/undel-bucket", Body::empty()))
.await
.unwrap();
app.clone()
.oneshot(
Request::builder()
.method(Method::PUT)
.uri("/undel-bucket?versioning")
.header("x-access-key", TEST_ACCESS_KEY)
.header("x-secret-key", TEST_SECRET_KEY)
.body(Body::from(
"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
))
.unwrap(),
)
.await
.unwrap();
app.clone()
.oneshot(signed_request(
Method::PUT,
"/undel-bucket/k",
Body::from("only"),
))
.await
.unwrap();
let del = app
.clone()
.oneshot(signed_request(
Method::DELETE,
"/undel-bucket/k",
Body::empty(),
))
.await
.unwrap();
let dm_version = del
.headers()
.get("x-amz-version-id")
.unwrap()
.to_str()
.unwrap()
.to_string();
assert_eq!(
del.headers()
.get("x-amz-delete-marker")
.and_then(|v| v.to_str().ok()),
Some("true")
);
let shadowed = app
.clone()
.oneshot(signed_request(
Method::GET,
"/undel-bucket/k",
Body::empty(),
))
.await
.unwrap();
assert_eq!(shadowed.status(), StatusCode::NOT_FOUND);
let del_dm = app
.clone()
.oneshot(signed_request(
Method::DELETE,
&format!("/undel-bucket/k?versionId={}", dm_version),
Body::empty(),
))
.await
.unwrap();
assert_eq!(del_dm.status(), StatusCode::NO_CONTENT);
let restored = app
.oneshot(signed_request(
Method::GET,
"/undel-bucket/k",
Body::empty(),
))
.await
.unwrap();
assert_eq!(restored.status(), StatusCode::OK);
let body = restored.into_body().collect().await.unwrap().to_bytes();
assert_eq!(&body[..], b"only");
}
#[tokio::test]
async fn test_versioned_get_on_delete_marker_returns_method_not_allowed() {
let (app, _tmp) = test_app();
app.clone()
.oneshot(signed_request(Method::PUT, "/dm-bucket", Body::empty()))
.await
.unwrap();
app.clone()
.oneshot(
Request::builder()
.method(Method::PUT)
.uri("/dm-bucket?versioning")
.header("x-access-key", TEST_ACCESS_KEY)
.header("x-secret-key", TEST_SECRET_KEY)
.body(Body::from(
"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
))
.unwrap(),
)
.await
.unwrap();
app.clone()
.oneshot(signed_request(
Method::PUT,
"/dm-bucket/k",
Body::from("x"),
))
.await
.unwrap();
let del = app
.clone()
.oneshot(signed_request(
Method::DELETE,
"/dm-bucket/k",
Body::empty(),
))
.await
.unwrap();
let dm_version = del
.headers()
.get("x-amz-version-id")
.unwrap()
.to_str()
.unwrap()
.to_string();
let versioned = app
.oneshot(signed_request(
Method::GET,
&format!("/dm-bucket/k?versionId={}", dm_version),
Body::empty(),
))
.await
.unwrap();
assert_eq!(versioned.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn test_retention_is_enforced_when_deleting_archived_version() {
let (app, _tmp) = test_app();
@@ -2586,9 +2939,16 @@ async fn test_retention_is_enforced_when_deleting_archived_version() {
)
.unwrap();
let archived_version_id = list_body
.split("<Version>")
.skip(1)
.find(|block| block.contains("<IsLatest>false</IsLatest>"))
.and_then(|block| {
block
.split("<VersionId>")
.filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
.find(|id| *id != "null")
.nth(1)
.and_then(|s| s.split_once("</VersionId>").map(|(id, _)| id))
})
.filter(|id| *id != "null")
.expect("archived version id")
.to_string();

2
crates/myfsio-storage/src/error.rs
View File

@@ -50,7 +50,7 @@ impl From<StorageError> for S3Error {
S3Error::from_code(S3ErrorCode::NoSuchBucket).with_resource(format!("/{}", name))
}
StorageError::BucketAlreadyExists(name) => {
S3Error::from_code(S3ErrorCode::BucketAlreadyExists)
S3Error::from_code(S3ErrorCode::BucketAlreadyOwnedByYou)
.with_resource(format!("/{}", name))
}
StorageError::BucketNotEmpty(name) => {

424
crates/myfsio-storage/src/fs_backend.rs
View File

@@ -16,6 +16,55 @@ use std::time::Instant;
use tokio::io::AsyncReadExt;
use uuid::Uuid;
const EMPTY_SEGMENT_SENTINEL: &str = ".__myfsio_empty__";
fn fs_encode_key(key: &str) -> String {
if key.is_empty() {
return String::new();
}
let trailing = key.ends_with('/');
let body = if trailing { &key[..key.len() - 1] } else { key };
if body.is_empty() {
return if trailing { "/".to_string() } else { String::new() };
}
let encoded: Vec<String> = body
.split('/')
.map(|seg| {
if seg.is_empty() {
EMPTY_SEGMENT_SENTINEL.to_string()
} else {
seg.to_string()
}
})
.collect();
let mut result = encoded.join("/");
if trailing {
result.push('/');
}
result
}
fn fs_decode_key(rel_path: &str) -> String {
let normalized: String;
let input = if cfg!(windows) && rel_path.contains('\\') {
normalized = rel_path.replace('\\', "/");
normalized.as_str()
} else {
rel_path
};
input
.split('/')
.map(|seg| {
if seg == EMPTY_SEGMENT_SENTINEL {
""
} else {
seg
}
})
.collect::<Vec<_>>()
.join("/")
}
fn validate_list_prefix(prefix: &str) -> StorageResult<()> {
if prefix.contains('\0') {
return Err(StorageError::InvalidObjectKey(
@@ -88,7 +137,7 @@ fn path_is_within(candidate: &Path, root: &Path) -> bool {
}
}
type ListCacheEntry = (String, u64, f64, Option<String>);
type ListCacheEntry = (String, u64, f64, Option<String>, Option<String>);
#[derive(Clone, Default)]
struct ShallowCacheEntry {
@@ -213,13 +262,27 @@ impl FsStorageBackend {
fn object_path(&self, bucket_name: &str, object_key: &str) -> StorageResult<PathBuf> {
self.validate_key(object_key)?;
let encoded = fs_encode_key(object_key);
if object_key.ends_with('/') {
let trimmed = encoded.trim_end_matches('/');
Ok(self
.bucket_path(bucket_name)
.join(object_key)
.join(trimmed)
.join(DIR_MARKER_FILE))
} else {
Ok(self.bucket_path(bucket_name).join(object_key))
Ok(self.bucket_path(bucket_name).join(&encoded))
}
}
fn object_live_path(&self, bucket_name: &str, object_key: &str) -> PathBuf {
let encoded = fs_encode_key(object_key);
if object_key.ends_with('/') {
let trimmed = encoded.trim_end_matches('/');
self.bucket_path(bucket_name)
.join(trimmed)
.join(DIR_MARKER_FILE)
} else {
self.bucket_path(bucket_name).join(&encoded)
}
}
@@ -247,7 +310,8 @@ impl FsStorageBackend {
fn index_file_for_key(&self, bucket_name: &str, key: &str) -> (PathBuf, String) {
let meta_root = self.bucket_meta_root(bucket_name);
if key.ends_with('/') {
let trimmed = key.trim_end_matches('/');
let encoded = fs_encode_key(key);
let trimmed = encoded.trim_end_matches('/');
if trimmed.is_empty() {
return (meta_root.join(INDEX_FILE), DIR_MARKER_FILE.to_string());
}
@@ -256,13 +320,14 @@ impl FsStorageBackend {
DIR_MARKER_FILE.to_string(),
);
}
let key_path = Path::new(key);
let entry_name = key_path
let encoded = fs_encode_key(key);
let encoded_path = Path::new(&encoded);
let entry_name = encoded_path
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| key.to_string());
.unwrap_or_else(|| encoded.clone());
let parent = key_path.parent();
let parent = encoded_path.parent();
match parent {
Some(p) if p != Path::new("") && p != Path::new(".") => {
(meta_root.join(p).join(INDEX_FILE), entry_name)
@@ -304,6 +369,37 @@ impl FsStorageBackend {
out
}
fn load_dir_index_full_sync(
&self,
bucket_name: &str,
rel_dir: &Path,
) -> HashMap<String, (Option<String>, Option<String>)> {
let index_path = self.index_file_for_dir(bucket_name, rel_dir);
if !index_path.exists() {
return HashMap::new();
}
let Ok(text) = std::fs::read_to_string(&index_path) else {
return HashMap::new();
};
let Ok(index) = serde_json::from_str::<HashMap<String, Value>>(&text) else {
return HashMap::new();
};
let mut out = HashMap::with_capacity(index.len());
for (name, entry) in index {
let meta = entry.get("metadata").and_then(|m| m.as_object());
let etag = meta
.and_then(|m| m.get("__etag__"))
.and_then(|v| v.as_str())
.map(ToOwned::to_owned);
let version_id = meta
.and_then(|m| m.get("__version_id__"))
.and_then(|v| v.as_str())
.map(ToOwned::to_owned);
out.insert(name, (etag, version_id));
}
out
}
fn get_meta_index_lock(&self, index_path: &str) -> Arc<Mutex<()>> {
self.meta_index_locks
.entry(index_path.to_string())
@@ -344,7 +440,9 @@ impl FsStorageBackend {
}
fn version_dir(&self, bucket_name: &str, key: &str) -> PathBuf {
self.bucket_versions_root(bucket_name).join(key)
let encoded = fs_encode_key(key);
let trimmed = encoded.trim_end_matches('/');
self.bucket_versions_root(bucket_name).join(trimmed)
}
fn delete_markers_root(&self, bucket_name: &str) -> PathBuf {
@@ -352,8 +450,10 @@ impl FsStorageBackend {
}
fn delete_marker_path(&self, bucket_name: &str, key: &str) -> PathBuf {
let encoded = fs_encode_key(key);
let trimmed = encoded.trim_end_matches('/');
self.delete_markers_root(bucket_name)
.join(format!("{}.json", key))
.join(format!("{}.json", trimmed))
}
fn read_delete_marker_sync(
@@ -804,8 +904,7 @@ impl FsStorageBackend {
key: &str,
reason: &str,
) -> std::io::Result<(u64, Option<String>)> {
let bucket_path = self.bucket_path(bucket_name);
let source = bucket_path.join(key);
let source = self.object_live_path(bucket_name, key);
if !source.exists() {
return Ok((0, None));
}
@@ -845,6 +944,101 @@ impl FsStorageBackend {
Ok((source_size, Some(version_id)))
}
fn promote_latest_archived_to_live_sync(
&self,
bucket_name: &str,
key: &str,
) -> std::io::Result<Option<String>> {
let version_dir = self.version_dir(bucket_name, key);
if !version_dir.exists() {
return Ok(None);
}
let entries = match std::fs::read_dir(&version_dir) {
Ok(e) => e,
Err(_) => return Ok(None),
};
let mut candidates: Vec<(DateTime<Utc>, String, PathBuf, Value)> = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let Ok(content) = std::fs::read_to_string(&path) else {
continue;
};
let Ok(record) = serde_json::from_str::<Value>(&content) else {
continue;
};
if record
.get("is_delete_marker")
.and_then(Value::as_bool)
.unwrap_or(false)
{
continue;
}
let version_id = record
.get("version_id")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
if version_id.is_empty() {
continue;
}
let archived_at = record
.get("archived_at")
.and_then(Value::as_str)
.and_then(|s| DateTime::parse_from_rfc3339(s).ok())
.map(|d| d.with_timezone(&Utc))
.unwrap_or_else(Utc::now);
candidates.push((archived_at, version_id, path, record));
}
candidates.sort_by(|a, b| b.0.cmp(&a.0));
let Some((_, version_id, manifest_path, record)) = candidates.into_iter().next() else {
return Ok(None);
};
let (_, data_path) = self.version_record_paths(bucket_name, key, &version_id);
if !data_path.is_file() {
return Ok(None);
}
let live_path = self.object_live_path(bucket_name, key);
if let Some(parent) = live_path.parent() {
std::fs::create_dir_all(parent)?;
}
if live_path.exists() {
std::fs::remove_file(&live_path).ok();
}
std::fs::rename(&data_path, &live_path)?;
let mut meta: HashMap<String, String> = record
.get("metadata")
.and_then(Value::as_object)
.map(|m| {
m.iter()
.filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
.collect()
})
.unwrap_or_default();
meta.insert("__version_id__".to_string(), version_id.clone());
if !meta.contains_key("__etag__") {
if let Some(etag) = record.get("etag").and_then(Value::as_str) {
if !etag.is_empty() {
meta.insert("__etag__".to_string(), etag.to_string());
}
}
}
self.write_metadata_sync(bucket_name, key, &meta)?;
Self::safe_unlink(&manifest_path)?;
Self::cleanup_empty_parents(&manifest_path, &self.bucket_versions_root(bucket_name));
Ok(Some(version_id))
}
fn write_delete_marker_sync(
&self,
bucket_name: &str,
@@ -918,8 +1112,13 @@ impl FsStorageBackend {
self.require_bucket(bucket_name)?;
self.validate_key(key)?;
Self::validate_version_id(bucket_name, key, version_id)?;
if let Some(record_and_path) = self.try_live_version_record_sync(bucket_name, key, version_id) {
return Ok(record_and_path);
}
let (manifest_path, data_path) = self.version_record_paths(bucket_name, key, version_id);
if !manifest_path.is_file() || !data_path.is_file() {
if !manifest_path.is_file() {
return Err(StorageError::VersionNotFound {
bucket: bucket_name.to_string(),
key: key.to_string(),
@@ -929,9 +1128,64 @@ impl FsStorageBackend {
let content = std::fs::read_to_string(&manifest_path).map_err(StorageError::Io)?;
let record = serde_json::from_str::<Value>(&content).map_err(StorageError::Json)?;
let is_delete_marker = record
.get("is_delete_marker")
.and_then(Value::as_bool)
.unwrap_or(false);
if !is_delete_marker && !data_path.is_file() {
return Err(StorageError::VersionNotFound {
bucket: bucket_name.to_string(),
key: key.to_string(),
version_id: version_id.to_string(),
});
}
Ok((record, data_path))
}
fn try_live_version_record_sync(
&self,
bucket_name: &str,
key: &str,
version_id: &str,
) -> Option<(Value, PathBuf)> {
let live_path = self.object_live_path(bucket_name, key);
if !live_path.is_file() {
return None;
}
let metadata = self.read_metadata_sync(bucket_name, key);
let live_version = metadata.get("__version_id__")?.clone();
if live_version != version_id {
return None;
}
let file_meta = std::fs::metadata(&live_path).ok()?;
let mtime = file_meta
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs_f64())
.unwrap_or(0.0);
let archived_at = Utc
.timestamp_opt(mtime as i64, ((mtime % 1.0) * 1_000_000_000.0) as u32)
.single()
.unwrap_or_else(Utc::now);
let etag = metadata.get("__etag__").cloned().unwrap_or_default();
let mut meta_json = serde_json::Map::new();
for (k, v) in &metadata {
meta_json.insert(k.clone(), Value::String(v.clone()));
}
let record = serde_json::json!({
"version_id": live_version,
"key": key,
"size": file_meta.len(),
"archived_at": archived_at.to_rfc3339(),
"etag": etag,
"metadata": Value::Object(meta_json),
"reason": "current",
"is_delete_marker": false,
});
Some((record, live_path))
}
fn version_metadata_from_record(record: &Value) -> HashMap<String, String> {
record
.get("metadata")
@@ -1125,7 +1379,8 @@ impl FsStorageBackend {
let bucket_path = self.require_bucket(bucket_name)?;
let mut all_keys: Vec<ListCacheEntry> = Vec::new();
let mut dir_etag_cache: HashMap<PathBuf, HashMap<String, String>> = HashMap::new();
let mut dir_idx_cache: HashMap<PathBuf, HashMap<String, (Option<String>, Option<String>)>> =
HashMap::new();
let internal = INTERNAL_FOLDERS;
let bucket_str = bucket_path.to_string_lossy().to_string();
let bucket_prefix_len = bucket_str.len() + 1;
@@ -1150,12 +1405,16 @@ impl FsStorageBackend {
stack.push(entry.path().to_string_lossy().to_string());
} else if ft.is_file() {
let full_path = entry.path().to_string_lossy().to_string();
let mut key = full_path[bucket_prefix_len..].replace('\\', "/");
let mut fs_rel = full_path[bucket_prefix_len..].to_string();
#[cfg(windows)]
{
fs_rel = fs_rel.replace('\\', "/");
}
let is_dir_marker = name_str.as_ref() == DIR_MARKER_FILE;
if is_dir_marker {
key = key
fs_rel = fs_rel
.strip_suffix(DIR_MARKER_FILE)
.unwrap_or(&key)
.unwrap_or(&fs_rel)
.to_string();
}
if let Ok(meta) = entry.metadata() {
@@ -1166,20 +1425,23 @@ impl FsStorageBackend {
.map(|d| d.as_secs_f64())
.unwrap_or(0.0);
let rel_dir = Path::new(&key)
let rel_dir = Path::new(&fs_rel)
.parent()
.map(|p| p.to_path_buf())
.unwrap_or_default();
let etags = dir_etag_cache
.entry(rel_dir.clone())
.or_insert_with(|| self.load_dir_index_sync(bucket_name, &rel_dir));
let etag = if is_dir_marker {
None
let idx = dir_idx_cache.entry(rel_dir.clone()).or_insert_with(|| {
self.load_dir_index_full_sync(bucket_name, &rel_dir)
});
let (etag, version_id) = if is_dir_marker {
(None, None)
} else {
etags.get(name_str.as_ref()).cloned()
idx.get(name_str.as_ref())
.cloned()
.unwrap_or((None, None))
};
all_keys.push((key, meta.len(), mtime, etag));
let key = fs_decode_key(&fs_rel);
all_keys.push((key, meta.len(), mtime, etag, version_id));
}
}
}
@@ -1256,13 +1518,14 @@ impl FsStorageBackend {
let objects: Vec<ObjectMeta> = prefix_filter[start_idx..end_idx]
.iter()
.map(|(key, size, mtime, etag)| {
.map(|(key, size, mtime, etag, version_id)| {
let lm = Utc
.timestamp_opt(*mtime as i64, ((*mtime % 1.0) * 1_000_000_000.0) as u32)
.single()
.unwrap_or_else(Utc::now);
let mut obj = ObjectMeta::new(key.clone(), *size, lm);
obj.etag = etag.clone();
obj.version_id = version_id.clone();
obj
})
.collect();
@@ -1307,11 +1570,16 @@ impl FsStorageBackend {
let rel_dir_prefix = if rel_dir.as_os_str().is_empty() {
String::new()
} else {
let mut s = rel_dir.to_string_lossy().replace('\\', "/");
if !s.ends_with('/') {
s.push('/');
let mut s = rel_dir.to_string_lossy().into_owned();
#[cfg(windows)]
{
s = s.replace('\\', "/");
}
s
let mut decoded = fs_decode_key(&s);
if !decoded.ends_with('/') {
decoded.push('/');
}
decoded
};
let entries = std::fs::read_dir(&target_dir).map_err(StorageError::Io)?;
@@ -1328,6 +1596,7 @@ impl FsStorageBackend {
Err(_) => continue,
};
let display_name = fs_decode_key(&name_str);
if ft.is_dir() {
let subdir_path = entry.path();
let marker_path = subdir_path.join(DIR_MARKER_FILE);
@@ -1344,7 +1613,7 @@ impl FsStorageBackend {
.single()
.unwrap_or_else(Utc::now);
let mut obj = ObjectMeta::new(
format!("{}{}/", rel_dir_prefix, name_str),
format!("{}{}/", rel_dir_prefix, display_name),
meta.len(),
lm,
);
@@ -1352,12 +1621,12 @@ impl FsStorageBackend {
files.push(obj);
}
}
dirs.push(format!("{}{}{}", rel_dir_prefix, name_str, delimiter));
dirs.push(format!("{}{}{}", rel_dir_prefix, display_name, delimiter));
} else if ft.is_file() {
if name_str == DIR_MARKER_FILE {
continue;
}
let rel = format!("{}{}", rel_dir_prefix, name_str);
let rel = format!("{}{}", rel_dir_prefix, display_name);
if let Ok(meta) = entry.metadata() {
let mtime = meta
.modified()
@@ -1431,7 +1700,8 @@ impl FsStorageBackend {
PathBuf::new()
} else {
validate_list_prefix(&params.prefix)?;
let prefix_path = Path::new(&params.prefix);
let encoded_prefix = fs_encode_key(&params.prefix);
let prefix_path = Path::new(&encoded_prefix);
if params.prefix.ends_with(&params.delimiter) {
prefix_path.to_path_buf()
} else {
@@ -1524,8 +1794,8 @@ impl FsStorageBackend {
new_size: u64,
metadata: Option<HashMap<String, String>>,
) -> StorageResult<ObjectMeta> {
let bucket_path = self.require_bucket(bucket_name)?;
let destination = bucket_path.join(key);
self.require_bucket(bucket_name)?;
let destination = self.object_live_path(bucket_name, key);
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent).map_err(StorageError::Io)?;
}
@@ -1788,7 +2058,9 @@ impl crate::traits::StorageEngine for FsStorageBackend {
drop(writer);
let etag = format!("{:x}", hasher.finalize());
run_blocking(|| {
self.finalize_put_sync(bucket, key, &tmp_path, etag, total_size, metadata)
})
}
async fn get_object(
@@ -1796,6 +2068,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
bucket: &str,
key: &str,
) -> StorageResult<(ObjectMeta, AsyncReadStream)> {
let (obj, path) = run_blocking(|| -> StorageResult<(ObjectMeta, PathBuf)> {
self.require_bucket(bucket)?;
let path = self.object_path(bucket, key)?;
if !path.is_file() {
@@ -1839,6 +2112,8 @@ impl crate::traits::StorageEngine for FsStorageBackend {
.into_iter()
.filter(|(k, _)| !k.starts_with("__"))
.collect();
Ok((obj, path))
})?;
let file = tokio::fs::File::open(&path)
.await
@@ -1869,6 +2144,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
}
async fn head_object(&self, bucket: &str, key: &str) -> StorageResult<ObjectMeta> {
run_blocking(|| {
self.require_bucket(bucket)?;
let path = self.object_path(bucket, key)?;
if !path.is_file() {
@@ -1913,6 +2189,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
.filter(|(k, _)| !k.starts_with("__"))
.collect();
Ok(obj)
})
}
async fn get_object_version(
@@ -1922,6 +2199,15 @@ impl crate::traits::StorageEngine for FsStorageBackend {
version_id: &str,
) -> StorageResult<(ObjectMeta, AsyncReadStream)> {
let (record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
if record
.get("is_delete_marker")
.and_then(Value::as_bool)
.unwrap_or(false)
{
return Err(StorageError::MethodNotAllowed(
"The specified method is not allowed against a delete marker".to_string(),
));
}
let obj = self.object_meta_from_version_record(key, &record, &data_path)?;
let file = tokio::fs::File::open(&data_path)
.await
@@ -1936,7 +2222,16 @@ impl crate::traits::StorageEngine for FsStorageBackend {
key: &str,
version_id: &str,
) -> StorageResult<PathBuf> {
let (_record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
let (record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
if record
.get("is_delete_marker")
.and_then(Value::as_bool)
.unwrap_or(false)
{
return Err(StorageError::MethodNotAllowed(
"The specified method is not allowed against a delete marker".to_string(),
));
}
Ok(data_path)
}
@@ -1947,6 +2242,15 @@ impl crate::traits::StorageEngine for FsStorageBackend {
version_id: &str,
) -> StorageResult<ObjectMeta> {
let (record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
if record
.get("is_delete_marker")
.and_then(Value::as_bool)
.unwrap_or(false)
{
return Err(StorageError::MethodNotAllowed(
"The specified method is not allowed against a delete marker".to_string(),
));
}
self.object_meta_from_version_record(key, &record, &data_path)
}
@@ -1961,6 +2265,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
}
async fn delete_object(&self, bucket: &str, key: &str) -> StorageResult<DeleteOutcome> {
run_blocking(|| {
let bucket_path = self.require_bucket(bucket)?;
let path = self.object_path(bucket, key)?;
let versioning_enabled = self.read_bucket_config_sync(bucket).versioning_enabled;
@@ -2000,6 +2305,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
is_delete_marker: false,
existed: true,
})
})
}
async fn delete_object_version(
@@ -2008,9 +2314,30 @@ impl crate::traits::StorageEngine for FsStorageBackend {
key: &str,
version_id: &str,
) -> StorageResult<DeleteOutcome> {
self.require_bucket(bucket)?;
run_blocking(|| {
let bucket_path = self.require_bucket(bucket)?;
self.validate_key(key)?;
Self::validate_version_id(bucket, key, version_id)?;
let live_path = self.object_live_path(bucket, key);
if live_path.is_file() {
let metadata = self.read_metadata_sync(bucket, key);
if metadata.get("__version_id__").map(String::as_str) == Some(version_id) {
Self::safe_unlink(&live_path).map_err(StorageError::Io)?;
self.delete_metadata_sync(bucket, key)
.map_err(StorageError::Io)?;
Self::cleanup_empty_parents(&live_path, &bucket_path);
self.promote_latest_archived_to_live_sync(bucket, key)
.map_err(StorageError::Io)?;
self.invalidate_bucket_caches(bucket);
return Ok(DeleteOutcome {
version_id: Some(version_id.to_string()),
is_delete_marker: false,
existed: true,
});
}
}
let (manifest_path, data_path) = self.version_record_paths(bucket, key, version_id);
if !manifest_path.is_file() && !data_path.is_file() {
return Err(StorageError::VersionNotFound {
@@ -2035,20 +2362,28 @@ impl crate::traits::StorageEngine for FsStorageBackend {
let versions_root = self.bucket_versions_root(bucket);
Self::cleanup_empty_parents(&manifest_path, &versions_root);
let mut was_active_dm = false;
if is_delete_marker {
if let Some((dm_version_id, _)) = self.read_delete_marker_sync(bucket, key) {
if dm_version_id == version_id {
self.clear_delete_marker_sync(bucket, key);
was_active_dm = true;
}
}
}
if was_active_dm && !live_path.is_file() {
self.promote_latest_archived_to_live_sync(bucket, key)
.map_err(StorageError::Io)?;
}
self.invalidate_bucket_caches(bucket);
Ok(DeleteOutcome {
version_id: Some(version_id.to_string()),
is_delete_marker,
existed: true,
})
})
}
async fn copy_object(
@@ -2076,7 +2411,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
bucket: &str,
key: &str,
) -> StorageResult<HashMap<String, String>> {
Ok(self.read_metadata_sync(bucket, key))
Ok(run_blocking(|| self.read_metadata_sync(bucket, key)))
}
async fn put_object_metadata(
@@ -2085,6 +2420,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
key: &str,
metadata: &HashMap<String, String>,
) -> StorageResult<()> {
run_blocking(|| {
let mut entry = self.read_index_entry_sync(bucket, key).unwrap_or_default();
let meta_map: serde_json::Map<String, Value> = metadata
.iter()
@@ -2095,6 +2431,7 @@ impl crate::traits::StorageEngine for FsStorageBackend {
.map_err(StorageError::Io)?;
self.invalidate_bucket_caches(bucket);
Ok(())
})
}
async fn list_objects(
@@ -2607,7 +2944,14 @@ impl crate::traits::StorageEngine for FsStorageBackend {
let fallback_key = path
.parent()
.and_then(|parent| parent.strip_prefix(&root).ok())
.map(|rel| rel.to_string_lossy().replace('\\', "/"))
.map(|rel| {
let mut s = rel.to_string_lossy().into_owned();
#[cfg(windows)]
{
s = s.replace('\\', "/");
}
fs_decode_key(&s)
})
.unwrap_or_default();
let info = self.version_info_from_record(&fallback_key, &record);
if prefix.is_some_and(|value| !info.key.starts_with(value)) {

6
crates/myfsio-storage/src/validation.rs
View File

@@ -47,6 +47,7 @@ pub fn validate_object_key(
normalized.split('/').collect()
};
for part in &parts {
if part.is_empty() {
continue;
@@ -105,7 +106,10 @@ pub fn validate_object_key(
}
for part in &non_empty_parts {
if *part == ".__myfsio_dirobj__" || part.starts_with("_index.json") {
if *part == ".__myfsio_dirobj__"
|| *part == ".__myfsio_empty__"
|| part.starts_with("_index.json")
{
return Some("Object key segment uses a reserved internal name".to_string());
}
}

41
crates/myfsio-xml/src/request.rs
View File

@@ -86,6 +86,11 @@ pub fn parse_complete_multipart_upload(xml: &str) -> Result<CompleteMultipartUpl
}
pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
let trimmed = xml.trim();
if trimmed.is_empty() {
return Err("Request body is empty".to_string());
}
let mut reader = Reader::from_str(xml);
let mut result = DeleteObjectsRequest::default();
let mut buf = Vec::new();
@@ -93,18 +98,43 @@ pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
let mut current_key: Option<String> = None;
let mut current_version_id: Option<String> = None;
let mut in_object = false;
let mut saw_delete_root = false;
let mut first_element_seen = false;
loop {
match reader.read_event_into(&mut buf) {
let event = reader.read_event_into(&mut buf);
match event {
Ok(Event::Start(ref e)) => {
let name = String::from_utf8_lossy(e.name().as_ref()).to_string();
current_tag = name.clone();
if name == "Object" {
if !first_element_seen {
first_element_seen = true;
if name != "Delete" {
return Err(format!(
"Expected <Delete> root element, found <{}>",
name
));
}
saw_delete_root = true;
} else if name == "Object" {
in_object = true;
current_key = None;
current_version_id = None;
}
}
Ok(Event::Empty(ref e)) => {
let name = String::from_utf8_lossy(e.name().as_ref()).to_string();
if !first_element_seen {
first_element_seen = true;
if name != "Delete" {
return Err(format!(
"Expected <Delete> root element, found <{}>",
name
));
}
saw_delete_root = true;
}
}
Ok(Event::Text(ref e)) => {
let text = e.unescape().map_err(|e| e.to_string())?.to_string();
match current_tag.as_str() {
@@ -139,6 +169,13 @@ pub fn parse_delete_objects(xml: &str) -> Result<DeleteObjectsRequest, String> {
buf.clear();
}
if !saw_delete_root {
return Err("Expected <Delete> root element".to_string());
}
if result.objects.is_empty() {
return Err("Delete request must contain at least one <Object>".to_string());
}
Ok(result)
}

17
crates/myfsio-xml/src/response.rs
View File

@@ -8,10 +8,21 @@ pub fn format_s3_datetime(dt: &DateTime<Utc>) -> String {
dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string()
}
pub fn rate_limit_exceeded_xml() -> String {
pub fn rate_limit_exceeded_xml(resource: &str, request_id: &str) -> String {
format!(
"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
<Error><Code>SlowDown</Code><Message>Rate limit exceeded</Message><Resource></Resource><RequestId></RequestId></Error>"
.to_string()
<Error><Code>SlowDown</Code><Message>Please reduce your request rate</Message><Resource>{}</Resource><RequestId>{}</RequestId></Error>",
xml_escape(resource),
xml_escape(request_id),
)
}
fn xml_escape(s: &str) -> String {
s.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&apos;")
}
pub fn list_buckets_xml(owner_id: &str, owner_name: &str, buckets: &[BucketMeta]) -> String {