From 9ec5797919c1928da1edc3bf18865f7cef7219e5 Mon Sep 17 00:00:00 2001
From: kqjy <jun@jzwsite.com>
Date: Wed, 22 Apr 2026 00:12:22 +0800
Subject: [PATCH] Applied max-keys to combined current + archived
 ListObjectVersions output and reports truncation

---
 README.md                                     |   2 +-
 docs.md                                       |   4 +-
 python/README.md                              |  14 +
 python/app/__init__.py                        |   2 +-
 python/run.py                                 |   7 +
 python/tests/test_website_hosting.py          |   2 +-
 rust/myfsio-engine/Cargo.lock                 |   2 +
 .../crates/myfsio-common/src/error.rs         |   8 +
 .../crates/myfsio-common/src/types.rs         |   2 +
 .../crates/myfsio-server/Cargo.toml           |   2 +
 .../myfsio-server/src/handlers/config.rs      |  68 +-
 .../crates/myfsio-server/src/handlers/mod.rs  | 848 ++++++++++++++++--
 .../myfsio-server/src/handlers/ui_api.rs      |   1 +
 .../crates/myfsio-server/src/lib.rs           |   2 +-
 .../myfsio-server/src/middleware/auth.rs      |  99 +-
 .../crates/myfsio-server/tests/integration.rs | 441 ++++++++-
 .../crates/myfsio-storage/src/error.rs        |  12 +
 .../crates/myfsio-storage/src/fs_backend.rs   | 310 ++++++-
 .../crates/myfsio-storage/src/traits.rs       |  41 +
 19 files changed, 1750 insertions(+), 117 deletions(-)
 create mode 100644 python/README.md

diff --git a/README.md b/README.md
index b9a6318..8d59b81 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
 
 MyFSIO is an S3-compatible object storage server with a Rust runtime and a filesystem-backed storage engine. The active server lives under `rust/myfsio-engine` and serves both the S3 API and the built-in web UI from a single process.
 
-The repository still contains a `python/` tree, but you do not need Python to run the current server.
+The `python/` implementation is deprecated as of 2026-04-21. It remains in the repository for migration reference and legacy tests, but new development and supported runtime usage should target the Rust server.
 
 ## Features
 
diff --git a/docs.md b/docs.md
index b2438b7..7b236a1 100644
--- a/docs.md
+++ b/docs.md
@@ -2,6 +2,8 @@
 
 This document describes the current Rust server in `rust/myfsio-engine`. It replaces the older Python-oriented runbook.
 
+The `python/` implementation is deprecated as of 2026-04-21. It is retained for migration reference and legacy validation only; production usage and new development should use the Rust server.
+
 ## 1. What Changed
 
 The active runtime is now Rust:
@@ -11,7 +13,7 @@ The active runtime is now Rust:
 - The main development workflow is `cargo run -p myfsio-server --`.
 - API-only mode is controlled with `UI_ENABLED=false`.
 
-The `python/` directory may still contain older implementation code, templates, and tests, but it is not required to run the current server.
+The deprecated `python/` directory may still contain older implementation code, templates, and tests, but it is not required to run the current server.
 
 ## 2. Quick Start
 
diff --git a/python/README.md b/python/README.md
new file mode 100644
index 0000000..853b88a
--- /dev/null
+++ b/python/README.md
@@ -0,0 +1,14 @@
+# Deprecated Python Implementation
+
+The Python implementation of MyFSIO is deprecated as of 2026-04-21.
+
+The supported server runtime now lives in `../rust/myfsio-engine` and serves the S3 API and web UI from the Rust `myfsio-server` binary. Keep this tree for migration reference, compatibility checks, and legacy tests only.
+
+For normal development and operations, run:
+
+```bash
+cd ../rust/myfsio-engine
+cargo run -p myfsio-server --
+```
+
+Do not add new product features to the Python implementation unless they are needed to unblock a migration or compare behavior with the Rust server.
diff --git a/python/app/__init__.py b/python/app/__init__.py
index f37a33a..fc465fb 100644
--- a/python/app/__init__.py
+++ b/python/app/__init__.py
@@ -720,7 +720,7 @@ def _configure_logging(app: Flask) -> None:
 
     def _website_error_response(status_code, message):
         if status_code == 404:
-            body = "404 page not found"
+            body = "<h1>404 page not found</h1>"
         else:
             body = f"{status_code} {message}"
         return Response(body, status=status_code, mimetype="text/html")
diff --git a/python/run.py b/python/run.py
index 40ab540..1e3c0c0 100644
--- a/python/run.py
+++ b/python/run.py
@@ -28,6 +28,11 @@ from app.config import AppConfig
 from app.iam import IamService, IamError, ALLOWED_ACTIONS, _derive_fernet_key
 from app.version import get_version
 
+PYTHON_DEPRECATION_MESSAGE = (
+    "The Python MyFSIO runtime is deprecated as of 2026-04-21. "
+    "Use the Rust server in rust/myfsio-engine for supported development and production usage."
+)
+
 
 def _server_host() -> str:
     """Return the bind host for API and UI servers."""
@@ -233,6 +238,8 @@ if __name__ == "__main__":
     parser.add_argument("--version", action="version", version=f"MyFSIO {get_version()}")
     args = parser.parse_args()
 
+    warnings.warn(PYTHON_DEPRECATION_MESSAGE, DeprecationWarning, stacklevel=1)
+
     if args.reset_cred or args.mode == "reset-cred":
         reset_credentials()
         sys.exit(0)
diff --git a/python/tests/test_website_hosting.py b/python/tests/test_website_hosting.py
index ae000b9..a12e65f 100644
--- a/python/tests/test_website_hosting.py
+++ b/python/tests/test_website_hosting.py
@@ -439,4 +439,4 @@ class TestWebsiteServing:
         store.set_mapping("noerr.example.com", "no-err")
         resp = website_client.get("/missing.html", headers={"Host": "noerr.example.com"})
         assert resp.status_code == 404
-        assert resp.data == b"404 page not found"
+        assert resp.data == b"<h1>404 page not found</h1>"
diff --git a/rust/myfsio-engine/Cargo.lock b/rust/myfsio-engine/Cargo.lock
index d8c061c..1c0817f 100644
--- a/rust/myfsio-engine/Cargo.lock
+++ b/rust/myfsio-engine/Cargo.lock
@@ -2707,6 +2707,7 @@ dependencies = [
  "futures",
  "http-body-util",
  "hyper 1.9.0",
+ "md-5 0.10.6",
  "mime_guess",
  "multer",
  "myfsio-auth",
@@ -2723,6 +2724,7 @@ dependencies = [
  "roxmltree",
  "serde",
  "serde_json",
+ "sha2 0.10.9",
  "subtle",
  "sysinfo",
  "tempfile",
diff --git a/rust/myfsio-engine/crates/myfsio-common/src/error.rs b/rust/myfsio-engine/crates/myfsio-common/src/error.rs
index 64e9110..5785d5f 100644
--- a/rust/myfsio-engine/crates/myfsio-common/src/error.rs
+++ b/rust/myfsio-engine/crates/myfsio-common/src/error.rs
@@ -3,6 +3,7 @@ use std::fmt;
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum S3ErrorCode {
     AccessDenied,
+    BadDigest,
     BucketAlreadyExists,
     BucketNotEmpty,
     EntityTooLarge,
@@ -14,6 +15,7 @@ pub enum S3ErrorCode {
     InvalidPolicyDocument,
     InvalidRange,
     InvalidRequest,
+    InvalidTag,
     MalformedXML,
     MethodNotAllowed,
     NoSuchBucket,
@@ -32,6 +34,7 @@ impl S3ErrorCode {
     pub fn http_status(&self) -> u16 {
         match self {
             Self::AccessDenied => 403,
+            Self::BadDigest => 400,
             Self::BucketAlreadyExists => 409,
             Self::BucketNotEmpty => 409,
             Self::EntityTooLarge => 413,
@@ -43,6 +46,7 @@ impl S3ErrorCode {
             Self::InvalidPolicyDocument => 400,
             Self::InvalidRange => 416,
             Self::InvalidRequest => 400,
+            Self::InvalidTag => 400,
             Self::MalformedXML => 400,
             Self::MethodNotAllowed => 405,
             Self::NoSuchBucket => 404,
@@ -61,6 +65,7 @@ impl S3ErrorCode {
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::AccessDenied => "AccessDenied",
+            Self::BadDigest => "BadDigest",
             Self::BucketAlreadyExists => "BucketAlreadyExists",
             Self::BucketNotEmpty => "BucketNotEmpty",
             Self::EntityTooLarge => "EntityTooLarge",
@@ -72,6 +77,7 @@ impl S3ErrorCode {
             Self::InvalidPolicyDocument => "InvalidPolicyDocument",
             Self::InvalidRange => "InvalidRange",
             Self::InvalidRequest => "InvalidRequest",
+            Self::InvalidTag => "InvalidTag",
             Self::MalformedXML => "MalformedXML",
             Self::MethodNotAllowed => "MethodNotAllowed",
             Self::NoSuchBucket => "NoSuchBucket",
@@ -90,6 +96,7 @@ impl S3ErrorCode {
     pub fn default_message(&self) -> &'static str {
         match self {
             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::BucketNotEmpty => "The bucket you tried to delete is not empty",
             Self::EntityTooLarge => "Your proposed upload exceeds the maximum allowed size",
@@ -101,6 +108,7 @@ impl S3ErrorCode {
             Self::InvalidPolicyDocument => "The content of the form does not meet the conditions specified in the policy document",
             Self::InvalidRange => "The requested range is not satisfiable",
             Self::InvalidRequest => "Invalid request",
+            Self::InvalidTag => "The Tagging header is invalid",
             Self::MalformedXML => "The XML you provided was not well-formed",
             Self::MethodNotAllowed => "The specified method is not allowed against this resource",
             Self::NoSuchBucket => "The specified bucket does not exist",
diff --git a/rust/myfsio-engine/crates/myfsio-common/src/types.rs b/rust/myfsio-engine/crates/myfsio-common/src/types.rs
index a07d565..91195f7 100644
--- a/rust/myfsio-engine/crates/myfsio-common/src/types.rs
+++ b/rust/myfsio-engine/crates/myfsio-common/src/types.rs
@@ -112,6 +112,8 @@ pub struct VersionInfo {
     pub last_modified: DateTime<Utc>,
     pub etag: Option<String>,
     pub is_latest: bool,
+    #[serde(default)]
+    pub is_delete_marker: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/rust/myfsio-engine/crates/myfsio-server/Cargo.toml b/rust/myfsio-engine/crates/myfsio-server/Cargo.toml
index 6a1092b..27d8f75 100644
--- a/rust/myfsio-engine/crates/myfsio-server/Cargo.toml
+++ b/rust/myfsio-engine/crates/myfsio-server/Cargo.toml
@@ -10,6 +10,7 @@ myfsio-crypto = { path = "../myfsio-crypto" }
 myfsio-storage = { path = "../myfsio-storage" }
 myfsio-xml = { path = "../myfsio-xml" }
 base64 = { workspace = true }
+md-5 = { workspace = true }
 axum = { workspace = true }
 tokio = { workspace = true }
 tower = { workspace = true }
@@ -29,6 +30,7 @@ percent-encoding = { workspace = true }
 quick-xml = { workspace = true }
 mime_guess = "2"
 crc32fast = { workspace = true }
+sha2 = { workspace = true }
 duckdb = { workspace = true }
 roxmltree = "0.20"
 parking_lot = { workspace = true }
diff --git a/rust/myfsio-engine/crates/myfsio-server/src/handlers/config.rs b/rust/myfsio-engine/crates/myfsio-server/src/handlers/config.rs
index 973c4db..5bb6163 100644
--- a/rust/myfsio-engine/crates/myfsio-server/src/handlers/config.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/src/handlers/config.rs
@@ -1038,7 +1038,12 @@ fn s3_error_response(code: S3ErrorCode, message: &str, status: StatusCode) -> Re
     (status, [("content-type", "application/xml")], err.to_xml()).into_response()
 }
 
-pub async fn list_object_versions(state: &AppState, bucket: &str) -> Response {
+pub async fn list_object_versions(
+    state: &AppState,
+    bucket: &str,
+    prefix: Option<&str>,
+    max_keys: usize,
+) -> Response {
     match state.storage.list_buckets().await {
         Ok(buckets) => {
             if !buckets.iter().any(|b| b.name == bucket) {
@@ -1050,13 +1055,24 @@ pub async fn list_object_versions(state: &AppState, bucket: &str) -> Response {
         Err(e) => return storage_err(e),
     }
 
+    let fetch_limit = max_keys.saturating_add(1).max(1);
     let params = myfsio_common::types::ListParams {
-        max_keys: 1000,
+        max_keys: fetch_limit,
+        prefix: prefix.map(ToOwned::to_owned),
         ..Default::default()
     };
 
-    let objects = match state.storage.list_objects(bucket, &params).await {
-        Ok(result) => result.objects,
+    let object_result = match state.storage.list_objects(bucket, &params).await {
+        Ok(result) => result,
+        Err(e) => return storage_err(e),
+    };
+    let objects = object_result.objects;
+    let archived_versions = match state
+        .storage
+        .list_bucket_object_versions(bucket, prefix)
+        .await
+    {
+        Ok(versions) => versions,
         Err(e) => return storage_err(e),
     };
 
@@ -1064,11 +1080,24 @@ pub async fn list_object_versions(state: &AppState, bucket: &str) -> Response {
         "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
         <ListVersionsResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">",
     );
-    xml.push_str(&format!("<Name>{}</Name>", bucket));
+    xml.push_str(&format!("<Name>{}</Name>", xml_escape(bucket)));
+    xml.push_str(&format!(
+        "<Prefix>{}</Prefix>",
+        xml_escape(prefix.unwrap_or(""))
+    ));
+    xml.push_str(&format!("<MaxKeys>{}</MaxKeys>", max_keys));
 
-    for obj in &objects {
+    let current_count = objects.len().min(max_keys);
+    let remaining = max_keys.saturating_sub(current_count);
+    let archived_count = archived_versions.len().min(remaining);
+    let is_truncated = object_result.is_truncated
+        || objects.len() > current_count
+        || archived_versions.len() > archived_count;
+    xml.push_str(&format!("<IsTruncated>{}</IsTruncated>", is_truncated));
+
+    for obj in objects.iter().take(current_count) {
         xml.push_str("<Version>");
-        xml.push_str(&format!("<Key>{}</Key>", obj.key));
+        xml.push_str(&format!("<Key>{}</Key>", xml_escape(&obj.key)));
         xml.push_str("<VersionId>null</VersionId>");
         xml.push_str("<IsLatest>true</IsLatest>");
         xml.push_str(&format!(
@@ -1076,9 +1105,32 @@ pub async fn list_object_versions(state: &AppState, bucket: &str) -> Response {
             myfsio_xml::response::format_s3_datetime(&obj.last_modified)
         ));
         if let Some(ref etag) = obj.etag {
-            xml.push_str(&format!("<ETag>\"{}\"</ETag>", etag));
+            xml.push_str(&format!("<ETag>\"{}\"</ETag>", xml_escape(etag)));
         }
         xml.push_str(&format!("<Size>{}</Size>", obj.size));
+        xml.push_str(&format!(
+            "<StorageClass>{}</StorageClass>",
+            xml_escape(obj.storage_class.as_deref().unwrap_or("STANDARD"))
+        ));
+        xml.push_str("</Version>");
+    }
+
+    for version in archived_versions.iter().take(archived_count) {
+        xml.push_str("<Version>");
+        xml.push_str(&format!("<Key>{}</Key>", xml_escape(&version.key)));
+        xml.push_str(&format!(
+            "<VersionId>{}</VersionId>",
+            xml_escape(&version.version_id)
+        ));
+        xml.push_str("<IsLatest>false</IsLatest>");
+        xml.push_str(&format!(
+            "<LastModified>{}</LastModified>",
+            myfsio_xml::response::format_s3_datetime(&version.last_modified)
+        ));
+        if let Some(ref etag) = version.etag {
+            xml.push_str(&format!("<ETag>\"{}\"</ETag>", xml_escape(etag)));
+        }
+        xml.push_str(&format!("<Size>{}</Size>", version.size));
         xml.push_str("<StorageClass>STANDARD</StorageClass>");
         xml.push_str("</Version>");
     }
diff --git a/rust/myfsio-engine/crates/myfsio-server/src/handlers/mod.rs b/rust/myfsio-engine/crates/myfsio-server/src/handlers/mod.rs
index de6f02e..45d1b51 100644
--- a/rust/myfsio-engine/crates/myfsio-server/src/handlers/mod.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/src/handlers/mod.rs
@@ -13,10 +13,13 @@ use axum::body::Body;
 use axum::extract::{Path, Query, State};
 use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
-use base64::engine::general_purpose::URL_SAFE;
+use base64::engine::general_purpose::{STANDARD, URL_SAFE};
 use base64::Engine;
 use chrono::{DateTime, Utc};
+use md5::Md5;
+use percent_encoding::percent_decode_str;
 use serde_json::json;
+use sha2::{Digest, Sha256};
 
 use myfsio_common::error::{S3Error, S3ErrorCode};
 use myfsio_common::types::PartInfo;
@@ -90,7 +93,44 @@ async fn ensure_object_lock_allows_write(
     }
 }
 
-pub async fn list_buckets(State(state): State<AppState>) -> Response {
+async fn ensure_object_version_lock_allows_delete(
+    state: &AppState,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    headers: &HeaderMap,
+) -> Result<(), Response> {
+    let metadata = match state
+        .storage
+        .get_object_version_metadata(bucket, key, version_id)
+        .await
+    {
+        Ok(metadata) => metadata,
+        Err(err) => return Err(storage_err_response(err)),
+    };
+    let bypass_governance = headers
+        .get("x-amz-bypass-governance-retention")
+        .and_then(|value| value.to_str().ok())
+        .map(|value| value.eq_ignore_ascii_case("true"))
+        .unwrap_or(false);
+    if let Err(message) = object_lock::can_delete_object(&metadata, bypass_governance) {
+        return Err(s3_error_response(S3Error::new(
+            S3ErrorCode::AccessDenied,
+            message,
+        )));
+    }
+    Ok(())
+}
+
+pub async fn list_buckets(
+    State(state): State<AppState>,
+    Query(query): Query<BucketQuery>,
+    headers: HeaderMap,
+) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        return get_bucket(State(state), Path(host_bucket), Query(query), headers).await;
+    }
+
     match state.storage.list_buckets().await {
         Ok(buckets) => {
             let xml = myfsio_xml::response::list_buckets_xml("myfsio", "myfsio", &buckets);
@@ -117,8 +157,22 @@ pub async fn create_bucket(
     State(state): State<AppState>,
     Path(bucket): Path<String>,
     Query(query): Query<BucketQuery>,
+    headers: HeaderMap,
     body: Body,
 ) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        if host_bucket != bucket {
+            return put_object(
+                State(state),
+                Path((host_bucket, bucket)),
+                Query(ObjectQuery::default()),
+                headers,
+                body,
+            )
+            .await;
+        }
+    }
+
     if query.quota.is_some() {
         return config::put_quota(&state, &bucket, body).await;
     }
@@ -205,11 +259,41 @@ pub struct BucketQuery {
     pub versions: Option<String>,
 }
 
+async fn virtual_host_bucket_from_headers(state: &AppState, headers: &HeaderMap) -> Option<String> {
+    let host = headers
+        .get("host")
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.split(':').next())?
+        .trim()
+        .to_ascii_lowercase();
+    let (candidate, _) = host.split_once('.')?;
+    if myfsio_storage::validation::validate_bucket_name(candidate).is_some() {
+        return None;
+    }
+    match state.storage.bucket_exists(candidate).await {
+        Ok(true) => Some(candidate.to_string()),
+        _ => None,
+    }
+}
+
 pub async fn get_bucket(
     State(state): State<AppState>,
     Path(bucket): Path<String>,
     Query(query): Query<BucketQuery>,
+    headers: HeaderMap,
 ) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        if host_bucket != bucket {
+            return get_object(
+                State(state),
+                Path((host_bucket, bucket)),
+                Query(ObjectQuery::default()),
+                headers,
+            )
+            .await;
+        }
+    }
+
     if !matches!(state.storage.bucket_exists(&bucket).await, Ok(true)) {
         return storage_err_response(myfsio_storage::error::StorageError::BucketNotFound(bucket));
     }
@@ -260,7 +344,13 @@ pub async fn get_bucket(
         return config::get_logging(&state, &bucket).await;
     }
     if query.versions.is_some() {
-        return config::list_object_versions(&state, &bucket).await;
+        return config::list_object_versions(
+            &state,
+            &bucket,
+            query.prefix.as_deref(),
+            query.max_keys.unwrap_or(1000),
+        )
+        .await;
     }
     if query.uploads.is_some() {
         return list_multipart_uploads_handler(&state, &bucket).await;
@@ -408,6 +498,19 @@ pub async fn post_bucket(
     headers: HeaderMap,
     body: Body,
 ) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        if host_bucket != bucket {
+            return post_object(
+                State(state),
+                Path((host_bucket, bucket)),
+                Query(ObjectQuery::default()),
+                headers,
+                body,
+            )
+            .await;
+        }
+    }
+
     if query.delete.is_some() {
         return delete_objects_handler(&state, &bucket, body).await;
     }
@@ -425,7 +528,20 @@ pub async fn delete_bucket(
     State(state): State<AppState>,
     Path(bucket): Path<String>,
     Query(query): Query<BucketQuery>,
+    headers: HeaderMap,
 ) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        if host_bucket != bucket {
+            return delete_object(
+                State(state),
+                Path((host_bucket, bucket)),
+                Query(ObjectQuery::default()),
+                headers,
+            )
+            .await;
+        }
+    }
+
     if query.quota.is_some() {
         return config::delete_quota(&state, &bucket).await;
     }
@@ -466,7 +582,23 @@ pub async fn delete_bucket(
     }
 }
 
-pub async fn head_bucket(State(state): State<AppState>, Path(bucket): Path<String>) -> Response {
+pub async fn head_bucket(
+    State(state): State<AppState>,
+    Path(bucket): Path<String>,
+    headers: HeaderMap,
+) -> Response {
+    if let Some(host_bucket) = virtual_host_bucket_from_headers(&state, &headers).await {
+        if host_bucket != bucket {
+            return head_object(
+                State(state),
+                Path((host_bucket, bucket)),
+                Query(ObjectQuery::default()),
+                headers,
+            )
+            .await;
+        }
+    }
+
     match state.storage.bucket_exists(&bucket).await {
         Ok(true) => {
             let mut headers = HeaderMap::new();
@@ -489,6 +621,8 @@ pub struct ObjectQuery {
     pub upload_id: Option<String>,
     #[serde(rename = "partNumber")]
     pub part_number: Option<u32>,
+    #[serde(rename = "versionId")]
+    pub version_id: Option<String>,
     pub tagging: Option<String>,
     pub acl: Option<String>,
     pub retention: Option<String>,
@@ -583,6 +717,314 @@ fn insert_content_type(headers: &mut HeaderMap, key: &str, explicit: Option<&str
     }
 }
 
+fn internal_header_pairs() -> &'static [(&'static str, &'static str, &'static str)] {
+    &[
+        ("cache-control", "__cache_control__", "cache-control"),
+        (
+            "content-disposition",
+            "__content_disposition__",
+            "content-disposition",
+        ),
+        (
+            "content-language",
+            "__content_language__",
+            "content-language",
+        ),
+        (
+            "content-encoding",
+            "__content_encoding__",
+            "content-encoding",
+        ),
+        ("expires", "__expires__", "expires"),
+        (
+            "x-amz-website-redirect-location",
+            "__website_redirect_location__",
+            "x-amz-website-redirect-location",
+        ),
+    ]
+}
+
+fn decoded_content_encoding(value: &str) -> Option<String> {
+    let filtered: Vec<&str> = value
+        .split(',')
+        .map(str::trim)
+        .filter(|part| !part.is_empty() && !part.eq_ignore_ascii_case("aws-chunked"))
+        .collect();
+    if filtered.is_empty() {
+        None
+    } else {
+        Some(filtered.join(", "))
+    }
+}
+
+fn insert_standard_object_metadata(
+    headers: &HeaderMap,
+    metadata: &mut HashMap<String, String>,
+) -> Result<(), Response> {
+    for (request_header, metadata_key, _) in internal_header_pairs() {
+        if let Some(value) = headers.get(*request_header).and_then(|v| v.to_str().ok()) {
+            if *request_header == "content-encoding" {
+                if let Some(decoded_encoding) = decoded_content_encoding(value) {
+                    metadata.insert((*metadata_key).to_string(), decoded_encoding);
+                }
+            } else {
+                metadata.insert((*metadata_key).to_string(), value.to_string());
+            }
+        }
+    }
+    if let Some(value) = headers
+        .get("x-amz-storage-class")
+        .and_then(|v| v.to_str().ok())
+    {
+        metadata.insert("__storage_class__".to_string(), value.to_ascii_uppercase());
+    }
+
+    if let Some(value) = headers
+        .get("x-amz-object-lock-legal-hold")
+        .and_then(|v| v.to_str().ok())
+    {
+        object_lock::set_legal_hold(metadata, value.eq_ignore_ascii_case("ON"));
+    }
+
+    let retention_mode = headers
+        .get("x-amz-object-lock-mode")
+        .and_then(|v| v.to_str().ok());
+    let retain_until = headers
+        .get("x-amz-object-lock-retain-until-date")
+        .and_then(|v| v.to_str().ok());
+    if let (Some(mode), Some(retain_until)) = (retention_mode, retain_until) {
+        let mode = match mode.to_ascii_uppercase().as_str() {
+            "GOVERNANCE" => object_lock::RetentionMode::GOVERNANCE,
+            "COMPLIANCE" => object_lock::RetentionMode::COMPLIANCE,
+            _ => {
+                return Err(s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidArgument,
+                    "Invalid x-amz-object-lock-mode",
+                )))
+            }
+        };
+        let retain_until_date = DateTime::parse_from_rfc3339(retain_until)
+            .map(|value| value.with_timezone(&Utc))
+            .map_err(|_| {
+                s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidArgument,
+                    "Invalid x-amz-object-lock-retain-until-date",
+                ))
+            })?;
+        object_lock::set_object_retention(
+            metadata,
+            &object_lock::ObjectLockRetention {
+                mode,
+                retain_until_date,
+            },
+        )
+        .map_err(|message| {
+            s3_error_response(S3Error::new(S3ErrorCode::InvalidArgument, message))
+        })?;
+    }
+    Ok(())
+}
+
+fn apply_stored_response_headers(headers: &mut HeaderMap, metadata: &HashMap<String, String>) {
+    for (_, metadata_key, response_header) in internal_header_pairs() {
+        if let Some(value) = metadata
+            .get(*metadata_key)
+            .and_then(|value| value.parse().ok())
+        {
+            headers.insert(*response_header, value);
+        }
+    }
+    if let Some(value) = metadata
+        .get("__storage_class__")
+        .and_then(|value| value.parse().ok())
+    {
+        headers.insert("x-amz-storage-class", value);
+    }
+}
+
+fn apply_user_metadata(headers: &mut HeaderMap, metadata: &HashMap<String, String>) {
+    for (k, v) in metadata {
+        if let Ok(header_val) = v.parse() {
+            let header_name = format!("x-amz-meta-{}", k);
+            if let Ok(name) = header_name.parse::<axum::http::HeaderName>() {
+                headers.insert(name, header_val);
+            }
+        }
+    }
+}
+
+fn is_null_version(version_id: Option<&str>) -> bool {
+    version_id.is_none_or(|value| value == "null")
+}
+
+fn bad_digest_response(message: impl Into<String>) -> Response {
+    s3_error_response(S3Error::new(S3ErrorCode::BadDigest, message))
+}
+
+fn base64_header_bytes(headers: &HeaderMap, name: &str) -> Result<Option<Vec<u8>>, Response> {
+    let Some(value) = headers.get(name).and_then(|v| v.to_str().ok()) else {
+        return Ok(None);
+    };
+    STANDARD
+        .decode(value.trim())
+        .map(Some)
+        .map_err(|_| bad_digest_response(format!("Invalid base64 value for {}", name)))
+}
+
+fn has_upload_checksum(headers: &HeaderMap) -> bool {
+    headers.contains_key("content-md5")
+        || headers.contains_key("x-amz-checksum-sha256")
+        || headers.contains_key("x-amz-checksum-crc32")
+}
+
+fn validate_upload_checksums(headers: &HeaderMap, data: &[u8]) -> Result<(), Response> {
+    if let Some(expected) = base64_header_bytes(headers, "content-md5")? {
+        if expected.len() != 16 || Md5::digest(data).as_slice() != expected.as_slice() {
+            return Err(bad_digest_response(
+                "The Content-MD5 you specified did not match what we received",
+            ));
+        }
+    }
+
+    if let Some(expected) = base64_header_bytes(headers, "x-amz-checksum-sha256")? {
+        if Sha256::digest(data).as_slice() != expected.as_slice() {
+            return Err(bad_digest_response(
+                "The x-amz-checksum-sha256 you specified did not match what we received",
+            ));
+        }
+    }
+
+    if let Some(expected) = base64_header_bytes(headers, "x-amz-checksum-crc32")? {
+        let actual = crc32fast::hash(data).to_be_bytes();
+        if expected.as_slice() != actual {
+            return Err(bad_digest_response(
+                "The x-amz-checksum-crc32 you specified did not match what we received",
+            ));
+        }
+    }
+
+    Ok(())
+}
+
+async fn collect_upload_body(body: Body, aws_chunked: bool) -> Result<Vec<u8>, Response> {
+    if aws_chunked {
+        let mut reader = chunked::decode_body(body);
+        let mut data = Vec::new();
+        reader.read_to_end(&mut data).await.map_err(|_| {
+            s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidRequest,
+                "Failed to read aws-chunked request body",
+            ))
+        })?;
+        return Ok(data);
+    }
+
+    http_body_util::BodyExt::collect(body)
+        .await
+        .map(|collected| collected.to_bytes().to_vec())
+        .map_err(|_| {
+            s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidRequest,
+                "Failed to read request body",
+            ))
+        })
+}
+
+fn parse_tagging_header(value: &str) -> Result<Vec<myfsio_common::types::Tag>, Response> {
+    let mut tags = Vec::new();
+    if value.trim().is_empty() {
+        return Ok(tags);
+    }
+
+    for pair in value.split('&') {
+        let (raw_key, raw_value) = pair.split_once('=').ok_or_else(|| {
+            s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidTag,
+                "The x-amz-tagging header must use query-string key=value pairs",
+            ))
+        })?;
+        let key = percent_decode_str(raw_key)
+            .decode_utf8()
+            .map_err(|_| {
+                s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidTag,
+                    "Tag keys must be valid UTF-8",
+                ))
+            })?
+            .to_string();
+        let value = percent_decode_str(raw_value)
+            .decode_utf8()
+            .map_err(|_| {
+                s3_error_response(S3Error::new(
+                    S3ErrorCode::InvalidTag,
+                    "Tag values must be valid UTF-8",
+                ))
+            })?
+            .to_string();
+        tags.push(myfsio_common::types::Tag { key, value });
+    }
+
+    Ok(tags)
+}
+
+fn parse_copy_source(copy_source: &str) -> Result<(String, String, Option<String>), Response> {
+    let source = copy_source.strip_prefix('/').unwrap_or(copy_source);
+    let (bucket_raw, key_and_query) = source.split_once('/').ok_or_else(|| {
+        s3_error_response(S3Error::new(
+            S3ErrorCode::InvalidArgument,
+            "Invalid x-amz-copy-source",
+        ))
+    })?;
+    let (key_raw, query) = key_and_query
+        .split_once('?')
+        .map(|(key, query)| (key, Some(query)))
+        .unwrap_or((key_and_query, None));
+
+    let bucket = percent_decode_str(bucket_raw)
+        .decode_utf8()
+        .map_err(|_| {
+            s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidArgument,
+                "Invalid x-amz-copy-source bucket encoding",
+            ))
+        })?
+        .to_string();
+    let key = percent_decode_str(key_raw)
+        .decode_utf8()
+        .map_err(|_| {
+            s3_error_response(S3Error::new(
+                S3ErrorCode::InvalidArgument,
+                "Invalid x-amz-copy-source key encoding",
+            ))
+        })?
+        .to_string();
+
+    let mut version_id = None;
+    if let Some(query) = query {
+        for pair in query.split('&') {
+            let Some((name, value)) = pair.split_once('=') else {
+                continue;
+            };
+            if name == "versionId" {
+                version_id = Some(
+                    percent_decode_str(value)
+                        .decode_utf8()
+                        .map_err(|_| {
+                            s3_error_response(S3Error::new(
+                                S3ErrorCode::InvalidArgument,
+                                "Invalid x-amz-copy-source versionId encoding",
+                            ))
+                        })?
+                        .to_string(),
+                );
+                break;
+            }
+        }
+    }
+
+    Ok((bucket, key, version_id))
+}
+
 pub async fn put_object(
     State(state): State<AppState>,
     Path((bucket, key)): Path<(String, String)>,
@@ -647,6 +1089,9 @@ pub async fn put_object(
     {
         return response;
     }
+    if let Some(response) = evaluate_put_preconditions(&state, &bucket, &key, &headers).await {
+        return response;
+    }
 
     let content_type = guessed_content_type(
         &key,
@@ -655,6 +1100,9 @@ pub async fn put_object(
 
     let mut metadata = HashMap::new();
     metadata.insert("__content_type__".to_string(), content_type);
+    if let Err(response) = insert_standard_object_metadata(&headers, &mut metadata) {
+        return response;
+    }
 
     for (name, value) in headers.iter() {
         let name_str = name.as_str();
@@ -665,7 +1113,27 @@ pub async fn put_object(
         }
     }
 
-    let boxed: myfsio_storage::traits::AsyncReadStream = if is_aws_chunked(&headers) {
+    let tags = match headers
+        .get("x-amz-tagging")
+        .and_then(|value| value.to_str().ok())
+        .map(parse_tagging_header)
+        .transpose()
+    {
+        Ok(tags) => tags,
+        Err(response) => return response,
+    };
+
+    let aws_chunked = is_aws_chunked(&headers);
+    let boxed: myfsio_storage::traits::AsyncReadStream = if has_upload_checksum(&headers) {
+        let data = match collect_upload_body(body, aws_chunked).await {
+            Ok(data) => data,
+            Err(response) => return response,
+        };
+        if let Err(response) = validate_upload_checksums(&headers, &data) {
+            return response;
+        }
+        Box::pin(std::io::Cursor::new(data))
+    } else if aws_chunked {
         Box::pin(chunked::decode_body(body))
     } else {
         let stream = tokio_util::io::StreamReader::new(
@@ -682,6 +1150,11 @@ pub async fn put_object(
         .await
     {
         Ok(meta) => {
+            if let Some(ref tags) = tags {
+                if let Err(e) = state.storage.set_object_tags(&bucket, &key, tags).await {
+                    return storage_err_response(e);
+                }
+            }
             if let Some(enc_ctx) = resolve_encryption_context(&state, &bucket, &headers).await {
                 if let Some(ref enc_svc) = state.encryption {
                     let obj_path = match state.storage.get_object_path(&bucket, &key).await {
@@ -801,9 +1274,23 @@ pub async fn get_object(
         return list_parts_handler(&state, &bucket, &key, upload_id).await;
     }
 
-    let head_meta = match state.storage.head_object(&bucket, &key).await {
-        Ok(m) => m,
-        Err(e) => return storage_err_response(e),
+    let version_id = query
+        .version_id
+        .as_deref()
+        .filter(|value| !is_null_version(Some(*value)));
+    let head_meta = match version_id {
+        Some(version_id) => match state
+            .storage
+            .head_object_version(&bucket, &key, version_id)
+            .await
+        {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
+        None => match state.storage.head_object(&bucket, &key).await {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
     };
     if let Some(resp) = evaluate_get_preconditions(&headers, &head_meta) {
         return resp;
@@ -818,17 +1305,34 @@ pub async fn get_object(
         return range_get_handler(&state, &bucket, &key, range_str, &query).await;
     }
 
-    let all_meta = state
-        .storage
-        .get_object_metadata(&bucket, &key)
-        .await
-        .unwrap_or_default();
+    let all_meta = match version_id {
+        Some(version_id) => state
+            .storage
+            .get_object_version_metadata(&bucket, &key, version_id)
+            .await
+            .unwrap_or_default(),
+        None => state
+            .storage
+            .get_object_metadata(&bucket, &key)
+            .await
+            .unwrap_or_default(),
+    };
     let enc_meta = myfsio_crypto::encryption::EncryptionMetadata::from_metadata(&all_meta);
 
     if let (Some(ref enc_info), Some(ref enc_svc)) = (&enc_meta, &state.encryption) {
-        let obj_path = match state.storage.get_object_path(&bucket, &key).await {
-            Ok(p) => p,
-            Err(e) => return storage_err_response(e),
+        let obj_path = match version_id {
+            Some(version_id) => match state
+                .storage
+                .get_object_version_path(&bucket, &key, version_id)
+                .await
+            {
+                Ok(p) => p,
+                Err(e) => return storage_err_response(e),
+            },
+            None => match state.storage.get_object_path(&bucket, &key).await {
+                Ok(p) => p,
+                Err(e) => return storage_err_response(e),
+            },
         };
         let tmp_dir = state.config.storage_root.join(".myfsio.sys").join("tmp");
         let _ = tokio::fs::create_dir_all(&tmp_dir).await;
@@ -886,22 +1390,31 @@ pub async fn get_object(
             "x-amz-server-side-encryption",
             enc_info.algorithm.parse().unwrap(),
         );
-
-        for (k, v) in &meta.metadata {
-            if let Ok(header_val) = v.parse() {
-                let header_name = format!("x-amz-meta-{}", k);
-                if let Ok(name) = header_name.parse::<axum::http::HeaderName>() {
-                    resp_headers.insert(name, header_val);
-                }
+        apply_stored_response_headers(&mut resp_headers, &all_meta);
+        if let Some(ref requested_version) = query.version_id {
+            if let Ok(value) = requested_version.parse() {
+                resp_headers.insert("x-amz-version-id", value);
             }
         }
 
+        apply_user_metadata(&mut resp_headers, &meta.metadata);
+
         apply_response_overrides(&mut resp_headers, &query);
 
         return (StatusCode::OK, resp_headers, body).into_response();
     }
 
-    match state.storage.get_object(&bucket, &key).await {
+    let object_result = match version_id {
+        Some(version_id) => {
+            state
+                .storage
+                .get_object_version(&bucket, &key, version_id)
+                .await
+        }
+        None => state.storage.get_object(&bucket, &key).await,
+    };
+
+    match object_result {
         Ok((meta, reader)) => {
             let stream = ReaderStream::new(reader);
             let body = Body::from_stream(stream);
@@ -921,16 +1434,15 @@ pub async fn get_object(
                     .unwrap(),
             );
             headers.insert("accept-ranges", "bytes".parse().unwrap());
-
-            for (k, v) in &meta.metadata {
-                if let Ok(header_val) = v.parse() {
-                    let header_name = format!("x-amz-meta-{}", k);
-                    if let Ok(name) = header_name.parse::<axum::http::HeaderName>() {
-                        headers.insert(name, header_val);
-                    }
+            apply_stored_response_headers(&mut headers, &all_meta);
+            if let Some(ref requested_version) = query.version_id {
+                if let Ok(value) = requested_version.parse() {
+                    headers.insert("x-amz-version-id", value);
                 }
             }
 
+            apply_user_metadata(&mut headers, &meta.metadata);
+
             apply_response_overrides(&mut headers, &query);
 
             (StatusCode::OK, headers, body).into_response()
@@ -978,6 +1490,35 @@ pub async fn delete_object(
         return abort_multipart_handler(&state, &bucket, upload_id).await;
     }
 
+    if let Some(version_id) = query
+        .version_id
+        .as_deref()
+        .filter(|value| !is_null_version(Some(*value)))
+    {
+        if let Err(response) =
+            ensure_object_version_lock_allows_delete(&state, &bucket, &key, version_id, &headers)
+                .await
+        {
+            return response;
+        }
+        return match state
+            .storage
+            .delete_object_version(&bucket, &key, version_id)
+            .await
+        {
+            Ok(()) => {
+                let mut resp_headers = HeaderMap::new();
+                if let Ok(value) = version_id.parse() {
+                    resp_headers.insert("x-amz-version-id", value);
+                }
+                notifications::emit_object_removed(&state, &bucket, &key, "", "", "", "Delete");
+                trigger_replication(&state, &bucket, &key, "delete");
+                (StatusCode::NO_CONTENT, resp_headers).into_response()
+            }
+            Err(e) => storage_err_response(e),
+        };
+    }
+
     if let Err(response) =
         ensure_object_lock_allows_write(&state, &bucket, &key, Some(&headers)).await
     {
@@ -997,13 +1538,40 @@ pub async fn delete_object(
 pub async fn head_object(
     State(state): State<AppState>,
     Path((bucket, key)): Path<(String, String)>,
+    Query(query): Query<ObjectQuery>,
     headers: HeaderMap,
 ) -> Response {
-    match state.storage.head_object(&bucket, &key).await {
+    let version_id = query
+        .version_id
+        .as_deref()
+        .filter(|value| !is_null_version(Some(*value)));
+    let result = match version_id {
+        Some(version_id) => {
+            state
+                .storage
+                .head_object_version(&bucket, &key, version_id)
+                .await
+        }
+        None => state.storage.head_object(&bucket, &key).await,
+    };
+
+    match result {
         Ok(meta) => {
             if let Some(resp) = evaluate_get_preconditions(&headers, &meta) {
                 return resp;
             }
+            let all_meta = match version_id {
+                Some(version_id) => state
+                    .storage
+                    .get_object_version_metadata(&bucket, &key, version_id)
+                    .await
+                    .unwrap_or_default(),
+                None => state
+                    .storage
+                    .get_object_metadata(&bucket, &key)
+                    .await
+                    .unwrap_or_default(),
+            };
             let mut headers = HeaderMap::new();
             headers.insert("content-length", meta.size.to_string().parse().unwrap());
             if let Some(ref etag) = meta.etag {
@@ -1019,16 +1587,15 @@ pub async fn head_object(
                     .unwrap(),
             );
             headers.insert("accept-ranges", "bytes".parse().unwrap());
-
-            for (k, v) in &meta.metadata {
-                if let Ok(header_val) = v.parse() {
-                    let header_name = format!("x-amz-meta-{}", k);
-                    if let Ok(name) = header_name.parse::<axum::http::HeaderName>() {
-                        headers.insert(name, header_val);
-                    }
+            apply_stored_response_headers(&mut headers, &all_meta);
+            if let Some(ref requested_version) = query.version_id {
+                if let Ok(value) = requested_version.parse() {
+                    headers.insert("x-amz-version-id", value);
                 }
             }
 
+            apply_user_metadata(&mut headers, &meta.metadata);
+
             (StatusCode::OK, headers).into_response()
         }
         Err(e) => storage_err_response(e),
@@ -1311,30 +1878,70 @@ async fn copy_object_handler(
         return response;
     }
 
-    let source = copy_source.strip_prefix('/').unwrap_or(copy_source);
-    let (src_bucket, src_key) = match source.split_once('/') {
-        Some(parts) => parts,
-        None => {
-            return s3_error_response(S3Error::new(
-                myfsio_common::error::S3ErrorCode::InvalidArgument,
-                "Invalid x-amz-copy-source",
-            ));
-        }
+    let (src_bucket, src_key, src_version_id) = match parse_copy_source(copy_source) {
+        Ok(parts) => parts,
+        Err(response) => return response,
     };
 
-    let source_meta = match state.storage.head_object(src_bucket, src_key).await {
-        Ok(m) => m,
-        Err(e) => return storage_err_response(e),
+    let source_meta = match src_version_id.as_deref() {
+        Some(version_id) if version_id != "null" => match state
+            .storage
+            .head_object_version(&src_bucket, &src_key, version_id)
+            .await
+        {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
+        _ => match state.storage.head_object(&src_bucket, &src_key).await {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
     };
     if let Some(resp) = evaluate_copy_preconditions(headers, &source_meta) {
         return resp;
     }
 
-    match state
-        .storage
-        .copy_object(src_bucket, src_key, dst_bucket, dst_key)
-        .await
+    let copy_result = if let Some(version_id) = src_version_id
+        .as_deref()
+        .filter(|value| !is_null_version(Some(*value)))
     {
+        let (_meta, mut reader) = match state
+            .storage
+            .get_object_version(&src_bucket, &src_key, version_id)
+            .await
+        {
+            Ok(result) => result,
+            Err(e) => return storage_err_response(e),
+        };
+        let mut data = Vec::new();
+        if let Err(e) = reader.read_to_end(&mut data).await {
+            return storage_err_response(myfsio_storage::error::StorageError::Io(e));
+        }
+        let metadata = match state
+            .storage
+            .get_object_version_metadata(&src_bucket, &src_key, version_id)
+            .await
+        {
+            Ok(metadata) => metadata,
+            Err(e) => return storage_err_response(e),
+        };
+        state
+            .storage
+            .put_object(
+                dst_bucket,
+                dst_key,
+                Box::pin(std::io::Cursor::new(data)),
+                Some(metadata),
+            )
+            .await
+    } else {
+        state
+            .storage
+            .copy_object(&src_bucket, &src_key, dst_bucket, dst_key)
+            .await
+    };
+
+    match copy_result {
         Ok(meta) => {
             let etag = meta.etag.as_deref().unwrap_or("");
             let last_modified = myfsio_xml::response::format_s3_datetime(&meta.last_modified);
@@ -1372,13 +1979,23 @@ async fn delete_objects_handler(state: &AppState, bucket: &str, body: Body) -> R
     let mut errors = Vec::new();
 
     for obj in &parsed.objects {
-        if let Err(message) = match state.storage.head_object(bucket, &obj.key).await {
-            Ok(_) => match state.storage.get_object_metadata(bucket, &obj.key).await {
+        if let Err(message) = match obj.version_id.as_deref() {
+            Some(version_id) if version_id != "null" => match state
+                .storage
+                .get_object_version_metadata(bucket, &obj.key, version_id)
+                .await
+            {
                 Ok(metadata) => object_lock::can_delete_object(&metadata, false),
                 Err(err) => Err(S3Error::from(err).message),
             },
-            Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => Ok(()),
-            Err(err) => Err(S3Error::from(err).message),
+            _ => match state.storage.head_object(bucket, &obj.key).await {
+                Ok(_) => match state.storage.get_object_metadata(bucket, &obj.key).await {
+                    Ok(metadata) => object_lock::can_delete_object(&metadata, false),
+                    Err(err) => Err(S3Error::from(err).message),
+                },
+                Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => Ok(()),
+                Err(err) => Err(S3Error::from(err).message),
+            },
         } {
             errors.push((
                 obj.key.clone(),
@@ -1387,7 +2004,20 @@ async fn delete_objects_handler(state: &AppState, bucket: &str, body: Body) -> R
             ));
             continue;
         }
-        match state.storage.delete_object(bucket, &obj.key).await {
+        let delete_result = if let Some(version_id) = obj.version_id.as_deref() {
+            if version_id == "null" {
+                state.storage.delete_object(bucket, &obj.key).await
+            } else {
+                state
+                    .storage
+                    .delete_object_version(bucket, &obj.key, version_id)
+                    .await
+            }
+        } else {
+            state.storage.delete_object(bucket, &obj.key).await
+        };
+
+        match delete_result {
             Ok(()) => {
                 notifications::emit_object_removed(state, bucket, &obj.key, "", "", "", "Delete");
                 trigger_replication(state, bucket, &obj.key, "delete");
@@ -1415,9 +2045,23 @@ async fn range_get_handler(
     range_str: &str,
     query: &ObjectQuery,
 ) -> Response {
-    let meta = match state.storage.head_object(bucket, key).await {
-        Ok(m) => m,
-        Err(e) => return storage_err_response(e),
+    let version_id = query
+        .version_id
+        .as_deref()
+        .filter(|value| !is_null_version(Some(*value)));
+    let meta = match version_id {
+        Some(version_id) => match state
+            .storage
+            .head_object_version(bucket, key, version_id)
+            .await
+        {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
+        None => match state.storage.head_object(bucket, key).await {
+            Ok(m) => m,
+            Err(e) => return storage_err_response(e),
+        },
     };
 
     let total_size = meta.size;
@@ -1431,9 +2075,19 @@ async fn range_get_handler(
         }
     };
 
-    let path = match state.storage.get_object_path(bucket, key).await {
-        Ok(p) => p,
-        Err(e) => return storage_err_response(e),
+    let path = match version_id {
+        Some(version_id) => match state
+            .storage
+            .get_object_version_path(bucket, key, version_id)
+            .await
+        {
+            Ok(p) => p,
+            Err(e) => return storage_err_response(e),
+        },
+        None => match state.storage.get_object_path(bucket, key).await {
+            Ok(p) => p,
+            Err(e) => return storage_err_response(e),
+        },
     };
 
     let mut file = match tokio::fs::File::open(&path).await {
@@ -1463,6 +2117,11 @@ async fn range_get_handler(
     }
     insert_content_type(&mut headers, key, meta.content_type.as_deref());
     headers.insert("accept-ranges", "bytes".parse().unwrap());
+    if let Some(ref requested_version) = query.version_id {
+        if let Ok(value) = requested_version.parse() {
+            headers.insert("x-amz-version-id", value);
+        }
+    }
 
     apply_response_overrides(&mut headers, query);
 
@@ -1514,6 +2173,49 @@ fn evaluate_get_preconditions(
     None
 }
 
+async fn evaluate_put_preconditions(
+    state: &AppState,
+    bucket: &str,
+    key: &str,
+    headers: &HeaderMap,
+) -> Option<Response> {
+    let has_if_match = headers.contains_key("if-match");
+    let has_if_none_match = headers.contains_key("if-none-match");
+    if !has_if_match && !has_if_none_match {
+        return None;
+    }
+
+    match state.storage.head_object(bucket, key).await {
+        Ok(meta) => {
+            if let Some(value) = headers.get("if-match").and_then(|v| v.to_str().ok()) {
+                if !etag_condition_matches(value, meta.etag.as_deref()) {
+                    return Some(s3_error_response(S3Error::from_code(
+                        S3ErrorCode::PreconditionFailed,
+                    )));
+                }
+            }
+            if let Some(value) = headers.get("if-none-match").and_then(|v| v.to_str().ok()) {
+                if etag_condition_matches(value, meta.etag.as_deref()) {
+                    return Some(s3_error_response(S3Error::from_code(
+                        S3ErrorCode::PreconditionFailed,
+                    )));
+                }
+            }
+            None
+        }
+        Err(myfsio_storage::error::StorageError::ObjectNotFound { .. }) => {
+            if has_if_match {
+                Some(s3_error_response(S3Error::from_code(
+                    S3ErrorCode::PreconditionFailed,
+                )))
+            } else {
+                None
+            }
+        }
+        Err(err) => Some(storage_err_response(err)),
+    }
+}
+
 fn evaluate_copy_preconditions(
     headers: &HeaderMap,
     source_meta: &myfsio_common::types::ObjectMeta,
@@ -2176,6 +2878,20 @@ mod tests {
             .unwrap()
     }
 
+    #[test]
+    fn aws_chunked_wire_encoding_is_not_persisted_as_object_encoding() {
+        let mut headers = HeaderMap::new();
+        headers.insert("content-encoding", "aws-chunked".parse().unwrap());
+        let mut metadata = HashMap::new();
+        insert_standard_object_metadata(&headers, &mut metadata).unwrap();
+        assert!(!metadata.contains_key("__content_encoding__"));
+
+        headers.insert("content-encoding", "aws-chunked, gzip".parse().unwrap());
+        let mut metadata = HashMap::new();
+        insert_standard_object_metadata(&headers, &mut metadata).unwrap();
+        assert_eq!(metadata.get("__content_encoding__").unwrap(), "gzip");
+    }
+
     #[tokio::test]
     async fn public_bucket_acl_allows_anonymous_reads() {
         let (state, _tmp) = test_state();
diff --git a/rust/myfsio-engine/crates/myfsio-server/src/handlers/ui_api.rs b/rust/myfsio-engine/crates/myfsio-server/src/handlers/ui_api.rs
index 82da4df..0217117 100644
--- a/rust/myfsio-engine/crates/myfsio-server/src/handlers/ui_api.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/src/handlers/ui_api.rs
@@ -117,6 +117,7 @@ fn storage_status(err: &StorageError) -> StatusCode {
     match err {
         StorageError::BucketNotFound(_)
         | StorageError::ObjectNotFound { .. }
+        | StorageError::VersionNotFound { .. }
         | StorageError::UploadNotFound(_) => StatusCode::NOT_FOUND,
         StorageError::InvalidBucketName(_)
         | StorageError::InvalidObjectKey(_)
diff --git a/rust/myfsio-engine/crates/myfsio-server/src/lib.rs b/rust/myfsio-engine/crates/myfsio-server/src/lib.rs
index 30ed1ec..8199121 100644
--- a/rust/myfsio-engine/crates/myfsio-server/src/lib.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/src/lib.rs
@@ -9,7 +9,7 @@ pub mod templates;
 
 use axum::Router;
 
-pub const SERVER_HEADER: &str = "MyFSIO";
+pub const SERVER_HEADER: &str = concat!("MyFSIO-Rust/", env!("CARGO_PKG_VERSION"));
 
 pub fn create_ui_router(state: state::AppState) -> Router {
     use axum::routing::{delete, get, post, put};
diff --git a/rust/myfsio-engine/crates/myfsio-server/src/middleware/auth.rs b/rust/myfsio-engine/crates/myfsio-server/src/middleware/auth.rs
index 9a8f189..dce2ea4 100644
--- a/rust/myfsio-engine/crates/myfsio-server/src/middleware/auth.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/src/middleware/auth.rs
@@ -1,5 +1,5 @@
 use axum::extract::{Request, State};
-use axum::http::{header, HeaderMap, Method, StatusCode};
+use axum::http::{header, HeaderMap, Method, StatusCode, Uri};
 use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 
@@ -15,6 +15,9 @@ use tokio::io::AsyncReadExt;
 use crate::services::acl::acl_from_bucket_config;
 use crate::state::AppState;
 
+#[derive(Clone, Debug)]
+struct OriginalCanonicalPath(String);
+
 fn website_error_response(
     status: StatusCode,
     body: Option<Vec<u8>>,
@@ -45,7 +48,7 @@ fn website_error_response(
 fn default_website_error_body(status: StatusCode) -> String {
     let code = status.as_u16();
     if status == StatusCode::NOT_FOUND {
-        "404 page not found".to_string()
+        "<h1>404 page not found</h1>".to_string()
     } else {
         let reason = status.canonical_reason().unwrap_or("Error");
         format!("{code} {reason}")
@@ -324,6 +327,67 @@ async fn maybe_serve_website(
     .await
 }
 
+fn virtual_host_candidate(host: &str) -> Option<String> {
+    let (candidate, _) = host.split_once('.')?;
+    if candidate.is_empty() || matches!(candidate, "www" | "s3" | "api" | "admin" | "kms") {
+        return None;
+    }
+    if myfsio_storage::validation::validate_bucket_name(candidate).is_some() {
+        return None;
+    }
+    Some(candidate.to_string())
+}
+
+async fn virtual_host_bucket(
+    state: &AppState,
+    host: &str,
+    path: &str,
+    method: &Method,
+) -> Option<String> {
+    if path.starts_with("/ui")
+        || path.starts_with("/admin")
+        || path.starts_with("/kms")
+        || path.starts_with("/myfsio")
+    {
+        return None;
+    }
+
+    let bucket = virtual_host_candidate(host)?;
+    if path == format!("/{}", bucket) || path.starts_with(&format!("/{}/", bucket)) {
+        return None;
+    }
+
+    match state.storage.bucket_exists(&bucket).await {
+        Ok(true) => Some(bucket),
+        Ok(false) if *method == Method::PUT && path == "/" => Some(bucket),
+        _ => None,
+    }
+}
+
+fn rewrite_uri_for_virtual_host(uri: &Uri, bucket: &str) -> Option<Uri> {
+    let path = uri.path();
+    let rewritten_path = if path == "/" {
+        format!("/{}/", bucket)
+    } else {
+        format!("/{}{}", bucket, path)
+    };
+    let path_and_query = match uri.query() {
+        Some(query) => format!("{}?{}", rewritten_path, query),
+        None => rewritten_path,
+    };
+
+    let mut parts = uri.clone().into_parts();
+    parts.path_and_query = Some(path_and_query.parse().ok()?);
+    Uri::from_parts(parts).ok()
+}
+
+fn sigv4_canonical_path(req: &Request) -> &str {
+    req.extensions()
+        .get::<OriginalCanonicalPath>()
+        .map(|path| path.0.as_str())
+        .unwrap_or_else(|| req.uri().path())
+}
+
 pub async fn auth_layer(State(state): State<AppState>, mut req: Request, next: Next) -> Response {
     let start = Instant::now();
     let uri = req.uri().clone();
@@ -360,7 +424,7 @@ pub async fn auth_layer(State(state): State<AppState>, mut req: Request, next: N
     } else if let Some(response) = maybe_serve_website(
         &state,
         method.clone(),
-        host.unwrap_or_default(),
+        host.clone().unwrap_or_default(),
         path.clone(),
         range_header,
     )
@@ -368,38 +432,53 @@ pub async fn auth_layer(State(state): State<AppState>, mut req: Request, next: N
     {
         response
     } else {
+        let auth_path = if let Some(bucket) =
+            virtual_host_bucket(&state, host.as_deref().unwrap_or_default(), &path, &method).await
+        {
+            if let Some(rewritten) = rewrite_uri_for_virtual_host(req.uri(), &bucket) {
+                req.extensions_mut()
+                    .insert(OriginalCanonicalPath(path.clone()));
+                *req.uri_mut() = rewritten;
+                req.uri().path().to_string()
+            } else {
+                path.clone()
+            }
+        } else {
+            path.clone()
+        };
+
         match try_auth(&state, &req) {
             AuthResult::NoAuth => match authorize_request(
                 &state,
                 None,
                 &method,
-                &path,
+                &auth_path,
                 &query,
                 copy_source.as_deref(),
             )
             .await
             {
                 Ok(()) => next.run(req).await,
-                Err(err) => error_response(err, &path),
+                Err(err) => error_response(err, &auth_path),
             },
             AuthResult::Ok(principal) => {
                 if let Err(err) = authorize_request(
                     &state,
                     Some(&principal),
                     &method,
-                    &path,
+                    &auth_path,
                     &query,
                     copy_source.as_deref(),
                 )
                 .await
                 {
-                    error_response(err, &path)
+                    error_response(err, &auth_path)
                 } else {
                     req.extensions_mut().insert(principal);
                     next.run(req).await
                 }
             }
-            AuthResult::Denied(err) => error_response(err, &path),
+            AuthResult::Denied(err) => error_response(err, &auth_path),
         }
     };
 
@@ -1078,7 +1157,7 @@ fn verify_sigv4_header(state: &AppState, req: &Request, auth_str: &str) -> AuthR
     };
 
     let method = req.method().as_str();
-    let canonical_uri = req.uri().path();
+    let canonical_uri = sigv4_canonical_path(req);
 
     let query_params = parse_query_params(req.uri().query().unwrap_or(""));
 
@@ -1234,7 +1313,7 @@ fn verify_sigv4_query(state: &AppState, req: &Request) -> AuthResult {
     };
 
     let method = req.method().as_str();
-    let canonical_uri = req.uri().path();
+    let canonical_uri = sigv4_canonical_path(req);
 
     let query_params_no_sig: Vec<(String, String)> = params
         .iter()
diff --git a/rust/myfsio-engine/crates/myfsio-server/tests/integration.rs b/rust/myfsio-engine/crates/myfsio-server/tests/integration.rs
index 09767f9..8a50ece 100644
--- a/rust/myfsio-engine/crates/myfsio-server/tests/integration.rs
+++ b/rust/myfsio-engine/crates/myfsio-server/tests/integration.rs
@@ -2121,6 +2121,445 @@ async fn test_bucket_versioning() {
     assert!(body.contains("<Status>Enabled</Status>"));
 }
 
+#[tokio::test]
+async fn test_versioned_object_can_be_read_and_deleted_by_version_id() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/versions-bucket",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/versions-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,
+            "/versions-bucket/doc.txt",
+            Body::from("first"),
+        ))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/versions-bucket/doc.txt",
+            Body::from("second"),
+        ))
+        .await
+        .unwrap();
+
+    let list_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/versions-bucket?versions",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(list_resp.status(), StatusCode::OK);
+    let list_body = String::from_utf8(
+        list_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    let archived_version_id = list_body
+        .split("<VersionId>")
+        .filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
+        .find(|id| *id != "null")
+        .expect("archived version id")
+        .to_string();
+
+    let version_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            &format!("/versions-bucket/doc.txt?versionId={}", archived_version_id),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(version_resp.status(), StatusCode::OK);
+    assert_eq!(
+        version_resp.headers()["x-amz-version-id"].to_str().unwrap(),
+        archived_version_id
+    );
+    let version_body = version_resp.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&version_body[..], b"first");
+
+    let traversal_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            &format!(
+                "/versions-bucket/doc.txt?versionId=../other/{}",
+                archived_version_id
+            ),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(traversal_resp.status(), StatusCode::NOT_FOUND);
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/versions-bucket/doc.txt",
+            Body::from("third"),
+        ))
+        .await
+        .unwrap();
+    let limited_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/versions-bucket?versions&max-keys=1",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(limited_resp.status(), StatusCode::OK);
+    let limited_body = String::from_utf8(
+        limited_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert_eq!(limited_body.matches("<Version>").count(), 1);
+    assert!(limited_body.contains("<IsTruncated>true</IsTruncated>"));
+
+    let delete_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            &format!("/versions-bucket/doc.txt?versionId={}", archived_version_id),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(delete_resp.status(), StatusCode::NO_CONTENT);
+
+    let missing_resp = app
+        .oneshot(signed_request(
+            Method::GET,
+            &format!("/versions-bucket/doc.txt?versionId={}", archived_version_id),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(missing_resp.status(), StatusCode::NOT_FOUND);
+}
+
+#[tokio::test]
+async fn test_retention_is_enforced_when_deleting_archived_version() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/locked-versions",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/locked-versions?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(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/locked-versions/doc.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("x-amz-object-lock-mode", "GOVERNANCE")
+                .header(
+                    "x-amz-object-lock-retain-until-date",
+                    "2099-01-01T00:00:00Z",
+                )
+                .body(Body::from("locked"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    app.clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/locked-versions/doc.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("x-amz-bypass-governance-retention", "true")
+                .body(Body::from("replacement"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    let list_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::GET,
+            "/locked-versions?versions",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(list_resp.status(), StatusCode::OK);
+    let list_body = String::from_utf8(
+        list_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    let archived_version_id = list_body
+        .split("<VersionId>")
+        .filter_map(|part| part.split_once("</VersionId>").map(|(id, _)| id))
+        .find(|id| *id != "null")
+        .expect("archived version id")
+        .to_string();
+
+    let denied = app
+        .clone()
+        .oneshot(signed_request(
+            Method::DELETE,
+            &format!("/locked-versions/doc.txt?versionId={}", archived_version_id),
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(denied.status(), StatusCode::FORBIDDEN);
+
+    let allowed = app
+        .oneshot(
+            Request::builder()
+                .method(Method::DELETE)
+                .uri(format!(
+                    "/locked-versions/doc.txt?versionId={}",
+                    archived_version_id
+                ))
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("x-amz-bypass-governance-retention", "true")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(allowed.status(), StatusCode::NO_CONTENT);
+}
+
+#[tokio::test]
+async fn test_put_object_validates_content_md5() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/md5-bucket", Body::empty()))
+        .await
+        .unwrap();
+
+    let bad_resp = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/md5-bucket/object.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("content-md5", "AAAAAAAAAAAAAAAAAAAAAA==")
+                .body(Body::from("hello"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(bad_resp.status(), StatusCode::BAD_REQUEST);
+    let bad_body = String::from_utf8(
+        bad_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(bad_body.contains("<Code>BadDigest</Code>"));
+
+    let good_resp = app
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/md5-bucket/object.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("content-md5", "XUFAKrxLKna5cZ2REBfFkg==")
+                .body(Body::from("hello"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(good_resp.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn test_put_object_tagging_and_standard_headers_are_persisted() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(
+            Method::PUT,
+            "/headers-bucket",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+
+    let put_resp = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/headers-bucket/report.txt")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .header("x-amz-tagging", "env=prod&name=quarter%201")
+                .header("cache-control", "max-age=60")
+                .header("content-disposition", "attachment")
+                .header("content-language", "en-US")
+                .header("x-amz-storage-class", "STANDARD_IA")
+                .body(Body::from("report"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(put_resp.status(), StatusCode::OK);
+
+    let head_resp = app
+        .clone()
+        .oneshot(signed_request(
+            Method::HEAD,
+            "/headers-bucket/report.txt",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(head_resp.status(), StatusCode::OK);
+    assert_eq!(head_resp.headers()["cache-control"], "max-age=60");
+    assert_eq!(head_resp.headers()["content-disposition"], "attachment");
+    assert_eq!(head_resp.headers()["content-language"], "en-US");
+    assert_eq!(head_resp.headers()["x-amz-storage-class"], "STANDARD_IA");
+
+    let tags_resp = app
+        .oneshot(signed_request(
+            Method::GET,
+            "/headers-bucket/report.txt?tagging",
+            Body::empty(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(tags_resp.status(), StatusCode::OK);
+    let tags_body = String::from_utf8(
+        tags_resp
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    assert!(tags_body.contains("<Key>env</Key>"));
+    assert!(tags_body.contains("<Value>prod</Value>"));
+    assert!(tags_body.contains("<Key>name</Key>"));
+    assert!(tags_body.contains("<Value>quarter 1</Value>"));
+}
+
+#[tokio::test]
+async fn test_virtual_host_bucket_routes_to_s3_object_handlers() {
+    let (app, _tmp) = test_app();
+
+    app.clone()
+        .oneshot(signed_request(Method::PUT, "/vh-bucket", Body::empty()))
+        .await
+        .unwrap();
+
+    let put_resp = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .method(Method::PUT)
+                .uri("/hello.txt")
+                .header("host", "vh-bucket.localhost")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::from("virtual host body"))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(put_resp.status(), StatusCode::OK);
+
+    let get_resp = app
+        .oneshot(
+            Request::builder()
+                .method(Method::GET)
+                .uri("/hello.txt")
+                .header("host", "vh-bucket.localhost")
+                .header("x-access-key", TEST_ACCESS_KEY)
+                .header("x-secret-key", TEST_SECRET_KEY)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(get_resp.status(), StatusCode::OK);
+    let body = get_resp.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body[..], b"virtual host body");
+}
+
 #[tokio::test]
 async fn test_bucket_tagging() {
     let (app, _tmp) = test_app();
@@ -3323,7 +3762,7 @@ async fn test_static_website_default_404_returns_html_body() {
     )
     .unwrap();
     assert_eq!(body.len(), content_length);
-    assert_eq!(body, "404 page not found");
+    assert_eq!(body, "<h1>404 page not found</h1>");
 
     let head_resp = app
         .oneshot(website_request(Method::HEAD, "/missing.html"))
diff --git a/rust/myfsio-engine/crates/myfsio-storage/src/error.rs b/rust/myfsio-engine/crates/myfsio-storage/src/error.rs
index 992d1fc..4bb652a 100644
--- a/rust/myfsio-engine/crates/myfsio-storage/src/error.rs
+++ b/rust/myfsio-engine/crates/myfsio-storage/src/error.rs
@@ -11,6 +11,12 @@ pub enum StorageError {
     BucketNotEmpty(String),
     #[error("Object not found: {bucket}/{key}")]
     ObjectNotFound { bucket: String, key: String },
+    #[error("Object version not found: {bucket}/{key}?versionId={version_id}")]
+    VersionNotFound {
+        bucket: String,
+        key: String,
+        version_id: String,
+    },
     #[error("Invalid bucket name: {0}")]
     InvalidBucketName(String),
     #[error("Invalid object key: {0}")]
@@ -46,6 +52,12 @@ impl From<StorageError> for S3Error {
                 S3Error::from_code(S3ErrorCode::NoSuchKey)
                     .with_resource(format!("/{}/{}", bucket, key))
             }
+            StorageError::VersionNotFound {
+                bucket,
+                key,
+                version_id,
+            } => S3Error::from_code(S3ErrorCode::NoSuchVersion)
+                .with_resource(format!("/{}/{}?versionId={}", bucket, key, version_id)),
             StorageError::InvalidBucketName(msg) => {
                 S3Error::new(S3ErrorCode::InvalidBucketName, msg)
             }
diff --git a/rust/myfsio-engine/crates/myfsio-storage/src/fs_backend.rs b/rust/myfsio-engine/crates/myfsio-storage/src/fs_backend.rs
index 7f5bd36..9f7a72f 100644
--- a/rust/myfsio-engine/crates/myfsio-storage/src/fs_backend.rs
+++ b/rust/myfsio-engine/crates/myfsio-storage/src/fs_backend.rs
@@ -605,6 +605,144 @@ impl FsStorageBackend {
         Ok(source_size)
     }
 
+    fn version_record_paths(
+        &self,
+        bucket_name: &str,
+        key: &str,
+        version_id: &str,
+    ) -> (PathBuf, PathBuf) {
+        let version_dir = self.version_dir(bucket_name, key);
+        (
+            version_dir.join(format!("{}.json", version_id)),
+            version_dir.join(format!("{}.bin", version_id)),
+        )
+    }
+
+    fn validate_version_id(bucket_name: &str, key: &str, version_id: &str) -> StorageResult<()> {
+        if version_id.is_empty()
+            || version_id.contains('/')
+            || version_id.contains('\\')
+            || version_id.contains("..")
+        {
+            return Err(StorageError::VersionNotFound {
+                bucket: bucket_name.to_string(),
+                key: key.to_string(),
+                version_id: version_id.to_string(),
+            });
+        }
+        Ok(())
+    }
+
+    fn read_version_record_sync(
+        &self,
+        bucket_name: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<(Value, PathBuf)> {
+        self.require_bucket(bucket_name)?;
+        self.validate_key(key)?;
+        Self::validate_version_id(bucket_name, key, version_id)?;
+        let (manifest_path, data_path) = self.version_record_paths(bucket_name, key, version_id);
+        if !manifest_path.is_file() || !data_path.is_file() {
+            return Err(StorageError::VersionNotFound {
+                bucket: bucket_name.to_string(),
+                key: key.to_string(),
+                version_id: version_id.to_string(),
+            });
+        }
+
+        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)?;
+        Ok((record, data_path))
+    }
+
+    fn version_metadata_from_record(record: &Value) -> HashMap<String, String> {
+        record
+            .get("metadata")
+            .and_then(Value::as_object)
+            .map(|meta| {
+                meta.iter()
+                    .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
+                    .collect::<HashMap<String, String>>()
+            })
+            .unwrap_or_default()
+    }
+
+    fn object_meta_from_version_record(
+        &self,
+        key: &str,
+        record: &Value,
+        data_path: &Path,
+    ) -> StorageResult<ObjectMeta> {
+        let metadata = Self::version_metadata_from_record(record);
+
+        let data_len = std::fs::metadata(data_path)
+            .map(|meta| meta.len())
+            .unwrap_or_default();
+        let size = record
+            .get("size")
+            .and_then(Value::as_u64)
+            .unwrap_or(data_len);
+        let last_modified = record
+            .get("archived_at")
+            .and_then(Value::as_str)
+            .and_then(|value| DateTime::parse_from_rfc3339(value).ok())
+            .map(|value| value.with_timezone(&Utc))
+            .unwrap_or_else(Utc::now);
+        let etag = record
+            .get("etag")
+            .and_then(Value::as_str)
+            .map(ToOwned::to_owned)
+            .or_else(|| metadata.get("__etag__").cloned());
+
+        let mut obj = ObjectMeta::new(key.to_string(), size, last_modified);
+        obj.etag = etag;
+        obj.content_type = metadata.get("__content_type__").cloned();
+        obj.storage_class = metadata
+            .get("__storage_class__")
+            .cloned()
+            .or_else(|| Some("STANDARD".to_string()));
+        obj.metadata = metadata
+            .into_iter()
+            .filter(|(k, _)| !k.starts_with("__"))
+            .collect();
+        Ok(obj)
+    }
+
+    fn version_info_from_record(&self, fallback_key: &str, record: &Value) -> VersionInfo {
+        let version_id = record
+            .get("version_id")
+            .and_then(Value::as_str)
+            .unwrap_or("")
+            .to_string();
+        let key = record
+            .get("key")
+            .and_then(Value::as_str)
+            .unwrap_or(fallback_key)
+            .to_string();
+        let size = record.get("size").and_then(Value::as_u64).unwrap_or(0);
+        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);
+        let etag = record
+            .get("etag")
+            .and_then(Value::as_str)
+            .map(|s| s.to_string());
+
+        VersionInfo {
+            version_id,
+            key,
+            size,
+            last_modified: archived_at,
+            etag,
+            is_latest: false,
+            is_delete_marker: false,
+        }
+    }
+
     fn bucket_stats_sync(&self, bucket_name: &str) -> StorageResult<BucketStats> {
         let bucket_path = self.require_bucket(bucket_name)?;
 
@@ -1241,6 +1379,10 @@ impl crate::traits::StorageEngine for FsStorageBackend {
         let mut obj = ObjectMeta::new(key.to_string(), meta.len(), lm);
         obj.etag = stored_meta.get("__etag__").cloned();
         obj.content_type = stored_meta.get("__content_type__").cloned();
+        obj.storage_class = stored_meta
+            .get("__storage_class__")
+            .cloned()
+            .or_else(|| Some("STANDARD".to_string()));
         obj.metadata = stored_meta
             .into_iter()
             .filter(|(k, _)| !k.starts_with("__"))
@@ -1289,6 +1431,10 @@ impl crate::traits::StorageEngine for FsStorageBackend {
         let mut obj = ObjectMeta::new(key.to_string(), meta.len(), lm);
         obj.etag = stored_meta.get("__etag__").cloned();
         obj.content_type = stored_meta.get("__content_type__").cloned();
+        obj.storage_class = stored_meta
+            .get("__storage_class__")
+            .cloned()
+            .or_else(|| Some("STANDARD".to_string()));
         obj.metadata = stored_meta
             .into_iter()
             .filter(|(k, _)| !k.starts_with("__"))
@@ -1296,6 +1442,51 @@ impl crate::traits::StorageEngine for FsStorageBackend {
         Ok(obj)
     }
 
+    async fn get_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<(ObjectMeta, AsyncReadStream)> {
+        let (record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
+        let obj = self.object_meta_from_version_record(key, &record, &data_path)?;
+        let file = tokio::fs::File::open(&data_path)
+            .await
+            .map_err(StorageError::Io)?;
+        let stream: AsyncReadStream = Box::pin(file);
+        Ok((obj, stream))
+    }
+
+    async fn get_object_version_path(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<PathBuf> {
+        let (_record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
+        Ok(data_path)
+    }
+
+    async fn head_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<ObjectMeta> {
+        let (record, data_path) = self.read_version_record_sync(bucket, key, version_id)?;
+        self.object_meta_from_version_record(key, &record, &data_path)
+    }
+
+    async fn get_object_version_metadata(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<HashMap<String, String>> {
+        let (record, _data_path) = self.read_version_record_sync(bucket, key, version_id)?;
+        Ok(Self::version_metadata_from_record(&record))
+    }
+
     async fn delete_object(&self, bucket: &str, key: &str) -> StorageResult<()> {
         let bucket_path = self.require_bucket(bucket)?;
         let path = self.object_path(bucket, key)?;
@@ -1317,6 +1508,32 @@ impl crate::traits::StorageEngine for FsStorageBackend {
         Ok(())
     }
 
+    async fn delete_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<()> {
+        self.require_bucket(bucket)?;
+        self.validate_key(key)?;
+        Self::validate_version_id(bucket, key, version_id)?;
+        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 {
+                bucket: bucket.to_string(),
+                key: key.to_string(),
+                version_id: version_id.to_string(),
+            });
+        }
+
+        Self::safe_unlink(&data_path).map_err(StorageError::Io)?;
+        Self::safe_unlink(&manifest_path).map_err(StorageError::Io)?;
+        let versions_root = self.bucket_versions_root(bucket);
+        Self::cleanup_empty_parents(&manifest_path, &versions_root);
+        self.stats_cache.remove(bucket);
+        Ok(())
+    }
+
     async fn copy_object(
         &self,
         src_bucket: &str,
@@ -1817,40 +2034,73 @@ impl crate::traits::StorageEngine for FsStorageBackend {
             }
             if let Ok(content) = std::fs::read_to_string(entry.path()) {
                 if let Ok(record) = serde_json::from_str::<Value>(&content) {
-                    let version_id = record
-                        .get("version_id")
-                        .and_then(|v| v.as_str())
-                        .unwrap_or("")
-                        .to_string();
-                    let size = record.get("size").and_then(|v| v.as_u64()).unwrap_or(0);
-                    let archived_at = record
-                        .get("archived_at")
-                        .and_then(|v| v.as_str())
-                        .and_then(|s| DateTime::parse_from_rfc3339(s).ok())
-                        .map(|d| d.with_timezone(&Utc))
-                        .unwrap_or_else(Utc::now);
-                    let etag = record
-                        .get("etag")
-                        .and_then(|v| v.as_str())
-                        .map(|s| s.to_string());
-
-                    versions.push(VersionInfo {
-                        version_id,
-                        key: key.to_string(),
-                        size,
-                        last_modified: archived_at,
-                        etag,
-                        is_latest: false,
-                    });
+                    versions.push(self.version_info_from_record(key, &record));
                 }
             }
         }
 
         versions.sort_by(|a, b| b.last_modified.cmp(&a.last_modified));
-        if let Some(first) = versions.first_mut() {
-            first.is_latest = true;
+
+        Ok(versions)
+    }
+
+    async fn list_bucket_object_versions(
+        &self,
+        bucket: &str,
+        prefix: Option<&str>,
+    ) -> StorageResult<Vec<VersionInfo>> {
+        self.require_bucket(bucket)?;
+        let root = self.bucket_versions_root(bucket);
+        if !root.exists() {
+            return Ok(Vec::new());
         }
 
+        let mut versions = Vec::new();
+        let mut stack = vec![root.clone()];
+        while let Some(current) = stack.pop() {
+            let entries = match std::fs::read_dir(&current) {
+                Ok(entries) => entries,
+                Err(_) => continue,
+            };
+            for entry in entries.flatten() {
+                let path = entry.path();
+                let ft = match entry.file_type() {
+                    Ok(ft) => ft,
+                    Err(_) => continue,
+                };
+                if ft.is_dir() {
+                    stack.push(path);
+                    continue;
+                }
+                if !ft.is_file() || path.extension().and_then(|ext| ext.to_str()) != Some("json") {
+                    continue;
+                }
+                let content = match std::fs::read_to_string(&path) {
+                    Ok(content) => content,
+                    Err(_) => continue,
+                };
+                let record = match serde_json::from_str::<Value>(&content) {
+                    Ok(record) => record,
+                    Err(_) => continue,
+                };
+                let fallback_key = path
+                    .parent()
+                    .and_then(|parent| parent.strip_prefix(&root).ok())
+                    .map(|rel| rel.to_string_lossy().replace('\\', "/"))
+                    .unwrap_or_default();
+                let info = self.version_info_from_record(&fallback_key, &record);
+                if prefix.is_some_and(|value| !info.key.starts_with(value)) {
+                    continue;
+                }
+                versions.push(info);
+            }
+        }
+
+        versions.sort_by(|a, b| {
+            a.key
+                .cmp(&b.key)
+                .then_with(|| b.last_modified.cmp(&a.last_modified))
+        });
         Ok(versions)
     }
 
@@ -2271,6 +2521,12 @@ mod tests {
             .unwrap();
         assert_eq!(versions.len(), 1);
         assert_eq!(versions[0].size, 8);
+
+        let invalid_version = format!("../other/{}", versions[0].version_id);
+        let result = backend
+            .get_object_version("test-bucket", "file.txt", &invalid_version)
+            .await;
+        assert!(matches!(result, Err(StorageError::VersionNotFound { .. })));
     }
 
     #[tokio::test]
diff --git a/rust/myfsio-engine/crates/myfsio-storage/src/traits.rs b/rust/myfsio-engine/crates/myfsio-storage/src/traits.rs
index 32ee5cb..af069b9 100644
--- a/rust/myfsio-engine/crates/myfsio-storage/src/traits.rs
+++ b/rust/myfsio-engine/crates/myfsio-storage/src/traits.rs
@@ -34,8 +34,43 @@ pub trait StorageEngine: Send + Sync {
 
     async fn head_object(&self, bucket: &str, key: &str) -> StorageResult<ObjectMeta>;
 
+    async fn get_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<(ObjectMeta, AsyncReadStream)>;
+
+    async fn get_object_version_path(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<PathBuf>;
+
+    async fn head_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<ObjectMeta>;
+
+    async fn get_object_version_metadata(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<HashMap<String, String>>;
+
     async fn delete_object(&self, bucket: &str, key: &str) -> StorageResult<()>;
 
+    async fn delete_object_version(
+        &self,
+        bucket: &str,
+        key: &str,
+        version_id: &str,
+    ) -> StorageResult<()>;
+
     async fn copy_object(
         &self,
         src_bucket: &str,
@@ -120,6 +155,12 @@ pub trait StorageEngine: Send + Sync {
         key: &str,
     ) -> StorageResult<Vec<VersionInfo>>;
 
+    async fn list_bucket_object_versions(
+        &self,
+        bucket: &str,
+        prefix: Option<&str>,
+    ) -> StorageResult<Vec<VersionInfo>>;
+
     async fn get_object_tags(&self, bucket: &str, key: &str) -> StorageResult<Vec<Tag>>;
 
     async fn set_object_tags(&self, bucket: &str, key: &str, tags: &[Tag]) -> StorageResult<()>;