Skip to main content

fakecloud_s3/service/
mod.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use bytes::Bytes;
5use chrono::{DateTime, Timelike, Utc};
6use http::{HeaderMap, Method, StatusCode};
7use md5::{Digest, Md5};
8
9use fakecloud_aws::arn::Arn;
10use fakecloud_core::delivery::DeliveryBus;
11use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
12use fakecloud_kms::SharedKmsState;
13use fakecloud_persistence::{MemoryS3Store, S3Store, StoreError};
14
15use base64::engine::general_purpose::STANDARD as BASE64;
16use base64::Engine as _;
17
18use crate::logging;
19use crate::state::{AclGrant, S3Bucket, S3Object, SharedS3State};
20
21mod access_points;
22mod acl;
23mod buckets;
24pub(crate) mod config;
25mod lock;
26mod multipart;
27mod notifications;
28mod objects;
29mod tags;
30
31// Re-export notification helpers for use in sub-modules
32#[cfg(test)]
33use notifications::replicate_object;
34pub(super) use notifications::{
35    deliver_notifications, normalize_notification_ids, normalize_replication_xml,
36    replicate_through_store,
37};
38
39// Used only within this file (parse_cors_config)
40use notifications::extract_all_xml_values;
41
42// Re-exports used only in tests
43#[cfg(test)]
44use notifications::{
45    event_matches, key_matches_filters, parse_notification_config, parse_replication_rules,
46    NotificationTargetType,
47};
48
49pub struct S3Service {
50    state: SharedS3State,
51    delivery: Arc<DeliveryBus>,
52    kms_state: Option<SharedKmsState>,
53    pub(crate) kms_hook: Option<Arc<dyn fakecloud_core::delivery::KmsHook>>,
54    store: Arc<dyn S3Store>,
55}
56
57/// Map a [`StoreError`] from the persistence layer to a 500 InternalError
58/// response. Invoked at every mutation site when the write-through persistence
59/// call fails: the in-memory mutation has already happened, but we surface the
60/// failure to the client so they know to retry (and so logs/metrics flag it).
61pub(crate) fn persistence_error(err: StoreError) -> AwsServiceError {
62    AwsServiceError::aws_error(
63        StatusCode::INTERNAL_SERVER_ERROR,
64        "InternalError",
65        format!("persistence store error: {err}"),
66    )
67}
68
69/// Convert a filesystem IO error from a disk-backed body read into an
70/// InternalError response.
71pub(crate) fn io_to_aws(err: std::io::Error) -> AwsServiceError {
72    AwsServiceError::aws_error(
73        StatusCode::INTERNAL_SERVER_ERROR,
74        "InternalError",
75        format!("failed to read object body from disk: {err}"),
76    )
77}
78
79impl S3Service {
80    pub fn new(state: SharedS3State, delivery: Arc<DeliveryBus>) -> Self {
81        Self::with_store(state, delivery, Arc::new(MemoryS3Store::new()))
82    }
83
84    pub fn with_store(
85        state: SharedS3State,
86        delivery: Arc<DeliveryBus>,
87        store: Arc<dyn S3Store>,
88    ) -> Self {
89        Self {
90            state,
91            delivery,
92            kms_state: None,
93            kms_hook: None,
94            store,
95        }
96    }
97
98    pub fn with_kms(mut self, kms_state: SharedKmsState) -> Self {
99        self.kms_state = Some(kms_state);
100        self
101    }
102
103    pub fn with_kms_hook(mut self, hook: Arc<dyn fakecloud_core::delivery::KmsHook>) -> Self {
104        self.kms_hook = Some(hook);
105        self
106    }
107
108    /// Encrypt object body bytes for SSE-KMS storage. Returns ciphertext as
109    /// raw bytes (a UTF-8 fakecloud-kms envelope) on success.
110    ///
111    /// Fail-closed: if the KMS hook reports an error (key denied, key not
112    /// found, etc.), this returns `Err` so PutObject aborts with a 500
113    /// rather than silently storing plaintext. AWS S3 has the same
114    /// behavior — `KMS.NotFoundException` and friends surface as
115    /// `AccessDenied` / `KMS.*` errors back to the caller. When no hook is
116    /// wired (legacy / in-process tests with no KMS dependency), the
117    /// plaintext is returned unchanged so existing tests keep working.
118    pub(crate) fn encrypt_object_body(
119        &self,
120        account_id: &str,
121        region: &str,
122        bucket: &str,
123        plaintext: &[u8],
124        kms_key_id: Option<&str>,
125    ) -> Result<bytes::Bytes, AwsServiceError> {
126        let Some(hook) = &self.kms_hook else {
127            return Ok(bytes::Bytes::copy_from_slice(plaintext));
128        };
129        let key = kms_key_id.filter(|k| !k.is_empty()).unwrap_or("aws/s3");
130        let bucket_arn = Arn::s3(bucket).to_string();
131        let mut ctx = std::collections::HashMap::new();
132        ctx.insert("aws:s3:arn".to_string(), bucket_arn);
133        match hook.encrypt(account_id, region, key, plaintext, "s3.amazonaws.com", ctx) {
134            Ok(envelope) => Ok(bytes::Bytes::from(envelope.into_bytes())),
135            Err(err) => {
136                tracing::warn!(bucket = %bucket, error = %err, "SSE-KMS encrypt failed");
137                Err(AwsServiceError::aws_error(
138                    StatusCode::INTERNAL_SERVER_ERROR,
139                    "KMS.InternalFailureException",
140                    format!("Failed to encrypt object via KMS: {err}"),
141                ))
142            }
143        }
144    }
145
146    /// Decrypt object body bytes that were stored as a fakecloud-kms
147    /// envelope. Caller is expected to gate this on
148    /// `obj.sse_algorithm == Some("aws:kms")`.
149    ///
150    /// Fail-closed: when a hook is wired and the bytes look like an
151    /// envelope but don't decrypt (key revoked, malformed ciphertext),
152    /// this returns `Err` so GetObject surfaces a 500. When no hook is
153    /// wired, or the bytes aren't UTF-8 (legacy snapshots from before
154    /// the hook landed), the bytes are returned unchanged.
155    pub(crate) fn decrypt_object_body(
156        &self,
157        account_id: &str,
158        bucket: &str,
159        ciphertext: &[u8],
160    ) -> Result<bytes::Bytes, AwsServiceError> {
161        let Some(hook) = &self.kms_hook else {
162            return Ok(bytes::Bytes::copy_from_slice(ciphertext));
163        };
164        // Stored envelope is base64 ASCII; non-UTF-8 bytes are pre-hook
165        // legacy snapshots, return as-is.
166        let envelope = match std::str::from_utf8(ciphertext) {
167            Ok(s) => s,
168            Err(_) => return Ok(bytes::Bytes::copy_from_slice(ciphertext)),
169        };
170        let bucket_arn = Arn::s3(bucket).to_string();
171        let mut ctx = std::collections::HashMap::new();
172        ctx.insert("aws:s3:arn".to_string(), bucket_arn);
173        match hook.decrypt(account_id, envelope, "s3.amazonaws.com", ctx) {
174            Ok(bytes) => Ok(bytes::Bytes::from(bytes)),
175            Err(err) => {
176                tracing::warn!(bucket = %bucket, error = %err, "SSE-KMS decrypt failed");
177                Err(AwsServiceError::aws_error(
178                    StatusCode::INTERNAL_SERVER_ERROR,
179                    "KMS.InternalFailureException",
180                    format!("Failed to decrypt object via KMS: {err}"),
181                ))
182            }
183        }
184    }
185}
186
187#[async_trait]
188impl AwsService for S3Service {
189    fn service_name(&self) -> &str {
190        "s3"
191    }
192
193    async fn handle(&self, mut req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
194        // PutObject / UploadPart enter dispatch via the streaming path
195        // with `body = Bytes::new()` and the raw HTTP body available on
196        // `req.body_stream`. Those handlers consume the stream directly
197        // — spooling chunks to disk while computing MD5 + size in
198        // constant memory — so a 1 GiB upload never materializes into
199        // RAM. Every *other* PUT-on-bucket-key the streaming dispatch
200        // flagged (PutObjectTagging, PutObjectAcl, PutObjectRetention,
201        // PutObjectLegalHold, CopyObject, …) reads a small XML/JSON
202        // body from `req.body`, so drain the stream for them.
203        let is_put_to_key = req.method == Method::PUT
204            && req.path_segments.len() >= 2
205            && req
206                .path_segments
207                .first()
208                .map(|s| !s.is_empty())
209                .unwrap_or(false);
210        let q = &req.query_params;
211        let is_put_object = is_put_to_key
212            && !q.contains_key("tagging")
213            && !q.contains_key("acl")
214            && !q.contains_key("retention")
215            && !q.contains_key("legal-hold")
216            && !q.contains_key("renameObject")
217            && !q.contains_key("encryption")
218            && !req.headers.contains_key("x-amz-copy-source");
219        // UploadPart requires both partNumber AND uploadId; checking
220        // only partNumber would skip body draining for stray PUTs that
221        // happened to carry a partNumber query param without being a
222        // real multipart upload.
223        let is_upload_part =
224            is_put_to_key && q.contains_key("partNumber") && q.contains_key("uploadId");
225        if !is_put_object && !is_upload_part {
226            if let Some(stream) = req.take_body_stream() {
227                req.body = fakecloud_core::service::drain_request_stream(stream).await?;
228            }
229        }
230
231        // Access point data plane: resolve alias -> bucket before routing.
232        access_points::resolve_access_point(self, &mut req)?;
233
234        let account_id = req.account_id.as_str();
235
236        // S3 Control endpoint (access point management).
237        let host = req
238            .headers
239            .get("host")
240            .and_then(|v| v.to_str().ok())
241            .unwrap_or("");
242        let is_control = host.to_ascii_lowercase().contains("s3-control");
243        if is_control {
244            let v1 = req.path_segments.first().map(|s| s.as_str());
245            let v2 = req.path_segments.get(1).map(|s| s.as_str());
246            let v3 = req.path_segments.get(2).map(|s| s.as_str());
247            if v1 == Some("v20180820") && v2 == Some("accesspoint") {
248                if let Some(name) = v3 {
249                    match req.method {
250                        Method::PUT => return self.create_access_point(account_id, &req, name),
251                        Method::GET => return self.get_access_point(account_id, &req, name),
252                        Method::DELETE => return self.delete_access_point(account_id, &req, name),
253                        _ => {}
254                    }
255                } else if req.method == Method::GET {
256                    return self.list_access_points(account_id, &req);
257                }
258            }
259        }
260
261        // S3 REST routing: method + path segments + query params
262        //
263        // Reject paths that start with `//` (an empty bucket segment). The
264        // path-segment splitter further down filters empty segments out,
265        // which would otherwise let a malformed URL like `PUT //my-key`
266        // collapse to a valid `PUT /my-key` (CreateBucket) and silently
267        // succeed. Real S3 rejects empty bucket segments with
268        // InvalidBucketName, so mirror that.
269        if req.raw_path.starts_with("//") {
270            return Err(AwsServiceError::aws_error(
271                StatusCode::BAD_REQUEST,
272                "InvalidBucketName",
273                "The specified bucket is not valid: bucket name cannot be empty",
274            ));
275        }
276        let bucket = req.path_segments.first().map(|s| s.as_str());
277        // Extract key from the raw path to preserve leading slashes and empty segments.
278        // The raw path is like "/bucket/key/parts" — we strip the bucket prefix.
279        let key = if let Some(b) = bucket {
280            let prefix = format!("/{b}/");
281            if req.raw_path.starts_with(&prefix) && req.raw_path.len() > prefix.len() {
282                let raw_key = &req.raw_path[prefix.len()..];
283                Some(
284                    percent_encoding::percent_decode_str(raw_key)
285                        .decode_utf8_lossy()
286                        .into_owned(),
287                )
288            } else if req.path_segments.len() > 1 {
289                let raw = req.path_segments[1..].join("/");
290                Some(
291                    percent_encoding::percent_decode_str(&raw)
292                        .decode_utf8_lossy()
293                        .into_owned(),
294                )
295            } else {
296                None
297            }
298        } else {
299            None
300        };
301
302        // Multipart upload operations (checked before main match)
303        if let Some(b) = bucket {
304            // POST /{bucket}/{key}?uploads — CreateMultipartUpload
305            if req.method == Method::POST
306                && key.is_some()
307                && req.query_params.contains_key("uploads")
308            {
309                return self.create_multipart_upload(account_id, &req, b, key.as_deref().unwrap());
310            }
311
312            // POST /{bucket}/{key}?restore
313            if req.method == Method::POST
314                && key.is_some()
315                && req.query_params.contains_key("restore")
316            {
317                return self.restore_object(account_id, &req, b, key.as_deref().unwrap());
318            }
319
320            // POST /{bucket}/{key}?uploadId=X — CompleteMultipartUpload
321            if req.method == Method::POST && key.is_some() {
322                if let Some(upload_id) = req.query_params.get("uploadId").cloned() {
323                    return self.complete_multipart_upload(
324                        account_id,
325                        &req,
326                        b,
327                        key.as_deref().unwrap(),
328                        &upload_id,
329                    );
330                }
331            }
332
333            // PUT /{bucket}/{key}?partNumber=N&uploadId=X — UploadPart or UploadPartCopy
334            if req.method == Method::PUT && key.is_some() {
335                if let (Some(part_num_str), Some(upload_id)) = (
336                    req.query_params.get("partNumber").cloned(),
337                    req.query_params.get("uploadId").cloned(),
338                ) {
339                    if let Ok(part_number) = part_num_str.parse::<i64>() {
340                        if req.headers.contains_key("x-amz-copy-source") {
341                            return self.upload_part_copy(
342                                account_id,
343                                &req,
344                                b,
345                                key.as_deref().unwrap(),
346                                &upload_id,
347                                part_number,
348                            );
349                        }
350                        return self
351                            .upload_part(
352                                account_id,
353                                &req,
354                                b,
355                                key.as_deref().unwrap(),
356                                &upload_id,
357                                part_number,
358                            )
359                            .await;
360                    }
361                }
362            }
363
364            // DELETE /{bucket}/{key}?uploadId=X — AbortMultipartUpload
365            if req.method == Method::DELETE && key.is_some() {
366                if let Some(upload_id) = req.query_params.get("uploadId").cloned() {
367                    return self.abort_multipart_upload(
368                        account_id,
369                        b,
370                        key.as_deref().unwrap(),
371                        &upload_id,
372                    );
373                }
374            }
375
376            // GET /{bucket}?uploads — ListMultipartUploads
377            if req.method == Method::GET
378                && key.is_none()
379                && req.query_params.contains_key("uploads")
380            {
381                return self.list_multipart_uploads(account_id, b, &req.query_params);
382            }
383
384            // GET /{bucket}/{key}?uploadId=X — ListParts
385            if req.method == Method::GET && key.is_some() {
386                if let Some(upload_id) = req.query_params.get("uploadId").cloned() {
387                    return self.list_parts(
388                        account_id,
389                        &req,
390                        b,
391                        key.as_deref().unwrap(),
392                        &upload_id,
393                    );
394                }
395            }
396        }
397
398        // Handle OPTIONS preflight requests (CORS)
399        if req.method == Method::OPTIONS {
400            if let Some(b_name) = bucket {
401                let cors_config = {
402                    let accounts = self.state.read();
403                    let _empty_s3 = crate::state::S3State::new(&req.account_id, &req.region);
404                    let state = accounts.get(&req.account_id).unwrap_or(&_empty_s3);
405                    state
406                        .buckets
407                        .get(b_name)
408                        .and_then(|b| b.cors_config.clone())
409                };
410                if let Some(ref config) = cors_config {
411                    let origin = req
412                        .headers
413                        .get("origin")
414                        .and_then(|v| v.to_str().ok())
415                        .unwrap_or("");
416                    let request_method = req
417                        .headers
418                        .get("access-control-request-method")
419                        .and_then(|v| v.to_str().ok())
420                        .unwrap_or("");
421                    let rules = parse_cors_config(config);
422                    if let Some(rule) = find_cors_rule(&rules, origin, Some(request_method)) {
423                        let mut headers = HeaderMap::new();
424                        let matched_origin = if rule.allowed_origins.contains(&"*".to_string()) {
425                            "*"
426                        } else {
427                            origin
428                        };
429                        headers.insert(
430                            "access-control-allow-origin",
431                            matched_origin
432                                .parse()
433                                .unwrap_or_else(|_| http::HeaderValue::from_static("")),
434                        );
435                        headers.insert(
436                            "access-control-allow-methods",
437                            rule.allowed_methods
438                                .join(", ")
439                                .parse()
440                                .unwrap_or_else(|_| http::HeaderValue::from_static("")),
441                        );
442                        if !rule.allowed_headers.is_empty() {
443                            let ah = if rule.allowed_headers.contains(&"*".to_string()) {
444                                req.headers
445                                    .get("access-control-request-headers")
446                                    .and_then(|v| v.to_str().ok())
447                                    .unwrap_or("*")
448                                    .to_string()
449                            } else {
450                                rule.allowed_headers.join(", ")
451                            };
452                            headers.insert(
453                                "access-control-allow-headers",
454                                ah.parse()
455                                    .unwrap_or_else(|_| http::HeaderValue::from_static("")),
456                            );
457                        }
458                        if let Some(max_age) = rule.max_age_seconds {
459                            headers.insert(
460                                "access-control-max-age",
461                                max_age
462                                    .to_string()
463                                    .parse()
464                                    .unwrap_or_else(|_| http::HeaderValue::from_static("")),
465                            );
466                        }
467                        return Ok(AwsResponse {
468                            status: StatusCode::OK,
469                            content_type: String::new(),
470                            body: Bytes::new().into(),
471                            headers,
472                        });
473                    }
474                }
475                return Err(AwsServiceError::aws_error(
476                    StatusCode::FORBIDDEN,
477                    "CORSResponse",
478                    "CORS is not enabled for this bucket",
479                ));
480            }
481        }
482
483        // Capture origin for CORS response headers
484        let origin_header = req
485            .headers
486            .get("origin")
487            .and_then(|v| v.to_str().ok())
488            .map(|s| s.to_string());
489
490        // Bucket-scoped sub-resource query params. If a request targets one
491        // of these without a bucket in the path, S3 returns an error rather
492        // than silently treating it as a top-level (service) operation.
493        // Real AWS rejects e.g. `GET /?tagging` with a routing/validation
494        // error; mirroring that prevents bucket-required ops from being
495        // accidentally answered by ListBuckets / WriteGetObjectResponse.
496        let bucket_subresources: &[&str] = &[
497            "tagging",
498            "acl",
499            "versioning",
500            "cors",
501            "notification",
502            "website",
503            "accelerate",
504            "publicAccessBlock",
505            "encryption",
506            "lifecycle",
507            "logging",
508            "policy",
509            "policyStatus",
510            "replication",
511            "ownershipControls",
512            "inventory",
513            "analytics",
514            "intelligent-tiering",
515            "metrics",
516            "requestPayment",
517            "abac",
518            "metadataConfiguration",
519            "metadataTable",
520            "metadataInventoryTable",
521            "metadataJournalTable",
522            "object-lock",
523            "versions",
524            "session",
525            "retention",
526            "legal-hold",
527            "attributes",
528            "torrent",
529            "renameObject",
530            "uploads",
531            "restore",
532            "select",
533            // Bucket-required ops the earlier list missed: their query
534            // markers (e.g. `?location` for GetBucketLocation,
535            // `?delete` for DeleteObjects, `?list-type=2` for
536            // ListObjectsV2) signal a bucket op so a request with the
537            // marker but no bucket segment is malformed, not a service-
538            // level call.
539            "location",
540            "delete",
541            "list-type",
542        ];
543        if bucket.is_none()
544            && bucket_subresources
545                .iter()
546                .any(|q| req.query_params.contains_key(*q))
547        {
548            return Err(AwsServiceError::aws_error(
549                StatusCode::BAD_REQUEST,
550                "InvalidBucketName",
551                "The specified bucket is not valid: bucket name is required for this operation",
552            ));
553        }
554
555        let mut result = match (&req.method, bucket, key.as_deref()) {
556            // ListBuckets: GET /
557            (&Method::GET, None, None) => {
558                if req.query_params.get("x-id").map(|s| s.as_str()) == Some("ListDirectoryBuckets")
559                {
560                    self.list_directory_buckets(account_id, &req)
561                } else {
562                    self.list_buckets(account_id, &req)
563                }
564            }
565
566            // Bucket-level operations (no key)
567            (&Method::PUT, Some(b), None) => {
568                if req.query_params.contains_key("tagging") {
569                    self.put_bucket_tagging(account_id, &req, b)
570                } else if req.query_params.contains_key("acl") {
571                    self.put_bucket_acl(account_id, &req, b)
572                } else if req.query_params.contains_key("versioning") {
573                    self.put_bucket_versioning(account_id, &req, b)
574                } else if req.query_params.contains_key("cors") {
575                    self.put_bucket_cors(account_id, &req, b)
576                } else if req.query_params.contains_key("notification") {
577                    self.put_bucket_notification(account_id, &req, b)
578                } else if req.query_params.contains_key("website") {
579                    self.put_bucket_website(account_id, &req, b)
580                } else if req.query_params.contains_key("accelerate") {
581                    self.put_bucket_accelerate(account_id, &req, b)
582                } else if req.query_params.contains_key("publicAccessBlock") {
583                    self.put_public_access_block(account_id, &req, b)
584                } else if req.query_params.contains_key("encryption") {
585                    self.put_bucket_encryption(account_id, &req, b)
586                } else if req.query_params.contains_key("lifecycle") {
587                    self.put_bucket_lifecycle(account_id, &req, b)
588                } else if req.query_params.contains_key("logging") {
589                    self.put_bucket_logging(account_id, &req, b)
590                } else if req.query_params.contains_key("policy") {
591                    self.put_bucket_policy(account_id, &req, b)
592                } else if req.query_params.contains_key("object-lock") {
593                    self.put_object_lock_config(account_id, &req, b)
594                } else if req.query_params.contains_key("replication") {
595                    self.put_bucket_replication(account_id, &req, b)
596                } else if req.query_params.contains_key("ownershipControls") {
597                    self.put_bucket_ownership_controls(account_id, &req, b)
598                } else if req.query_params.contains_key("inventory") {
599                    self.put_bucket_inventory(account_id, &req, b)
600                } else if req.query_params.contains_key("analytics") {
601                    self.put_bucket_analytics_config(account_id, &req, b)
602                } else if req.query_params.contains_key("intelligent-tiering") {
603                    self.put_bucket_intelligent_tiering_config(account_id, &req, b)
604                } else if req.query_params.contains_key("metrics") {
605                    self.put_bucket_metrics_config(account_id, &req, b)
606                } else if req.query_params.contains_key("requestPayment") {
607                    self.put_bucket_request_payment(account_id, &req, b)
608                } else if req.query_params.contains_key("abac") {
609                    self.put_bucket_abac(account_id, &req, b)
610                } else if req.query_params.contains_key("metadataInventoryTable") {
611                    self.update_bucket_metadata_inventory_table(account_id, &req, b)
612                } else if req.query_params.contains_key("metadataJournalTable") {
613                    self.update_bucket_metadata_journal_table(account_id, &req, b)
614                } else {
615                    self.create_bucket(account_id, &req, b)
616                }
617            }
618            (&Method::DELETE, Some(b), None) => {
619                if req.query_params.contains_key("tagging") {
620                    self.delete_bucket_tagging(account_id, &req, b)
621                } else if req.query_params.contains_key("cors") {
622                    self.delete_bucket_cors(account_id, b)
623                } else if req.query_params.contains_key("website") {
624                    self.delete_bucket_website(account_id, b)
625                } else if req.query_params.contains_key("publicAccessBlock") {
626                    self.delete_public_access_block(account_id, b)
627                } else if req.query_params.contains_key("encryption") {
628                    self.delete_bucket_encryption(account_id, b)
629                } else if req.query_params.contains_key("lifecycle") {
630                    self.delete_bucket_lifecycle(account_id, b)
631                } else if req.query_params.contains_key("policy") {
632                    self.delete_bucket_policy(account_id, b)
633                } else if req.query_params.contains_key("replication") {
634                    self.delete_bucket_replication(account_id, b)
635                } else if req.query_params.contains_key("ownershipControls") {
636                    self.delete_bucket_ownership_controls(account_id, b)
637                } else if req.query_params.contains_key("inventory") {
638                    self.delete_bucket_inventory(account_id, &req, b)
639                } else if req.query_params.contains_key("analytics") {
640                    self.delete_bucket_analytics_config(account_id, &req, b)
641                } else if req.query_params.contains_key("intelligent-tiering") {
642                    self.delete_bucket_intelligent_tiering_config(account_id, &req, b)
643                } else if req.query_params.contains_key("metrics") {
644                    self.delete_bucket_metrics_config(account_id, &req, b)
645                } else if req.query_params.contains_key("metadataConfiguration") {
646                    self.delete_bucket_metadata_config(account_id, b)
647                } else if req.query_params.contains_key("metadataTable") {
648                    self.delete_bucket_metadata_table_config(account_id, b)
649                } else {
650                    self.delete_bucket(account_id, &req, b)
651                }
652            }
653            (&Method::HEAD, Some(b), None) => self.head_bucket(account_id, b),
654            (&Method::GET, Some(b), None) => {
655                if req.query_params.contains_key("tagging") {
656                    self.get_bucket_tagging(account_id, &req, b)
657                } else if req.query_params.contains_key("location") {
658                    self.get_bucket_location(account_id, b)
659                } else if req.query_params.contains_key("acl") {
660                    self.get_bucket_acl(account_id, &req, b)
661                } else if req.query_params.contains_key("versioning") {
662                    self.get_bucket_versioning(account_id, b)
663                } else if req.query_params.contains_key("versions") {
664                    self.list_object_versions(account_id, &req, b)
665                } else if req.query_params.contains_key("object-lock") {
666                    self.get_object_lock_configuration(account_id, b)
667                } else if req.query_params.contains_key("cors") {
668                    self.get_bucket_cors(account_id, b)
669                } else if req.query_params.contains_key("notification") {
670                    self.get_bucket_notification(account_id, b)
671                } else if req.query_params.contains_key("website") {
672                    self.get_bucket_website(account_id, b)
673                } else if req.query_params.contains_key("accelerate") {
674                    self.get_bucket_accelerate(account_id, b)
675                } else if req.query_params.contains_key("publicAccessBlock") {
676                    self.get_public_access_block(account_id, b)
677                } else if req.query_params.contains_key("encryption") {
678                    self.get_bucket_encryption(account_id, b)
679                } else if req.query_params.contains_key("lifecycle") {
680                    self.get_bucket_lifecycle(account_id, b)
681                } else if req.query_params.contains_key("logging") {
682                    self.get_bucket_logging(account_id, b)
683                } else if req.query_params.contains_key("policy") {
684                    self.get_bucket_policy(account_id, b)
685                } else if req.query_params.contains_key("replication") {
686                    self.get_bucket_replication(account_id, b)
687                } else if req.query_params.contains_key("ownershipControls") {
688                    self.get_bucket_ownership_controls(account_id, b)
689                } else if req.query_params.contains_key("inventory") {
690                    if req.query_params.contains_key("id") {
691                        self.get_bucket_inventory(account_id, &req, b)
692                    } else {
693                        self.list_bucket_inventory_configurations(account_id, b)
694                    }
695                } else if req.query_params.contains_key("analytics") {
696                    if req.query_params.contains_key("id") {
697                        self.get_bucket_analytics_config(account_id, &req, b)
698                    } else {
699                        self.list_bucket_analytics_configurations(account_id, b)
700                    }
701                } else if req.query_params.contains_key("intelligent-tiering") {
702                    if req.query_params.contains_key("id") {
703                        self.get_bucket_intelligent_tiering_config(account_id, &req, b)
704                    } else {
705                        self.list_bucket_intelligent_tiering_configurations(account_id, b)
706                    }
707                } else if req.query_params.contains_key("metrics") {
708                    if req.query_params.contains_key("id") {
709                        self.get_bucket_metrics_config(account_id, &req, b)
710                    } else {
711                        self.list_bucket_metrics_configurations(account_id, b)
712                    }
713                } else if req.query_params.contains_key("requestPayment") {
714                    self.get_bucket_request_payment(account_id, b)
715                } else if req.query_params.contains_key("abac") {
716                    self.get_bucket_abac(account_id, b)
717                } else if req.query_params.contains_key("policyStatus") {
718                    self.get_bucket_policy_status(account_id, b)
719                } else if req.query_params.contains_key("metadataConfiguration") {
720                    self.get_bucket_metadata_config(account_id, b)
721                } else if req.query_params.contains_key("metadataTable") {
722                    self.get_bucket_metadata_table_config(account_id, b)
723                } else if req.query_params.contains_key("session") {
724                    self.create_session(account_id, &req, b)
725                } else if req.query_params.get("list-type").map(|s| s.as_str()) == Some("2") {
726                    self.list_objects_v2(account_id, &req, b)
727                } else if req.query_params.is_empty() {
728                    // If bucket has website config and no query params, serve index document
729                    let website_config = {
730                        let accounts = self.state.read();
731                        let _empty_s3 = crate::state::S3State::new(&req.account_id, &req.region);
732                        let state = accounts.get(&req.account_id).unwrap_or(&_empty_s3);
733                        state
734                            .buckets
735                            .get(b)
736                            .and_then(|bkt| bkt.website_config.clone())
737                    };
738                    if let Some(ref config) = website_config {
739                        if let Some(index_doc) = extract_xml_value(config, "Suffix").or_else(|| {
740                            extract_xml_value(config, "IndexDocument").and_then(|inner| {
741                                let open = "<Suffix>";
742                                let close = "</Suffix>";
743                                let s = inner.find(open)? + open.len();
744                                let e = inner.find(close)?;
745                                Some(inner[s..e].trim().to_string())
746                            })
747                        }) {
748                            self.serve_website_object(account_id, &req, b, &index_doc, config)
749                        } else {
750                            self.list_objects_v1(account_id, &req, b)
751                        }
752                    } else {
753                        self.list_objects_v1(account_id, &req, b)
754                    }
755                } else {
756                    self.list_objects_v1(account_id, &req, b)
757                }
758            }
759
760            // Object-level operations
761            (&Method::PUT, Some(b), Some(k)) => {
762                if req.query_params.contains_key("tagging") {
763                    self.put_object_tagging(account_id, &req, b, k)
764                } else if req.query_params.contains_key("acl") {
765                    self.put_object_acl(account_id, &req, b, k)
766                } else if req.query_params.contains_key("retention") {
767                    self.put_object_retention(account_id, &req, b, k)
768                } else if req.query_params.contains_key("legal-hold") {
769                    self.put_object_legal_hold(account_id, &req, b, k)
770                } else if req.query_params.contains_key("renameObject") {
771                    self.rename_object(account_id, &req, b, k)
772                } else if req.query_params.contains_key("encryption") {
773                    self.update_object_encryption(account_id, &req, b, k)
774                } else if req.headers.contains_key("x-amz-copy-source") {
775                    self.copy_object(account_id, &req, b, k)
776                } else {
777                    self.put_object(account_id, &req, b, k).await
778                }
779            }
780            (&Method::GET, Some(b), Some(k)) => {
781                if req.query_params.contains_key("tagging") {
782                    self.get_object_tagging(account_id, &req, b, k)
783                } else if req.query_params.contains_key("acl") {
784                    self.get_object_acl(account_id, &req, b, k)
785                } else if req.query_params.contains_key("retention") {
786                    self.get_object_retention(account_id, &req, b, k)
787                } else if req.query_params.contains_key("legal-hold") {
788                    self.get_object_legal_hold(account_id, &req, b, k)
789                } else if req.query_params.contains_key("attributes") {
790                    self.get_object_attributes(account_id, &req, b, k)
791                } else if req.query_params.contains_key("torrent") {
792                    self.get_object_torrent(account_id, &req, b, k)
793                } else {
794                    let result = self.get_object(account_id, &req, b, k);
795                    // If object not found and bucket has website config, serve error document
796                    let is_not_found = matches!(
797                        &result,
798                        Err(e) if e.code() == "NoSuchKey"
799                    );
800                    if is_not_found {
801                        let website_config = {
802                            let accounts = self.state.read();
803                            let _empty_s3 =
804                                crate::state::S3State::new(&req.account_id, &req.region);
805                            let state = accounts.get(&req.account_id).unwrap_or(&_empty_s3);
806                            state
807                                .buckets
808                                .get(b)
809                                .and_then(|bkt| bkt.website_config.clone())
810                        };
811                        if let Some(ref config) = website_config {
812                            if let Some(error_key) = extract_xml_value(config, "ErrorDocument")
813                                .and_then(|inner| {
814                                    let open = "<Key>";
815                                    let close = "</Key>";
816                                    let s = inner.find(open)? + open.len();
817                                    let e = inner.find(close)?;
818                                    Some(inner[s..e].trim().to_string())
819                                })
820                                .or_else(|| extract_xml_value(config, "Key"))
821                            {
822                                return self.serve_website_error(account_id, &req, b, &error_key);
823                            }
824                        }
825                    }
826                    result
827                }
828            }
829            (&Method::DELETE, Some(b), Some(k)) => {
830                if req.query_params.contains_key("tagging") {
831                    self.delete_object_tagging(account_id, b, k)
832                } else {
833                    self.delete_object(account_id, &req, b, k)
834                }
835            }
836            (&Method::HEAD, Some(b), Some(k)) => self.head_object(account_id, &req, b, k),
837
838            // POST /{bucket}?delete — batch delete
839            (&Method::POST, Some(b), None) if req.query_params.contains_key("delete") => {
840                self.delete_objects(account_id, &req, b)
841            }
842            (&Method::POST, Some(b), None)
843                if req.query_params.contains_key("metadataConfiguration") =>
844            {
845                self.create_bucket_metadata_config(account_id, &req, b)
846            }
847            (&Method::POST, Some(b), None) if req.query_params.contains_key("metadataTable") => {
848                self.create_bucket_metadata_table_config(account_id, &req, b)
849            }
850            (&Method::POST, Some(b), Some(k))
851                if req.query_params.get("select-type").map(|s| s.as_str()) == Some("2") =>
852            {
853                self.select_object_content(account_id, &req, b, k)
854            }
855            (&Method::POST, Some("WriteGetObjectResponse"), None) => {
856                self.write_get_object_response(account_id, &req)
857            }
858
859            _ => Err(AwsServiceError::aws_error(
860                StatusCode::METHOD_NOT_ALLOWED,
861                "MethodNotAllowed",
862                "The specified method is not allowed against this resource",
863            )),
864        };
865
866        // Apply CORS headers to the response if Origin was present
867        if let (Some(ref origin), Some(b_name)) = (&origin_header, bucket) {
868            let cors_config = {
869                let accounts = self.state.read();
870                let _empty_s3 = crate::state::S3State::new(&req.account_id, &req.region);
871                let state = accounts.get(&req.account_id).unwrap_or(&_empty_s3);
872                state
873                    .buckets
874                    .get(b_name)
875                    .and_then(|b| b.cors_config.clone())
876            };
877            if let Some(ref config) = cors_config {
878                let rules = parse_cors_config(config);
879                if let Some(rule) = find_cors_rule(&rules, origin, None) {
880                    if let Ok(ref mut resp) = result {
881                        let matched_origin = if rule.allowed_origins.contains(&"*".to_string()) {
882                            "*"
883                        } else {
884                            origin
885                        };
886                        resp.headers.insert(
887                            "access-control-allow-origin",
888                            matched_origin
889                                .parse()
890                                .unwrap_or_else(|_| http::HeaderValue::from_static("")),
891                        );
892                        if !rule.expose_headers.is_empty() {
893                            resp.headers.insert(
894                                "access-control-expose-headers",
895                                rule.expose_headers
896                                    .join(", ")
897                                    .parse()
898                                    .unwrap_or_else(|_| http::HeaderValue::from_static("")),
899                            );
900                        }
901                    }
902                }
903            }
904        }
905
906        // Write S3 access log entry if the source bucket has logging enabled
907        if let Some(b_name) = bucket {
908            let status_code = match &result {
909                Ok(resp) => resp.status.as_u16(),
910                Err(e) => e.status().as_u16(),
911            };
912            let op = logging::operation_name(&req.method, key.as_deref());
913            logging::maybe_write_access_log(
914                &self.state,
915                &self.store,
916                b_name,
917                &logging::AccessLogRequest {
918                    operation: op,
919                    key: key.as_deref(),
920                    status: status_code,
921                    request_id: &req.request_id,
922                    method: req.method.as_str(),
923                    path: &req.raw_path,
924                },
925            );
926        }
927
928        result
929    }
930
931    fn supported_actions(&self) -> &[&str] {
932        &[
933            // Buckets
934            "ListBuckets",
935            "CreateBucket",
936            "DeleteBucket",
937            "HeadBucket",
938            "GetBucketLocation",
939            // Objects
940            "PutObject",
941            "GetObject",
942            "DeleteObject",
943            "HeadObject",
944            "CopyObject",
945            "DeleteObjects",
946            "ListObjectsV2",
947            "ListObjects",
948            "ListObjectVersions",
949            "GetObjectAttributes",
950            "RestoreObject",
951            // Object properties
952            "PutObjectTagging",
953            "GetObjectTagging",
954            "DeleteObjectTagging",
955            "PutObjectAcl",
956            "GetObjectAcl",
957            "PutObjectRetention",
958            "GetObjectRetention",
959            "PutObjectLegalHold",
960            "GetObjectLegalHold",
961            // Bucket configuration
962            "PutBucketTagging",
963            "GetBucketTagging",
964            "DeleteBucketTagging",
965            "PutBucketAcl",
966            "GetBucketAcl",
967            "PutBucketVersioning",
968            "GetBucketVersioning",
969            "PutBucketCors",
970            "GetBucketCors",
971            "DeleteBucketCors",
972            "PutBucketNotificationConfiguration",
973            "GetBucketNotificationConfiguration",
974            "PutBucketWebsite",
975            "GetBucketWebsite",
976            "DeleteBucketWebsite",
977            "PutBucketAccelerateConfiguration",
978            "GetBucketAccelerateConfiguration",
979            "PutPublicAccessBlock",
980            "GetPublicAccessBlock",
981            "DeletePublicAccessBlock",
982            "PutBucketEncryption",
983            "GetBucketEncryption",
984            "DeleteBucketEncryption",
985            "PutBucketLifecycleConfiguration",
986            "GetBucketLifecycleConfiguration",
987            "DeleteBucketLifecycle",
988            "PutBucketLogging",
989            "GetBucketLogging",
990            "PutBucketPolicy",
991            "GetBucketPolicy",
992            "DeleteBucketPolicy",
993            "PutObjectLockConfiguration",
994            "GetObjectLockConfiguration",
995            "PutBucketReplication",
996            "GetBucketReplication",
997            "DeleteBucketReplication",
998            "PutBucketOwnershipControls",
999            "GetBucketOwnershipControls",
1000            "DeleteBucketOwnershipControls",
1001            "PutBucketInventoryConfiguration",
1002            "GetBucketInventoryConfiguration",
1003            "DeleteBucketInventoryConfiguration",
1004            "ListBucketInventoryConfigurations",
1005            "PutBucketAnalyticsConfiguration",
1006            "GetBucketAnalyticsConfiguration",
1007            "DeleteBucketAnalyticsConfiguration",
1008            "ListBucketAnalyticsConfigurations",
1009            "PutBucketIntelligentTieringConfiguration",
1010            "GetBucketIntelligentTieringConfiguration",
1011            "DeleteBucketIntelligentTieringConfiguration",
1012            "ListBucketIntelligentTieringConfigurations",
1013            "PutBucketMetricsConfiguration",
1014            "GetBucketMetricsConfiguration",
1015            "DeleteBucketMetricsConfiguration",
1016            "ListBucketMetricsConfigurations",
1017            "PutBucketRequestPayment",
1018            "GetBucketRequestPayment",
1019            "PutBucketAbac",
1020            "GetBucketAbac",
1021            "GetBucketPolicyStatus",
1022            "CreateBucketMetadataConfiguration",
1023            "GetBucketMetadataConfiguration",
1024            "DeleteBucketMetadataConfiguration",
1025            "CreateBucketMetadataTableConfiguration",
1026            "GetBucketMetadataTableConfiguration",
1027            "DeleteBucketMetadataTableConfiguration",
1028            "UpdateBucketMetadataInventoryTableConfiguration",
1029            "UpdateBucketMetadataJournalTableConfiguration",
1030            "GetObjectTorrent",
1031            "RenameObject",
1032            "SelectObjectContent",
1033            "UpdateObjectEncryption",
1034            "WriteGetObjectResponse",
1035            "ListDirectoryBuckets",
1036            "CreateSession",
1037            // Multipart uploads
1038            "CreateMultipartUpload",
1039            "UploadPart",
1040            "UploadPartCopy",
1041            "CompleteMultipartUpload",
1042            "AbortMultipartUpload",
1043            "ListParts",
1044            "ListMultipartUploads",
1045        ]
1046    }
1047
1048    fn iam_enforceable(&self) -> bool {
1049        true
1050    }
1051
1052    /// S3 resources are either:
1053    /// - Bucket ARN (`arn:aws:s3:::bucket`) for bucket-level actions
1054    /// - Object ARN (`arn:aws:s3:::bucket/key`) for object-level actions
1055    /// - Wildcard (`*`) for `ListBuckets` which doesn't target a specific
1056    ///   resource
1057    ///
1058    /// S3 ARNs notably omit the account id and region — this is the one
1059    /// AWS service that carries neither in its ARN, because bucket names
1060    /// are globally unique.
1061    fn iam_action_for(&self, request: &AwsRequest) -> Option<fakecloud_core::auth::IamAction> {
1062        // S3 doesn't set `request.action` — the handler dispatches on
1063        // method + path + query params directly. Re-derive the action
1064        // name here so enforcement can match against IAM policies the
1065        // same way the real service would.
1066        let bucket = request.path_segments.first().map(|s| s.as_str());
1067        let key = if request.path_segments.len() > 1 {
1068            Some(request.path_segments[1..].join("/"))
1069        } else {
1070            None
1071        };
1072        // S3 Control access-point management (host `s3-control...`, path
1073        // `/v20180820/accesspoint[/name]`) is authorized under the dedicated
1074        // s3:CreateAccessPoint / GetAccessPoint / DeleteAccessPoint /
1075        // ListAccessPoints actions — NOT the object-level GetObject/PutObject/
1076        // DeleteObject the generic method-based detection would pick, which made
1077        // access-point policies ineffective.
1078        let host = request
1079            .headers
1080            .get("host")
1081            .and_then(|v| v.to_str().ok())
1082            .unwrap_or("");
1083        if host.to_ascii_lowercase().contains("s3-control")
1084            && request.path_segments.first().map(|s| s.as_str()) == Some("v20180820")
1085            && request.path_segments.get(1).map(|s| s.as_str()) == Some("accesspoint")
1086        {
1087            let has_name = request.path_segments.get(2).is_some();
1088            let ap_action = match (request.method.as_str(), has_name) {
1089                ("PUT", true) => Some("CreateAccessPoint"),
1090                ("GET", true) => Some("GetAccessPoint"),
1091                ("DELETE", true) => Some("DeleteAccessPoint"),
1092                ("GET", false) => Some("ListAccessPoints"),
1093                _ => None,
1094            };
1095            if let Some(action) = ap_action {
1096                let resource = request
1097                    .path_segments
1098                    .get(2)
1099                    .map(|name| format!("arn:aws:s3:::accesspoint/{name}"))
1100                    .unwrap_or_else(|| "*".to_string());
1101                return Some(fakecloud_core::auth::IamAction {
1102                    service: "s3",
1103                    action,
1104                    resource,
1105                });
1106            }
1107        }
1108        let action = s3_detect_action(
1109            request.method.as_str(),
1110            bucket,
1111            key.as_deref(),
1112            &request.query_params,
1113        )?;
1114        // A handful of S3-adjacent ops are authorized under sibling IAM
1115        // service prefixes, not `s3:`. Select the right prefix so a policy
1116        // targeting the real action string actually matches.
1117        let service = match action {
1118            "CreateSession" | "ListAllMyDirectoryBuckets" => "s3express",
1119            "WriteGetObjectResponse" => "s3-object-lambda",
1120            _ => "s3",
1121        };
1122        let resource = s3_resource_for(action, bucket, key.as_deref());
1123        Some(fakecloud_core::auth::IamAction {
1124            service,
1125            action,
1126            resource,
1127        })
1128    }
1129
1130    fn iam_condition_keys_for(
1131        &self,
1132        request: &AwsRequest,
1133        action: &fakecloud_core::auth::IamAction,
1134    ) -> std::collections::BTreeMap<String, Vec<String>> {
1135        s3_condition_keys(action.action, &request.query_params)
1136    }
1137
1138    fn resource_tags_for(
1139        &self,
1140        resource_arn: &str,
1141    ) -> Option<std::collections::HashMap<String, String>> {
1142        s3_resource_tags(&self.state, resource_arn)
1143            .map(|m| m.into_iter().collect::<std::collections::HashMap<_, _>>())
1144    }
1145
1146    fn request_tags_from(
1147        &self,
1148        request: &AwsRequest,
1149        action: &str,
1150    ) -> Option<std::collections::HashMap<String, String>> {
1151        s3_request_tags(request, action)
1152            .map(|m| m.into_iter().collect::<std::collections::HashMap<_, _>>())
1153    }
1154}
1155
1156/// Extract service-specific IAM condition keys from an S3 request.
1157///
1158/// Today only `ListObjects` / `ListObjectsV2` expose keys (`s3:prefix`,
1159/// `s3:delimiter`, `s3:max-keys`) via their query params. Other actions
1160/// return an empty map so the evaluator's safe-fail semantics treat any
1161/// policy condition referencing an unknown key as "doesn't apply".
1162fn s3_condition_keys(
1163    action: &str,
1164    query: &std::collections::HashMap<String, String>,
1165) -> std::collections::BTreeMap<String, Vec<String>> {
1166    let mut out = std::collections::BTreeMap::new();
1167    if matches!(action, "ListObjects" | "ListObjectsV2") {
1168        // Both list variants share the same query param shape.
1169        if let Some(prefix) = query.get("prefix") {
1170            out.insert("s3:prefix".to_string(), vec![prefix.clone()]);
1171        }
1172        if let Some(delimiter) = query.get("delimiter") {
1173            out.insert("s3:delimiter".to_string(), vec![delimiter.clone()]);
1174        }
1175        if let Some(max_keys) = query.get("max-keys") {
1176            out.insert("s3:max-keys".to_string(), vec![max_keys.clone()]);
1177        }
1178    }
1179    out
1180}
1181
1182/// Look up resource tags for an S3 ARN.
1183///
1184/// Bucket-level ARN (`arn:aws:s3:::bucket`) -> bucket tags.
1185/// Object-level ARN (`arn:aws:s3:::bucket/key`) -> object tags.
1186/// `*` (ListBuckets) -> `Some(empty)` (no resource to tag).
1187fn s3_resource_tags(
1188    state: &SharedS3State,
1189    resource_arn: &str,
1190) -> Option<std::collections::BTreeMap<String, String>> {
1191    if resource_arn == "*" {
1192        return Some(std::collections::BTreeMap::new());
1193    }
1194    // S3 ARNs: arn:aws:s3:::bucket or arn:aws:s3:::bucket/key
1195    let after_prefix = resource_arn.strip_prefix("arn:aws:s3:::")?;
1196    let mas = state.read();
1197    // S3 bucket names are globally unique; scan all accounts to find the bucket
1198    let bucket_name = after_prefix.split('/').next().unwrap_or(after_prefix);
1199    let state = mas
1200        .find_account(|s| s.buckets.contains_key(bucket_name))
1201        .and_then(|id| mas.get(id))
1202        .or_else(|| Some(mas.default_ref()))?;
1203    if let Some(slash_pos) = after_prefix.find('/') {
1204        // Object-level: bucket/key
1205        let bucket_name = &after_prefix[..slash_pos];
1206        let key = &after_prefix[slash_pos + 1..];
1207        let bucket = state.buckets.get(bucket_name)?;
1208        // Try current objects first, then versioned objects (latest version)
1209        if let Some(obj) = bucket.objects.get(key) {
1210            Some(obj.tags.clone())
1211        } else if let Some(versions) = bucket.object_versions.get(key) {
1212            versions.last().map(|v| v.tags.clone())
1213        } else {
1214            // Object doesn't exist yet (e.g. PutObject creating it)
1215            Some(std::collections::BTreeMap::new())
1216        }
1217    } else {
1218        // Bucket-level
1219        let bucket = state.buckets.get(after_prefix)?;
1220        Some(bucket.tags.clone())
1221    }
1222}
1223
1224/// Extract tags from an S3 request body/headers.
1225///
1226/// S3 sends tags via:
1227/// - `x-amz-tagging` header (URL-encoded `key=value&...`) on PutObject
1228/// - XML body on PutBucketTagging / PutObjectTagging
1229fn s3_request_tags(
1230    request: &AwsRequest,
1231    action: &str,
1232) -> Option<std::collections::BTreeMap<String, String>> {
1233    match action {
1234        "PutObject" | "CopyObject" | "CreateMultipartUpload" => {
1235            // Tags come via x-amz-tagging header
1236            if let Some(tagging) = request.headers.get("x-amz-tagging") {
1237                let tags = parse_url_encoded_tags(tagging.to_str().unwrap_or(""));
1238                Some(tags.into_iter().collect())
1239            } else {
1240                Some(std::collections::BTreeMap::new())
1241            }
1242        }
1243        "PutBucketTagging" | "PutObjectTagging" => {
1244            // Tags come in XML body
1245            let body = std::str::from_utf8(&request.body).unwrap_or("");
1246            let tags = parse_tagging_xml(body);
1247            Some(tags.into_iter().collect())
1248        }
1249        _ => Some(std::collections::BTreeMap::new()),
1250    }
1251}
1252
1253/// Derive the IAM action name from an S3 REST request. Handles the
1254/// common cases (GetObject, PutObject, DeleteObject, ListObjectsV2,
1255/// CreateBucket, ...) plus a subset of sub-resource operations
1256/// (`?acl`, `?tagging`, `?versioning`, `?policy`, `?cors`, `?website`,
1257/// `?lifecycle`, `?encryption`, `?logging`, `?notification`, `?replication`,
1258/// `?ownershipControls`, `?publicAccessBlock`, `?accelerate`, `?inventory`,
1259/// `?object-lock`, `?uploads`, `?uploadId`).
1260///
1261/// Returns `None` for requests that don't map to a known action — the
1262/// dispatch layer then skips enforcement for that request rather than
1263/// guessing (and a warn log fires via the "service is iam_enforceable
1264/// but has no mapping" branch in dispatch.rs).
1265fn s3_detect_action(
1266    method: &str,
1267    bucket: Option<&str>,
1268    key: Option<&str>,
1269    query: &std::collections::HashMap<String, String>,
1270) -> Option<&'static str> {
1271    let has = |q: &str| query.contains_key(q);
1272    let is_get = method == "GET";
1273    let is_put = method == "PUT";
1274    let is_post = method == "POST";
1275    let is_delete = method == "DELETE";
1276
1277    // Service root
1278    if bucket.is_none() {
1279        return match method {
1280            // A directory-bucket listing (`GET /?x-id=ListDirectoryBuckets`) is
1281            // authorized under `s3express:ListAllMyDirectoryBuckets`, NOT
1282            // `s3:ListBuckets`. Detecting it as ListBuckets let an
1283            // `s3:ListBuckets` grant wrongly enumerate directory buckets and
1284            // denied a policy that only granted the s3express action (bug-hunt
1285            // 2026-06-24, 5.5).
1286            "GET" if query.get("x-id").map(|s| s.as_str()) == Some("ListDirectoryBuckets") => {
1287                Some("ListAllMyDirectoryBuckets")
1288            }
1289            "GET" => Some("ListBuckets"),
1290            _ => None,
1291        };
1292    }
1293
1294    // WriteGetObjectResponse is an Object Lambda data-plane op routed as
1295    // `POST /WriteGetObjectResponse` (the literal value occupies the bucket
1296    // segment). AWS authorizes it under the separate
1297    // `s3-object-lambda:WriteGetObjectResponse` action — handled in
1298    // `iam_action_for`, which selects the `s3-object-lambda` service prefix
1299    // for this op. Previously unmapped -> enforcement skipped.
1300    if is_post && bucket == Some("WriteGetObjectResponse") && key.is_none() {
1301        return Some("WriteGetObjectResponse");
1302    }
1303
1304    let has_key = key.is_some();
1305
1306    // Multipart sub-resource forms
1307    if has_key && is_post && has("uploads") {
1308        return Some("CreateMultipartUpload");
1309    }
1310    if has_key && is_post && has("uploadId") {
1311        return Some("CompleteMultipartUpload");
1312    }
1313    if has_key && is_put && has("partNumber") && has("uploadId") {
1314        return Some("UploadPart");
1315    }
1316    if has_key && is_delete && has("uploadId") {
1317        return Some("AbortMultipartUpload");
1318    }
1319    if has_key && is_get && has("uploadId") {
1320        return Some("ListParts");
1321    }
1322    if !has_key && is_get && has("uploads") {
1323        return Some("ListMultipartUploads");
1324    }
1325
1326    // Sub-resource-keyed actions (?acl, ?tagging, ...). Order matters
1327    // since a request can carry multiple; we pick the most specific.
1328    // Object-level sub-resources come first (key present).
1329    if has_key {
1330        if has("tagging") {
1331            return Some(match method {
1332                "GET" => "GetObjectTagging",
1333                "PUT" => "PutObjectTagging",
1334                "DELETE" => "DeleteObjectTagging",
1335                _ => return None,
1336            });
1337        }
1338        if has("acl") {
1339            return Some(match method {
1340                "GET" => "GetObjectAcl",
1341                "PUT" => "PutObjectAcl",
1342                _ => return None,
1343            });
1344        }
1345        if has("retention") {
1346            return Some(match method {
1347                "GET" => "GetObjectRetention",
1348                "PUT" => "PutObjectRetention",
1349                _ => return None,
1350            });
1351        }
1352        if has("legal-hold") {
1353            return Some(match method {
1354                "GET" => "GetObjectLegalHold",
1355                "PUT" => "PutObjectLegalHold",
1356                _ => return None,
1357            });
1358        }
1359        // Identified by cubic on PR #399: both ?attributes and ?restore
1360        // need method guards — otherwise e.g. GET /bucket/key?restore
1361        // would be classified as RestoreObject (POST-only in AWS) and
1362        // IAM-evaluated against s3:RestoreObject instead of s3:GetObject.
1363        if has("attributes") && is_get {
1364            return Some("GetObjectAttributes");
1365        }
1366        if has("restore") && is_post {
1367            return Some("RestoreObject");
1368        }
1369        // RenameObject is an object-level op (`PUT /bucket/newkey?renameObject`
1370        // + `x-amz-rename-source`). It carries a key, so it must be detected
1371        // here — the `!has_key` arm below never sees it. Without this it fell
1372        // through to `PutObject`, so a policy denying `s3:RenameObject` was
1373        // silently ineffective.
1374        if has("renameObject") && is_put {
1375            return Some("RenameObject");
1376        }
1377        // UpdateObjectEncryption (`PUT /bucket/key?encryption`) changes an
1378        // existing object's server-side encryption. AWS gates this under
1379        // `s3:PutObject` (encryption is a write attribute), and there is no
1380        // distinct public `s3:UpdateObjectEncryption` IAM action — so map to
1381        // PutObject deliberately rather than letting it fall through.
1382        if has("encryption") && is_put {
1383            return Some("PutObject");
1384        }
1385        // SelectObjectContent (`POST /bucket/key?select&select-type=2`) reads
1386        // object data, so AWS authorizes it with `s3:GetObject`. Previously
1387        // unmapped -> `_ => None` -> enforcement skipped entirely.
1388        if has("select-type") && is_post {
1389            return Some("GetObject");
1390        }
1391    }
1392
1393    // Bucket-level sub-resources (key absent).
1394    if !has_key {
1395        if has("tagging") {
1396            return Some(match method {
1397                "GET" => "GetBucketTagging",
1398                "PUT" => "PutBucketTagging",
1399                "DELETE" => "DeleteBucketTagging",
1400                _ => return None,
1401            });
1402        }
1403        if has("acl") {
1404            return Some(match method {
1405                "GET" => "GetBucketAcl",
1406                "PUT" => "PutBucketAcl",
1407                _ => return None,
1408            });
1409        }
1410        if has("versioning") {
1411            return Some(match method {
1412                "GET" => "GetBucketVersioning",
1413                "PUT" => "PutBucketVersioning",
1414                _ => return None,
1415            });
1416        }
1417        if has("cors") {
1418            return Some(match method {
1419                "GET" => "GetBucketCors",
1420                "PUT" => "PutBucketCors",
1421                "DELETE" => "DeleteBucketCors",
1422                _ => return None,
1423            });
1424        }
1425        if has("policy") {
1426            return Some(match method {
1427                "GET" => "GetBucketPolicy",
1428                "PUT" => "PutBucketPolicy",
1429                "DELETE" => "DeleteBucketPolicy",
1430                _ => return None,
1431            });
1432        }
1433        if has("website") {
1434            return Some(match method {
1435                "GET" => "GetBucketWebsite",
1436                "PUT" => "PutBucketWebsite",
1437                "DELETE" => "DeleteBucketWebsite",
1438                _ => return None,
1439            });
1440        }
1441        if has("lifecycle") {
1442            return Some(match method {
1443                "GET" => "GetBucketLifecycleConfiguration",
1444                "PUT" => "PutBucketLifecycleConfiguration",
1445                "DELETE" => "DeleteBucketLifecycle",
1446                _ => return None,
1447            });
1448        }
1449        if has("encryption") {
1450            return Some(match method {
1451                "GET" => "GetBucketEncryption",
1452                "PUT" => "PutBucketEncryption",
1453                "DELETE" => "DeleteBucketEncryption",
1454                _ => return None,
1455            });
1456        }
1457        if has("logging") {
1458            return Some(match method {
1459                "GET" => "GetBucketLogging",
1460                "PUT" => "PutBucketLogging",
1461                _ => return None,
1462            });
1463        }
1464        if has("notification") {
1465            return Some(match method {
1466                "GET" => "GetBucketNotificationConfiguration",
1467                "PUT" => "PutBucketNotificationConfiguration",
1468                _ => return None,
1469            });
1470        }
1471        if has("replication") {
1472            return Some(match method {
1473                "GET" => "GetBucketReplication",
1474                "PUT" => "PutBucketReplication",
1475                "DELETE" => "DeleteBucketReplication",
1476                _ => return None,
1477            });
1478        }
1479        if has("ownershipControls") {
1480            return Some(match method {
1481                "GET" => "GetBucketOwnershipControls",
1482                "PUT" => "PutBucketOwnershipControls",
1483                "DELETE" => "DeleteBucketOwnershipControls",
1484                _ => return None,
1485            });
1486        }
1487        if has("publicAccessBlock") {
1488            return Some(match method {
1489                "GET" => "GetPublicAccessBlock",
1490                "PUT" => "PutPublicAccessBlock",
1491                "DELETE" => "DeletePublicAccessBlock",
1492                _ => return None,
1493            });
1494        }
1495        if has("accelerate") {
1496            return Some(match method {
1497                "GET" => "GetBucketAccelerateConfiguration",
1498                "PUT" => "PutBucketAccelerateConfiguration",
1499                _ => return None,
1500            });
1501        }
1502        if has("inventory") {
1503            return Some(match method {
1504                "GET" => "GetBucketInventoryConfiguration",
1505                "PUT" => "PutBucketInventoryConfiguration",
1506                "DELETE" => "DeleteBucketInventoryConfiguration",
1507                _ => return None,
1508            });
1509        }
1510        if has("analytics") {
1511            return Some(match method {
1512                "GET" if has("id") => "GetBucketAnalyticsConfiguration",
1513                "GET" => "ListBucketAnalyticsConfigurations",
1514                "PUT" => "PutBucketAnalyticsConfiguration",
1515                "DELETE" => "DeleteBucketAnalyticsConfiguration",
1516                _ => return None,
1517            });
1518        }
1519        if has("intelligent-tiering") {
1520            return Some(match method {
1521                "GET" if has("id") => "GetBucketIntelligentTieringConfiguration",
1522                "GET" => "ListBucketIntelligentTieringConfigurations",
1523                "PUT" => "PutBucketIntelligentTieringConfiguration",
1524                "DELETE" => "DeleteBucketIntelligentTieringConfiguration",
1525                _ => return None,
1526            });
1527        }
1528        if has("metrics") {
1529            return Some(match method {
1530                "GET" if has("id") => "GetBucketMetricsConfiguration",
1531                "GET" => "ListBucketMetricsConfigurations",
1532                "PUT" => "PutBucketMetricsConfiguration",
1533                "DELETE" => "DeleteBucketMetricsConfiguration",
1534                _ => return None,
1535            });
1536        }
1537        if has("requestPayment") {
1538            return Some(match method {
1539                "GET" => "GetBucketRequestPayment",
1540                "PUT" => "PutBucketRequestPayment",
1541                _ => return None,
1542            });
1543        }
1544        if has("policyStatus") && is_get {
1545            return Some("GetBucketPolicyStatus");
1546        }
1547        if has("metadataConfiguration") {
1548            return Some(match method {
1549                "GET" => "GetBucketMetadataConfiguration",
1550                "POST" => "CreateBucketMetadataConfiguration",
1551                "DELETE" => "DeleteBucketMetadataConfiguration",
1552                _ => return None,
1553            });
1554        }
1555        if has("metadataTable") {
1556            return Some(match method {
1557                "GET" => "GetBucketMetadataTableConfiguration",
1558                "POST" => "CreateBucketMetadataTableConfiguration",
1559                "DELETE" => "DeleteBucketMetadataTableConfiguration",
1560                _ => return None,
1561            });
1562        }
1563        if has("metadataInventoryTable") && is_put {
1564            return Some("UpdateBucketMetadataInventoryTableConfiguration");
1565        }
1566        if has("metadataJournalTable") && is_put {
1567            return Some("UpdateBucketMetadataJournalTableConfiguration");
1568        }
1569        if has("abac") {
1570            // PUT  -> PutBucketAbacConfiguration
1571            // GET  -> GetBucketAbacConfiguration (previously fell through to
1572            //         the plain `GET bucket` arm and was IAM-evaluated as
1573            //         s3:ListObjects — a policy on the ABAC config was never
1574            //         honored).
1575            return Some(match method {
1576                "PUT" => "PutBucketAbacConfiguration",
1577                "GET" => "GetBucketAbacConfiguration",
1578                _ => return None,
1579            });
1580        }
1581        if has("session") && is_get {
1582            // CreateSession (S3 Express directory buckets). AWS authorizes it
1583            // under the separate `s3express:CreateSession` action — handled in
1584            // `iam_action_for`, which selects the `s3express` service prefix
1585            // for this op. Previously fell through to the plain `GET bucket`
1586            // arm and was evaluated as s3:ListObjects.
1587            return Some("CreateSession");
1588        }
1589        if has("object-lock") {
1590            return Some(match method {
1591                "GET" => "GetObjectLockConfiguration",
1592                "PUT" => "PutObjectLockConfiguration",
1593                _ => return None,
1594            });
1595        }
1596        if has("location") {
1597            return Some("GetBucketLocation");
1598        }
1599        if is_post && has("delete") {
1600            return Some("DeleteObjects");
1601        }
1602        if is_get && has("versions") {
1603            return Some("ListObjectVersions");
1604        }
1605    }
1606
1607    // GET on an object with `?torrent` is GetObjectTorrent, authorized under
1608    // s3:GetObjectTorrent -- not s3:GetObject. Without this arm a policy that
1609    // allows GetObject but not GetObjectTorrent would wrongly permit it
1610    // (bug-audit 2026-06-20, 5.3).
1611    if is_get && has_key && has("torrent") {
1612        return Some("GetObjectTorrent");
1613    }
1614
1615    // Plain bucket/object methods.
1616    match (method, has_key) {
1617        ("GET", true) => Some("GetObject"),
1618        ("PUT", true) => {
1619            // CopyObject uses x-amz-copy-source but we don't have headers
1620            // handy here — treat both PutObject and CopyObject as PutObject
1621            // for IAM purposes; CopyObject additionally requires
1622            // s3:GetObject on the source but that's evaluated per-request
1623            // by real AWS, not on the PUT call itself.
1624            Some("PutObject")
1625        }
1626        ("DELETE", true) => Some("DeleteObject"),
1627        ("HEAD", true) => Some("HeadObject"),
1628        ("GET", false) => {
1629            if query.contains_key("list-type") {
1630                Some("ListObjectsV2")
1631            } else {
1632                Some("ListObjects")
1633            }
1634        }
1635        ("PUT", false) => Some("CreateBucket"),
1636        ("DELETE", false) => Some("DeleteBucket"),
1637        ("HEAD", false) => Some("HeadBucket"),
1638        _ => None,
1639    }
1640}
1641
1642/// Build the S3 resource ARN for an action. Returns `*` for
1643/// `ListBuckets` (account-scoped), a bucket ARN for bucket-level
1644/// configuration actions, or an object ARN for object-level actions.
1645fn s3_resource_for(action: &'static str, bucket: Option<&str>, key: Option<&str>) -> String {
1646    // Object-level actions work on `bucket/key`.
1647    const OBJECT_ACTIONS: &[&str] = &[
1648        "PutObject",
1649        "GetObject",
1650        "DeleteObject",
1651        "HeadObject",
1652        "CopyObject",
1653        "GetObjectAttributes",
1654        "RestoreObject",
1655        "PutObjectTagging",
1656        "GetObjectTagging",
1657        "DeleteObjectTagging",
1658        "PutObjectAcl",
1659        "GetObjectAcl",
1660        "PutObjectRetention",
1661        "GetObjectRetention",
1662        "PutObjectLegalHold",
1663        "GetObjectLegalHold",
1664        "CreateMultipartUpload",
1665        "UploadPart",
1666        "UploadPartCopy",
1667        "CompleteMultipartUpload",
1668        "AbortMultipartUpload",
1669        "ListParts",
1670    ];
1671    if action == "ListBuckets" {
1672        return "*".to_string();
1673    }
1674    let Some(bucket) = bucket else {
1675        return "*".to_string();
1676    };
1677    if OBJECT_ACTIONS.contains(&action) {
1678        match key {
1679            Some(k) if !k.is_empty() => Arn::s3(&format!("{bucket}/{k}")).to_string(),
1680            _ => Arn::s3(&format!("{bucket}/*")).to_string(),
1681        }
1682    } else {
1683        // Bucket-level actions (ListObjectsV2, GetBucketTagging, ...).
1684        Arn::s3(bucket).to_string()
1685    }
1686}
1687
1688// Conditional request helpers
1689
1690/// Truncate a DateTime to second-level precision (HTTP dates have no sub-second info).
1691pub(crate) fn truncate_to_seconds(dt: DateTime<Utc>) -> DateTime<Utc> {
1692    dt.with_nanosecond(0).unwrap_or(dt)
1693}
1694
1695pub(crate) fn check_get_conditionals(
1696    req: &AwsRequest,
1697    obj: &S3Object,
1698) -> Result<(), AwsServiceError> {
1699    let obj_etag = format!("\"{}\"", obj.etag);
1700    let obj_time = truncate_to_seconds(obj.last_modified);
1701
1702    // If-Match
1703    if let Some(if_match) = req.headers.get("if-match").and_then(|v| v.to_str().ok()) {
1704        if !etag_matches(if_match, &obj_etag) {
1705            return Err(precondition_failed("If-Match"));
1706        }
1707    }
1708
1709    // If-None-Match
1710    if let Some(if_none_match) = req
1711        .headers
1712        .get("if-none-match")
1713        .and_then(|v| v.to_str().ok())
1714    {
1715        if etag_matches(if_none_match, &obj_etag) {
1716            return Err(not_modified_with_etag(&obj_etag));
1717        }
1718    }
1719
1720    // If-Unmodified-Since — RFC 7232 §3.4: ignored when If-Match is present.
1721    if req.headers.get("if-match").is_none() {
1722        if let Some(since) = req
1723            .headers
1724            .get("if-unmodified-since")
1725            .and_then(|v| v.to_str().ok())
1726        {
1727            if let Some(dt) = parse_http_date(since) {
1728                if obj_time > dt {
1729                    return Err(precondition_failed("If-Unmodified-Since"));
1730                }
1731            }
1732        }
1733    }
1734
1735    // If-Modified-Since — RFC 7232 §3.3: ignored when If-None-Match is present.
1736    if req.headers.get("if-none-match").is_none() {
1737        if let Some(since) = req
1738            .headers
1739            .get("if-modified-since")
1740            .and_then(|v| v.to_str().ok())
1741        {
1742            if let Some(dt) = parse_http_date(since) {
1743                if obj_time <= dt {
1744                    return Err(not_modified());
1745                }
1746            }
1747        }
1748    }
1749
1750    Ok(())
1751}
1752
1753pub(crate) fn check_head_conditionals(
1754    req: &AwsRequest,
1755    obj: &S3Object,
1756) -> Result<(), AwsServiceError> {
1757    let obj_etag = format!("\"{}\"", obj.etag);
1758    let obj_time = truncate_to_seconds(obj.last_modified);
1759
1760    // If-Match
1761    if let Some(if_match) = req.headers.get("if-match").and_then(|v| v.to_str().ok()) {
1762        if !etag_matches(if_match, &obj_etag) {
1763            return Err(AwsServiceError::aws_error(
1764                StatusCode::PRECONDITION_FAILED,
1765                "412",
1766                "Precondition Failed",
1767            ));
1768        }
1769    }
1770
1771    // If-None-Match
1772    if let Some(if_none_match) = req
1773        .headers
1774        .get("if-none-match")
1775        .and_then(|v| v.to_str().ok())
1776    {
1777        if etag_matches(if_none_match, &obj_etag) {
1778            return Err(not_modified_with_etag(&obj_etag));
1779        }
1780    }
1781
1782    // If-Unmodified-Since — RFC 7232 §3.4: ignored when If-Match is present.
1783    if req.headers.get("if-match").is_none() {
1784        if let Some(since) = req
1785            .headers
1786            .get("if-unmodified-since")
1787            .and_then(|v| v.to_str().ok())
1788        {
1789            if let Some(dt) = parse_http_date(since) {
1790                if obj_time > dt {
1791                    return Err(AwsServiceError::aws_error(
1792                        StatusCode::PRECONDITION_FAILED,
1793                        "412",
1794                        "Precondition Failed",
1795                    ));
1796                }
1797            }
1798        }
1799    }
1800
1801    // If-Modified-Since — RFC 7232 §3.3: ignored when If-None-Match is present.
1802    if req.headers.get("if-none-match").is_none() {
1803        if let Some(since) = req
1804            .headers
1805            .get("if-modified-since")
1806            .and_then(|v| v.to_str().ok())
1807        {
1808            if let Some(dt) = parse_http_date(since) {
1809                if obj_time <= dt {
1810                    return Err(not_modified());
1811                }
1812            }
1813        }
1814    }
1815
1816    Ok(())
1817}
1818
1819pub(crate) fn etag_matches(condition: &str, obj_etag: &str) -> bool {
1820    let condition = condition.trim();
1821    if condition == "*" {
1822        return true;
1823    }
1824    let clean_etag = obj_etag.replace('"', "");
1825    // Split on comma to handle multi-value If-Match / If-None-Match
1826    for part in condition.split(',') {
1827        let part = part.trim().replace('"', "");
1828        if part == clean_etag {
1829            return true;
1830        }
1831    }
1832    false
1833}
1834
1835pub(crate) fn parse_http_date(s: &str) -> Option<DateTime<Utc>> {
1836    // Try RFC 2822 format: "Sat, 01 Jan 2000 00:00:00 GMT"
1837    if let Ok(dt) = DateTime::parse_from_rfc2822(s) {
1838        return Some(dt.with_timezone(&Utc));
1839    }
1840    // Try RFC 3339
1841    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
1842        return Some(dt.with_timezone(&Utc));
1843    }
1844    // Try common HTTP date format: "%a, %d %b %Y %H:%M:%S GMT"
1845    if let Ok(dt) =
1846        chrono::NaiveDateTime::parse_from_str(s.trim_end_matches(" GMT"), "%a, %d %b %Y %H:%M:%S")
1847    {
1848        return Some(dt.and_utc());
1849    }
1850    // Try ISO 8601
1851    if let Ok(dt) = s.parse::<DateTime<Utc>>() {
1852        return Some(dt);
1853    }
1854    None
1855}
1856
1857pub(crate) fn not_modified() -> AwsServiceError {
1858    AwsServiceError::aws_error(StatusCode::NOT_MODIFIED, "304", "Not Modified")
1859}
1860
1861pub(crate) fn not_modified_with_etag(etag: &str) -> AwsServiceError {
1862    AwsServiceError::aws_error_with_headers(
1863        StatusCode::NOT_MODIFIED,
1864        "304",
1865        "Not Modified",
1866        vec![("etag".to_string(), etag.to_string())],
1867    )
1868}
1869
1870pub(crate) fn precondition_failed(condition: &str) -> AwsServiceError {
1871    AwsServiceError::aws_error_with_fields(
1872        StatusCode::PRECONDITION_FAILED,
1873        "PreconditionFailed",
1874        "At least one of the pre-conditions you specified did not hold",
1875        vec![("Condition".to_string(), condition.to_string())],
1876    )
1877}
1878
1879// ACL helpers
1880
1881pub(crate) fn build_acl_xml(owner_id: &str, grants: &[AclGrant], _account_id: &str) -> String {
1882    let mut grants_xml = String::new();
1883    for g in grants {
1884        let grantee_xml = if g.grantee_type == "Group" {
1885            let uri = g.grantee_uri.as_deref().unwrap_or("");
1886            format!(
1887                "<Grantee xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xsi:type=\"Group\">\
1888                 <URI>{}</URI></Grantee>",
1889                xml_escape(uri),
1890            )
1891        } else {
1892            let id = g.grantee_id.as_deref().unwrap_or("");
1893            format!(
1894                "<Grantee xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xsi:type=\"CanonicalUser\">\
1895                 <ID>{}</ID></Grantee>",
1896                xml_escape(id),
1897            )
1898        };
1899        grants_xml.push_str(&format!(
1900            "<Grant>{grantee_xml}<Permission>{}</Permission></Grant>",
1901            xml_escape(&g.permission),
1902        ));
1903    }
1904
1905    format!(
1906        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
1907         <AccessControlPolicy xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">\
1908         <Owner><ID>{owner_id}</ID><DisplayName>{owner_id}</DisplayName></Owner>\
1909         <AccessControlList>{grants_xml}</AccessControlList>\
1910         </AccessControlPolicy>",
1911        owner_id = xml_escape(owner_id),
1912    )
1913}
1914
1915pub(crate) fn canned_acl_grants(acl: &str, owner_id: &str) -> Vec<AclGrant> {
1916    let owner_grant = AclGrant {
1917        grantee_type: "CanonicalUser".to_string(),
1918        grantee_id: Some(owner_id.to_string()),
1919        grantee_display_name: Some(owner_id.to_string()),
1920        grantee_uri: None,
1921        permission: "FULL_CONTROL".to_string(),
1922    };
1923    match acl {
1924        "private" => vec![owner_grant],
1925        "public-read" => vec![
1926            owner_grant,
1927            AclGrant {
1928                grantee_type: "Group".to_string(),
1929                grantee_id: None,
1930                grantee_display_name: None,
1931                grantee_uri: Some("http://acs.amazonaws.com/groups/global/AllUsers".to_string()),
1932                permission: "READ".to_string(),
1933            },
1934        ],
1935        "public-read-write" => vec![
1936            owner_grant,
1937            AclGrant {
1938                grantee_type: "Group".to_string(),
1939                grantee_id: None,
1940                grantee_display_name: None,
1941                grantee_uri: Some("http://acs.amazonaws.com/groups/global/AllUsers".to_string()),
1942                permission: "READ".to_string(),
1943            },
1944            AclGrant {
1945                grantee_type: "Group".to_string(),
1946                grantee_id: None,
1947                grantee_display_name: None,
1948                grantee_uri: Some("http://acs.amazonaws.com/groups/global/AllUsers".to_string()),
1949                permission: "WRITE".to_string(),
1950            },
1951        ],
1952        "authenticated-read" => vec![
1953            owner_grant,
1954            AclGrant {
1955                grantee_type: "Group".to_string(),
1956                grantee_id: None,
1957                grantee_display_name: None,
1958                grantee_uri: Some(
1959                    "http://acs.amazonaws.com/groups/global/AuthenticatedUsers".to_string(),
1960                ),
1961                permission: "READ".to_string(),
1962            },
1963        ],
1964        "bucket-owner-full-control" => vec![owner_grant],
1965        _ => vec![owner_grant],
1966    }
1967}
1968
1969pub(crate) fn canned_acl_grants_for_object(acl: &str, owner_id: &str) -> Vec<AclGrant> {
1970    // For objects, canned ACLs work the same way
1971    canned_acl_grants(acl, owner_id)
1972}
1973
1974pub(crate) fn parse_grant_headers(headers: &HeaderMap) -> Vec<AclGrant> {
1975    let mut grants = Vec::new();
1976    let header_permission_map = [
1977        ("x-amz-grant-read", "READ"),
1978        ("x-amz-grant-write", "WRITE"),
1979        ("x-amz-grant-read-acp", "READ_ACP"),
1980        ("x-amz-grant-write-acp", "WRITE_ACP"),
1981        ("x-amz-grant-full-control", "FULL_CONTROL"),
1982    ];
1983
1984    for (header, permission) in &header_permission_map {
1985        if let Some(value) = headers.get(*header).and_then(|v| v.to_str().ok()) {
1986            // Parse "id=xxx" or "uri=xxx" or "emailAddress=xxx"
1987            for part in value.split(',') {
1988                let part = part.trim();
1989                if let Some((key, val)) = part.split_once('=') {
1990                    let val = val.trim().trim_matches('"');
1991                    let key = key.trim().to_lowercase();
1992                    match key.as_str() {
1993                        "id" => {
1994                            grants.push(AclGrant {
1995                                grantee_type: "CanonicalUser".to_string(),
1996                                grantee_id: Some(val.to_string()),
1997                                grantee_display_name: Some(val.to_string()),
1998                                grantee_uri: None,
1999                                permission: permission.to_string(),
2000                            });
2001                        }
2002                        "uri" | "url" => {
2003                            grants.push(AclGrant {
2004                                grantee_type: "Group".to_string(),
2005                                grantee_id: None,
2006                                grantee_display_name: None,
2007                                grantee_uri: Some(val.to_string()),
2008                                permission: permission.to_string(),
2009                            });
2010                        }
2011                        _ => {}
2012                    }
2013                }
2014            }
2015        }
2016    }
2017    grants
2018}
2019
2020pub(crate) fn parse_acl_xml(xml: &str) -> Result<Vec<AclGrant>, AwsServiceError> {
2021    // Check for Owner presence
2022    if xml.contains("<AccessControlPolicy") && !xml.contains("<Owner>") {
2023        return Err(AwsServiceError::aws_error(
2024            StatusCode::BAD_REQUEST,
2025            "MalformedACLError",
2026            "The XML you provided was not well-formed or did not validate against our published schema",
2027        ));
2028    }
2029
2030    let valid_permissions = ["READ", "WRITE", "READ_ACP", "WRITE_ACP", "FULL_CONTROL"];
2031
2032    let mut grants = Vec::new();
2033    let mut remaining = xml;
2034    while let Some(start) = remaining.find("<Grant>") {
2035        let after = &remaining[start + 7..];
2036        if let Some(end) = after.find("</Grant>") {
2037            let grant_body = &after[..end];
2038
2039            // Extract permission
2040            let permission = extract_xml_value(grant_body, "Permission").unwrap_or_default();
2041            if !valid_permissions.contains(&permission.as_str()) {
2042                return Err(AwsServiceError::aws_error(
2043                    StatusCode::BAD_REQUEST,
2044                    "MalformedACLError",
2045                    "The XML you provided was not well-formed or did not validate against our published schema",
2046                ));
2047            }
2048
2049            // Determine grantee type
2050            if grant_body.contains("xsi:type=\"Group\"") || grant_body.contains("<URI>") {
2051                let uri = extract_xml_value(grant_body, "URI").unwrap_or_default();
2052                grants.push(AclGrant {
2053                    grantee_type: "Group".to_string(),
2054                    grantee_id: None,
2055                    grantee_display_name: None,
2056                    grantee_uri: Some(uri),
2057                    permission,
2058                });
2059            } else {
2060                let id = extract_xml_value(grant_body, "ID").unwrap_or_default();
2061                let display =
2062                    extract_xml_value(grant_body, "DisplayName").unwrap_or_else(|| id.clone());
2063                grants.push(AclGrant {
2064                    grantee_type: "CanonicalUser".to_string(),
2065                    grantee_id: Some(id),
2066                    grantee_display_name: Some(display),
2067                    grantee_uri: None,
2068                    permission,
2069                });
2070            }
2071
2072            remaining = &after[end + 8..];
2073        } else {
2074            break;
2075        }
2076    }
2077    Ok(grants)
2078}
2079
2080// Range helpers
2081
2082pub(crate) enum RangeResult {
2083    Satisfiable { start: usize, end: usize },
2084    NotSatisfiable,
2085    Ignored,
2086}
2087
2088pub(crate) fn parse_range_header(range_str: &str, total_size: usize) -> Option<RangeResult> {
2089    let range_str = range_str.strip_prefix("bytes=")?;
2090    let (start_str, end_str) = range_str.split_once('-')?;
2091    if start_str.is_empty() {
2092        let suffix_len: usize = end_str.parse().ok()?;
2093        if suffix_len == 0 || total_size == 0 {
2094            return Some(RangeResult::NotSatisfiable);
2095        }
2096        let start = total_size.saturating_sub(suffix_len);
2097        Some(RangeResult::Satisfiable {
2098            start,
2099            end: total_size - 1,
2100        })
2101    } else {
2102        let start: usize = start_str.parse().ok()?;
2103        if start >= total_size {
2104            return Some(RangeResult::NotSatisfiable);
2105        }
2106        let end = if end_str.is_empty() {
2107            total_size - 1
2108        } else {
2109            let e: usize = end_str.parse().ok()?;
2110            if e < start {
2111                return Some(RangeResult::Ignored);
2112            }
2113            std::cmp::min(e, total_size - 1)
2114        };
2115        Some(RangeResult::Satisfiable { start, end })
2116    }
2117}
2118
2119// Helpers
2120
2121/// S3 XML response with `application/xml` content type (unlike Query protocol's `text/xml`).
2122pub(crate) fn s3_xml(status: StatusCode, body: impl Into<Bytes>) -> AwsResponse {
2123    AwsResponse {
2124        status,
2125        content_type: "application/xml".to_string(),
2126        body: body.into().into(),
2127        headers: HeaderMap::new(),
2128    }
2129}
2130
2131pub(crate) fn empty_response(status: StatusCode) -> AwsResponse {
2132    AwsResponse {
2133        status,
2134        content_type: "application/xml".to_string(),
2135        body: Bytes::new().into(),
2136        headers: HeaderMap::new(),
2137    }
2138}
2139
2140/// Returns true when the object is stored in a "cold" storage class (GLACIER, DEEP_ARCHIVE)
2141/// and has NOT been restored (or restore is still in progress).
2142pub(crate) fn is_frozen(obj: &S3Object) -> bool {
2143    matches!(obj.storage_class.as_str(), "GLACIER" | "DEEP_ARCHIVE")
2144        && obj.restore_ongoing != Some(false)
2145}
2146
2147pub(crate) fn no_such_bucket(bucket: &str) -> AwsServiceError {
2148    AwsServiceError::aws_error_with_fields(
2149        StatusCode::NOT_FOUND,
2150        "NoSuchBucket",
2151        "The specified bucket does not exist",
2152        vec![("BucketName".to_string(), bucket.to_string())],
2153    )
2154}
2155
2156pub(crate) fn no_such_key(key: &str) -> AwsServiceError {
2157    AwsServiceError::aws_error_with_fields(
2158        StatusCode::NOT_FOUND,
2159        "NoSuchKey",
2160        "The specified key does not exist.",
2161        vec![("Key".to_string(), key.to_string())],
2162    )
2163}
2164
2165pub(crate) fn no_such_upload(upload_id: &str) -> AwsServiceError {
2166    AwsServiceError::aws_error_with_fields(
2167        StatusCode::NOT_FOUND,
2168        "NoSuchUpload",
2169        "The specified upload does not exist. The upload ID may be invalid, \
2170         or the upload may have been aborted or completed.",
2171        vec![("UploadId".to_string(), upload_id.to_string())],
2172    )
2173}
2174
2175pub(crate) fn no_such_key_with_detail(key: &str) -> AwsServiceError {
2176    AwsServiceError::aws_error_with_fields(
2177        StatusCode::NOT_FOUND,
2178        "NoSuchKey",
2179        "The specified key does not exist.",
2180        vec![("Key".to_string(), key.to_string())],
2181    )
2182}
2183
2184pub(crate) fn compute_md5(data: &[u8]) -> String {
2185    let digest = Md5::digest(data);
2186    format!("{:x}", digest)
2187}
2188
2189pub(crate) fn compute_checksum(algorithm: &str, data: &[u8]) -> String {
2190    let raw = compute_checksum_raw(algorithm, data);
2191    if raw.is_empty() {
2192        String::new()
2193    } else {
2194        BASE64.encode(raw)
2195    }
2196}
2197
2198/// Compute the raw (un-base64'd) checksum digest bytes for `data`.
2199///
2200/// Returns the network-order digest bytes so callers can either base64
2201/// them ([`compute_checksum`]) or concatenate them when computing an S3
2202/// multipart COMPOSITE checksum (which hashes the concatenation of each
2203/// part's raw digest, then base64s the result and appends `-N`).
2204pub(crate) fn compute_checksum_raw(algorithm: &str, data: &[u8]) -> Vec<u8> {
2205    match algorithm {
2206        "CRC32" => crc32fast::hash(data).to_be_bytes().to_vec(),
2207        // CRC32C and CRC64NVME use the same crates as the streaming path so
2208        // CopyObject / CompleteMultipartUpload produce identical results to
2209        // a streaming PutObject (1.7); the COMPOSITE multipart checksum
2210        // concatenates these raw digests before hashing (1.18).
2211        "CRC32C" => crc32c::crc32c(data).to_be_bytes().to_vec(),
2212        "CRC64NVME" => {
2213            let mut hasher = crc64fast_nvme::Digest::new();
2214            hasher.write(data);
2215            hasher.sum64().to_be_bytes().to_vec()
2216        }
2217        "SHA1" => {
2218            use sha1::Digest as _;
2219            sha1::Sha1::digest(data).to_vec()
2220        }
2221        "SHA256" => {
2222            use sha2::Digest as _;
2223            sha2::Sha256::digest(data).to_vec()
2224        }
2225        _ => Vec::new(),
2226    }
2227}
2228
2229/// Compute an S3 multipart COMPOSITE additional checksum from each
2230/// part's raw digest bytes (in part-number order). AWS defines this as
2231/// `base64(HASH(concat(raw_part_digests)))-N` where HASH is the chosen
2232/// algorithm and N is the part count. Returns `None` for an unsupported
2233/// algorithm or empty part list.
2234pub(crate) fn compute_composite_checksum(
2235    algorithm: &str,
2236    part_raw_digests: &[Vec<u8>],
2237) -> Option<String> {
2238    if part_raw_digests.is_empty() {
2239        return None;
2240    }
2241    let mut concat = Vec::new();
2242    for d in part_raw_digests {
2243        concat.extend_from_slice(d);
2244    }
2245    let digest = compute_checksum_raw(algorithm, &concat);
2246    if digest.is_empty() {
2247        return None;
2248    }
2249    Some(format!(
2250        "{}-{}",
2251        BASE64.encode(digest),
2252        part_raw_digests.len()
2253    ))
2254}
2255
2256/// Streaming variant of [`compute_checksum`] for spool files. Reads
2257/// the file in 1 MiB chunks and feeds each chunk into the hasher so a
2258/// 1 GiB upload computes its CRC32 / SHA-1 / SHA-256 in constant
2259/// memory rather than via `tokio::fs::read` (which would allocate the
2260/// whole file as one buffer).
2261pub(crate) async fn compute_checksum_streaming(
2262    algorithm: &str,
2263    path: &std::path::Path,
2264) -> Result<String, std::io::Error> {
2265    use tokio::io::AsyncReadExt;
2266    let mut file = tokio::fs::File::open(path).await?;
2267    let mut buf = vec![0u8; 1024 * 1024];
2268    match algorithm {
2269        "CRC32" => {
2270            let mut hasher = crc32fast::Hasher::new();
2271            loop {
2272                let n = file.read(&mut buf).await?;
2273                if n == 0 {
2274                    break;
2275                }
2276                hasher.update(&buf[..n]);
2277            }
2278            Ok(BASE64.encode(hasher.finalize().to_be_bytes()))
2279        }
2280        "CRC32C" => {
2281            let mut crc: u32 = 0;
2282            loop {
2283                let n = file.read(&mut buf).await?;
2284                if n == 0 {
2285                    break;
2286                }
2287                crc = crc32c::crc32c_append(crc, &buf[..n]);
2288            }
2289            Ok(BASE64.encode(crc.to_be_bytes()))
2290        }
2291        "CRC64NVME" => {
2292            let mut hasher = crc64fast_nvme::Digest::new();
2293            loop {
2294                let n = file.read(&mut buf).await?;
2295                if n == 0 {
2296                    break;
2297                }
2298                hasher.write(&buf[..n]);
2299            }
2300            Ok(BASE64.encode(hasher.sum64().to_be_bytes()))
2301        }
2302        "SHA1" => {
2303            use sha1::Digest as _;
2304            let mut hasher = sha1::Sha1::new();
2305            loop {
2306                let n = file.read(&mut buf).await?;
2307                if n == 0 {
2308                    break;
2309                }
2310                hasher.update(&buf[..n]);
2311            }
2312            Ok(BASE64.encode(hasher.finalize()))
2313        }
2314        "SHA256" => {
2315            use sha2::Digest as _;
2316            let mut hasher = sha2::Sha256::new();
2317            loop {
2318                let n = file.read(&mut buf).await?;
2319                if n == 0 {
2320                    break;
2321                }
2322                hasher.update(&buf[..n]);
2323            }
2324            Ok(BASE64.encode(hasher.finalize()))
2325        }
2326        _ => Ok(String::new()),
2327    }
2328}
2329
2330pub(crate) fn url_encode_s3_key(s: &str) -> String {
2331    let mut out = String::with_capacity(s.len() * 2);
2332    for byte in s.bytes() {
2333        match byte {
2334            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' | b'/' => {
2335                out.push(byte as char);
2336            }
2337            _ => {
2338                out.push_str(&format!("%{:02X}", byte));
2339            }
2340        }
2341    }
2342    out
2343}
2344
2345pub(crate) use fakecloud_aws::xml::xml_escape;
2346
2347pub(crate) fn extract_user_metadata(
2348    headers: &HeaderMap,
2349) -> std::collections::BTreeMap<String, String> {
2350    let mut meta = std::collections::BTreeMap::new();
2351    for (name, value) in headers {
2352        if let Some(key) = name.as_str().strip_prefix("x-amz-meta-") {
2353            if let Ok(v) = value.to_str() {
2354                meta.insert(key.to_string(), v.to_string());
2355            }
2356        }
2357    }
2358    meta
2359}
2360
2361pub(crate) fn is_valid_storage_class(class: &str) -> bool {
2362    matches!(
2363        class,
2364        "STANDARD"
2365            | "REDUCED_REDUNDANCY"
2366            | "STANDARD_IA"
2367            | "ONEZONE_IA"
2368            | "INTELLIGENT_TIERING"
2369            | "GLACIER"
2370            | "DEEP_ARCHIVE"
2371            | "GLACIER_IR"
2372            | "OUTPOSTS"
2373            | "SNOW"
2374            | "EXPRESS_ONEZONE"
2375    )
2376}
2377
2378pub(crate) fn is_valid_bucket_name(name: &str) -> bool {
2379    // General-purpose bucket naming rules (AWS S3):
2380    // https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
2381    if name.len() < 3 || name.len() > 63 {
2382        return false;
2383    }
2384    // Only lowercase letters, digits, hyphens, and dots. Underscores are NOT
2385    // permitted for general-purpose buckets.
2386    if !name
2387        .bytes()
2388        .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'.')
2389    {
2390        return false;
2391    }
2392    // Must start and end with a letter or number.
2393    let bytes = name.as_bytes();
2394    if !bytes[0].is_ascii_alphanumeric() || !bytes[bytes.len() - 1].is_ascii_alphanumeric() {
2395        return false;
2396    }
2397    // Must not contain two adjacent periods.
2398    if name.contains("..") {
2399        return false;
2400    }
2401    // Must not be formatted as an IPv4 address (e.g. 192.168.5.4).
2402    if is_ipv4_format(name) {
2403        return false;
2404    }
2405    // Reserved prefixes / suffixes.
2406    if name.starts_with("xn--")
2407        || name.starts_with("sthree-")
2408        || name.starts_with("amzn-s3-demo-")
2409        || name.ends_with("-s3alias")
2410        || name.ends_with("--ol-s3")
2411        || name.ends_with(".mrap")
2412    {
2413        return false;
2414    }
2415    true
2416}
2417
2418/// True when `name` is four dot-separated decimal octets (0-255), i.e. looks
2419/// like an IPv4 address, which S3 forbids as a bucket name.
2420fn is_ipv4_format(name: &str) -> bool {
2421    let parts: Vec<&str> = name.split('.').collect();
2422    parts.len() == 4
2423        && parts.iter().all(|p| {
2424            !p.is_empty()
2425                && p.len() <= 3
2426                && p.bytes().all(|b| b.is_ascii_digit())
2427                && p.parse::<u8>().is_ok()
2428        })
2429}
2430
2431pub(crate) fn is_valid_region(region: &str) -> bool {
2432    // Basic validation: region should match pattern like us-east-1, eu-west-2, etc.
2433    let valid_regions = [
2434        "us-east-1",
2435        "us-east-2",
2436        "us-west-1",
2437        "us-west-2",
2438        "af-south-1",
2439        "ap-east-1",
2440        "ap-south-1",
2441        "ap-south-2",
2442        "ap-southeast-1",
2443        "ap-southeast-2",
2444        "ap-southeast-3",
2445        "ap-southeast-4",
2446        "ap-northeast-1",
2447        "ap-northeast-2",
2448        "ap-northeast-3",
2449        "ca-central-1",
2450        "ca-west-1",
2451        "eu-central-1",
2452        "eu-central-2",
2453        "eu-west-1",
2454        "eu-west-2",
2455        "eu-west-3",
2456        "eu-south-1",
2457        "eu-south-2",
2458        "eu-north-1",
2459        "il-central-1",
2460        "me-south-1",
2461        "me-central-1",
2462        "sa-east-1",
2463        "cn-north-1",
2464        "cn-northwest-1",
2465        "us-gov-east-1",
2466        "us-gov-east-2",
2467        "us-gov-west-1",
2468        "us-iso-east-1",
2469        "us-iso-west-1",
2470        "us-isob-east-1",
2471        "us-isof-south-1",
2472    ];
2473    valid_regions.contains(&region)
2474}
2475
2476pub(crate) fn resolve_object<'a>(
2477    b: &'a S3Bucket,
2478    key: &str,
2479    version_id: Option<&String>,
2480) -> Result<&'a S3Object, AwsServiceError> {
2481    if let Some(vid) = version_id {
2482        // "null" version ID refers to an object with no version_id (pre-versioning)
2483        if vid == "null" {
2484            // Check versions for a pre-versioning object (version_id == None or Some("null"))
2485            if let Some(versions) = b.object_versions.get(key) {
2486                if let Some(obj) = versions
2487                    .iter()
2488                    .find(|o| o.version_id.is_none() || o.version_id.as_deref() == Some("null"))
2489                {
2490                    return Ok(obj);
2491                }
2492            }
2493            // Also check current object if it has no version_id
2494            if let Some(obj) = b.objects.get(key) {
2495                if obj.version_id.is_none() || obj.version_id.as_deref() == Some("null") {
2496                    return Ok(obj);
2497                }
2498            }
2499        } else {
2500            // When a specific versionId is requested, check versions first
2501            if let Some(versions) = b.object_versions.get(key) {
2502                if let Some(obj) = versions
2503                    .iter()
2504                    .find(|o| o.version_id.as_deref() == Some(vid.as_str()))
2505                {
2506                    return Ok(obj);
2507                }
2508            }
2509            // Also check current object
2510            if let Some(obj) = b.objects.get(key) {
2511                if obj.version_id.as_deref() == Some(vid.as_str()) {
2512                    return Ok(obj);
2513                }
2514            }
2515        }
2516        // For versioned buckets, return NoSuchVersion; for non-versioned, return 400
2517        if b.versioning.is_some() {
2518            Err(AwsServiceError::aws_error_with_fields(
2519                StatusCode::NOT_FOUND,
2520                "NoSuchVersion",
2521                "The specified version does not exist.",
2522                vec![
2523                    ("Key".to_string(), key.to_string()),
2524                    ("VersionId".to_string(), vid.to_string()),
2525                ],
2526            ))
2527        } else {
2528            Err(AwsServiceError::aws_error(
2529                StatusCode::BAD_REQUEST,
2530                "InvalidArgument",
2531                "Invalid version id specified",
2532            ))
2533        }
2534    } else {
2535        b.objects.get(key).ok_or_else(|| no_such_key(key))
2536    }
2537}
2538
2539pub(crate) fn make_delete_marker(key: &str, dm_id: &str) -> S3Object {
2540    S3Object {
2541        key: key.to_string(),
2542        last_modified: Utc::now(),
2543        storage_class: "STANDARD".to_string(),
2544        version_id: Some(dm_id.to_string()),
2545        is_delete_marker: true,
2546        ..Default::default()
2547    }
2548}
2549
2550/// Represents an object to delete in a batch delete request.
2551pub(crate) struct DeleteObjectEntry {
2552    key: String,
2553    version_id: Option<String>,
2554}
2555
2556pub(crate) fn parse_delete_objects_xml(xml: &str) -> Vec<DeleteObjectEntry> {
2557    let mut entries = Vec::new();
2558    let mut remaining = xml;
2559    while let Some(obj_start) = remaining.find("<Object>") {
2560        let after = &remaining[obj_start + 8..];
2561        if let Some(obj_end) = after.find("</Object>") {
2562            let obj_body = &after[..obj_end];
2563            let key = extract_xml_value(obj_body, "Key");
2564            let version_id = extract_xml_value(obj_body, "VersionId");
2565            if let Some(k) = key {
2566                entries.push(DeleteObjectEntry { key: k, version_id });
2567            }
2568            remaining = &after[obj_end + 9..];
2569        } else {
2570            break;
2571        }
2572    }
2573    entries
2574}
2575
2576/// Returns true when the request body's top-level <Quiet> element is "true".
2577/// AWS docs: when Quiet=true, only failed deletes are emitted.
2578pub(crate) fn parse_delete_objects_quiet(xml: &str) -> bool {
2579    extract_xml_value(xml, "Quiet")
2580        .map(|v| v.eq_ignore_ascii_case("true"))
2581        .unwrap_or(false)
2582}
2583
2584/// Minimal XML parser for `<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag>...`.
2585/// Returns a Vec to preserve insertion order and detect duplicates.
2586pub(crate) fn parse_tagging_xml(xml: &str) -> Vec<(String, String)> {
2587    let mut tags = Vec::new();
2588    let mut remaining = xml;
2589    while let Some(tag_start) = remaining.find("<Tag>") {
2590        let after = &remaining[tag_start + 5..];
2591        if let Some(tag_end) = after.find("</Tag>") {
2592            let tag_body = &after[..tag_end];
2593            let key = extract_xml_value(tag_body, "Key");
2594            let value = extract_xml_value(tag_body, "Value");
2595            if let (Some(k), Some(v)) = (key, value) {
2596                tags.push((k, v));
2597            }
2598            remaining = &after[tag_end + 6..];
2599        } else {
2600            break;
2601        }
2602    }
2603    tags
2604}
2605
2606pub(crate) fn validate_tags(tags: &[(String, String)]) -> Result<(), AwsServiceError> {
2607    // Check for duplicate keys
2608    let mut seen = std::collections::HashSet::new();
2609    for (k, _) in tags {
2610        if !seen.insert(k.as_str()) {
2611            return Err(AwsServiceError::aws_error(
2612                StatusCode::BAD_REQUEST,
2613                "InvalidTag",
2614                "Cannot provide multiple Tags with the same key",
2615            ));
2616        }
2617        // Check for aws: prefix
2618        if k.starts_with("aws:") {
2619            return Err(AwsServiceError::aws_error(
2620                StatusCode::BAD_REQUEST,
2621                "InvalidTag",
2622                "System tags cannot be added/updated by requester",
2623            ));
2624        }
2625    }
2626    Ok(())
2627}
2628
2629pub(crate) fn extract_xml_value(xml: &str, tag: &str) -> Option<String> {
2630    // Handle self-closing tags like <Value /> or <Value/>
2631    let self_closing1 = format!("<{tag} />");
2632    let self_closing2 = format!("<{tag}/>");
2633    if xml.contains(&self_closing1) || xml.contains(&self_closing2) {
2634        // Check if the self-closing tag appears before any open+close pair
2635        let self_pos = xml
2636            .find(&self_closing1)
2637            .or_else(|| xml.find(&self_closing2));
2638        let open = format!("<{tag}>");
2639        let open_pos = xml.find(&open);
2640        match (self_pos, open_pos) {
2641            (Some(sp), Some(op)) if sp < op => return Some(String::new()),
2642            (Some(_), None) => return Some(String::new()),
2643            _ => {}
2644        }
2645    }
2646
2647    let open = format!("<{tag}>");
2648    let close = format!("</{tag}>");
2649    let start = xml.find(&open)? + open.len();
2650    // Search for the closing tag AFTER the opening one so a malformed body
2651    // (closing tag before opening) can't make end < start and panic the
2652    // slice (bug-audit 2026-05-28, 2.2).
2653    let end = xml[start..].find(&close)? + start;
2654    // Decode XML entities in the text node. S3 request bodies arrive
2655    // XML-encoded — every SDK escapes `&`, `<`, `>` (and sometimes `"`/`'`
2656    // plus numeric refs) — so a key/value like `a&b` is on the wire as
2657    // `a&amp;b`. Without decoding, an object keyed `a&b` (the key arrives
2658    // URL-decoded from the request path) is never matched by a
2659    // DeleteObjects/PutObjectTagging body, and a tag value round-trips with
2660    // an extra layer of escaping on every GET.
2661    Some(decode_xml_entities(&xml[start..end]))
2662}
2663
2664/// Decode the standard XML entities in a text node value using a single
2665/// left-to-right pass. A chain of `str::replace` calls would double-decode
2666/// (e.g. `&amp;lt;` — the escaped literal text `&lt;` — must decode to
2667/// `&lt;`, not to `<`), so the scan resolves each `&…;` exactly once.
2668/// Handles the named entities (`amp`, `lt`, `gt`, `quot`, `apos`) and both
2669/// decimal (`&#38;`) and hex (`&#x26;`) numeric character references; an
2670/// unrecognized `&…` sequence is emitted verbatim, matching how lenient XML
2671/// readers treat a bare ampersand.
2672pub(crate) fn decode_xml_entities(s: &str) -> String {
2673    if !s.contains('&') {
2674        return s.to_string();
2675    }
2676    let mut out = String::with_capacity(s.len());
2677    let bytes = s.as_bytes();
2678    let mut i = 0;
2679    while i < s.len() {
2680        if bytes[i] == b'&' {
2681            // Bound the entity name so a stray '&' followed by a distant ';'
2682            // doesn't swallow a long span of text.
2683            if let Some(rel) = s[i + 1..].find(';') {
2684                if rel <= 10 {
2685                    let entity = &s[i + 1..i + 1 + rel];
2686                    let decoded = match entity {
2687                        "amp" => Some('&'),
2688                        "lt" => Some('<'),
2689                        "gt" => Some('>'),
2690                        "quot" => Some('"'),
2691                        "apos" => Some('\''),
2692                        _ => entity
2693                            .strip_prefix("#x")
2694                            .or_else(|| entity.strip_prefix("#X"))
2695                            .and_then(|hex| u32::from_str_radix(hex, 16).ok())
2696                            .or_else(|| {
2697                                entity.strip_prefix('#').and_then(|d| d.parse::<u32>().ok())
2698                            })
2699                            .and_then(char::from_u32),
2700                    };
2701                    if let Some(c) = decoded {
2702                        out.push(c);
2703                        i += 1 + rel + 1;
2704                        continue;
2705                    }
2706                }
2707            }
2708            out.push('&');
2709            i += 1;
2710        } else {
2711            let ch = s[i..].chars().next().unwrap();
2712            out.push(ch);
2713            i += ch.len_utf8();
2714        }
2715    }
2716    out
2717}
2718
2719/// Parse the CompleteMultipartUpload XML body into (part_number, etag) pairs.
2720/// Strip the surrounding quotes AWS clients wrap a part ETag in, so the value
2721/// matches the bare hex ETag fakecloud stores. The quote character arrives in
2722/// several encodings depending on the client's XML serializer: a literal `"`
2723/// (aws-cli/boto), the named entity `&quot;`, or the numeric character
2724/// references `&#34;` / `&#x22;` (aws-sdk-go-v2 — the terraform provider). All
2725/// must be removed, else a valid part is rejected with `InvalidPart` on
2726/// CompleteMultipartUpload. (2026-07-07: go-SDK multipart uploads of large
2727/// objects, e.g. the KinesisAnalyticsV2 tfacc app JAR, failed on this.)
2728pub(crate) fn strip_etag_quotes(s: &str) -> String {
2729    s.replace("&quot;", "")
2730        .replace("&#34;", "")
2731        .replace("&#x22;", "")
2732        .replace("&#X22;", "")
2733        .replace('"', "")
2734        .trim()
2735        .to_string()
2736}
2737
2738pub(crate) fn parse_complete_multipart_xml(xml: &str) -> Vec<(u32, String)> {
2739    let mut parts = Vec::new();
2740    let mut remaining = xml;
2741    while let Some(part_start) = remaining.find("<Part>") {
2742        let after = &remaining[part_start + 6..];
2743        if let Some(part_end) = after.find("</Part>") {
2744            let part_body = &after[..part_end];
2745            let part_num =
2746                extract_xml_value(part_body, "PartNumber").and_then(|s| s.parse::<u32>().ok());
2747            let etag = extract_xml_value(part_body, "ETag").map(|s| strip_etag_quotes(&s));
2748            if let (Some(num), Some(e)) = (part_num, etag) {
2749                parts.push((num, e));
2750            }
2751            remaining = &after[part_end + 7..];
2752        } else {
2753            break;
2754        }
2755    }
2756    parts
2757}
2758
2759pub(crate) fn parse_url_encoded_tags(s: &str) -> Vec<(String, String)> {
2760    let mut tags = Vec::new();
2761    for pair in s.split('&') {
2762        if pair.is_empty() {
2763            continue;
2764        }
2765        let (key, value) = match pair.find('=') {
2766            Some(pos) => (&pair[..pos], &pair[pos + 1..]),
2767            None => (pair, ""),
2768        };
2769        tags.push((
2770            percent_encoding::percent_decode_str(key)
2771                .decode_utf8_lossy()
2772                .to_string(),
2773            percent_encoding::percent_decode_str(value)
2774                .decode_utf8_lossy()
2775                .to_string(),
2776        ));
2777    }
2778    tags
2779}
2780
2781/// Validate lifecycle configuration XML. Returns MalformedXML on invalid configs.
2782pub(crate) fn validate_lifecycle_xml(xml: &str) -> Result<(), AwsServiceError> {
2783    let malformed = || {
2784        AwsServiceError::aws_error(
2785            StatusCode::BAD_REQUEST,
2786            "MalformedXML",
2787            "The XML you provided was not well-formed or did not validate against our published schema",
2788        )
2789    };
2790
2791    let mut remaining = xml;
2792    while let Some(rule_start) = remaining.find("<Rule>") {
2793        let after = &remaining[rule_start + 6..];
2794        if let Some(rule_end) = after.find("</Rule>") {
2795            let rule_body = &after[..rule_end];
2796
2797            // Must have Filter or Prefix
2798            let has_filter = rule_body.contains("<Filter>")
2799                || rule_body.contains("<Filter/>")
2800                || rule_body.contains("<Filter />");
2801
2802            // Check for <Prefix> at rule level (outside of <Filter>...</Filter>)
2803            let has_prefix_outside_filter = {
2804                if !rule_body.contains("<Prefix") {
2805                    false
2806                } else if !has_filter {
2807                    true // No filter means any Prefix is at rule level
2808                } else {
2809                    // Remove the Filter block and check if Prefix remains
2810                    let mut stripped = rule_body.to_string();
2811                    // Remove <Filter>...</Filter> or self-closing variants
2812                    if let Some(fs) = stripped.find("<Filter") {
2813                        if let Some(fe) = stripped.find("</Filter>") {
2814                            stripped = format!("{}{}", &stripped[..fs], &stripped[fe + 9..]);
2815                        }
2816                    }
2817                    stripped.contains("<Prefix")
2818                }
2819            };
2820
2821            if !has_filter && !has_prefix_outside_filter {
2822                return Err(malformed());
2823            }
2824            // Can't have both Filter and rule-level Prefix
2825            if has_filter && has_prefix_outside_filter {
2826                return Err(malformed());
2827            }
2828
2829            // Expiration: if has ExpiredObjectDeleteMarker, cannot also have Days or Date
2830            // (only check within <Expiration> block)
2831            if let Some(exp_start) = rule_body.find("<Expiration>") {
2832                if let Some(exp_end) = rule_body[exp_start..].find("</Expiration>") {
2833                    let exp_body = &rule_body[exp_start..exp_start + exp_end];
2834                    if exp_body.contains("<ExpiredObjectDeleteMarker>")
2835                        && (exp_body.contains("<Days>") || exp_body.contains("<Date>"))
2836                    {
2837                        return Err(malformed());
2838                    }
2839                }
2840            }
2841
2842            // Filter validation
2843            if has_filter {
2844                if let Some(fs) = rule_body.find("<Filter>") {
2845                    if let Some(fe) = rule_body.find("</Filter>") {
2846                        // `find` locates the two tags independently, so a body
2847                        // with `</Filter>` before `<Filter>` would slice with
2848                        // begin > end and panic (dropping the connection -- a
2849                        // reachable DoS). Reject as malformed instead.
2850                        if fe < fs + 8 {
2851                            return Err(malformed());
2852                        }
2853                        let filter_body = &rule_body[fs + 8..fe];
2854                        let has_prefix_in_filter = filter_body.contains("<Prefix");
2855                        let has_tag_in_filter = filter_body.contains("<Tag>");
2856                        let has_and_in_filter = filter_body.contains("<And>");
2857                        // Can't have both Prefix and Tag without And
2858                        if has_prefix_in_filter && has_tag_in_filter && !has_and_in_filter {
2859                            return Err(malformed());
2860                        }
2861                        // Can't have Tag and And simultaneously at the Filter level
2862                        if has_tag_in_filter && has_and_in_filter {
2863                            // Check if the <Tag> is outside <And>
2864                            let and_start = filter_body.find("<And>").unwrap_or(0);
2865                            let tag_pos = filter_body.find("<Tag>").unwrap_or(0);
2866                            if tag_pos < and_start {
2867                                return Err(malformed());
2868                            }
2869                        }
2870                    }
2871                }
2872            }
2873
2874            // NoncurrentVersionTransition must have NoncurrentDays and StorageClass
2875            if rule_body.contains("<NoncurrentVersionTransition>") {
2876                let mut nvt_remaining = rule_body;
2877                while let Some(nvt_start) = nvt_remaining.find("<NoncurrentVersionTransition>") {
2878                    let nvt_after = &nvt_remaining[nvt_start + 29..];
2879                    if let Some(nvt_end) = nvt_after.find("</NoncurrentVersionTransition>") {
2880                        let nvt_body = &nvt_after[..nvt_end];
2881                        if !nvt_body.contains("<NoncurrentDays>") {
2882                            return Err(malformed());
2883                        }
2884                        if !nvt_body.contains("<StorageClass>") {
2885                            return Err(malformed());
2886                        }
2887                        nvt_remaining = &nvt_after[nvt_end + 30..];
2888                    } else {
2889                        break;
2890                    }
2891                }
2892            }
2893
2894            remaining = &after[rule_end + 7..];
2895        } else {
2896            break;
2897        }
2898    }
2899
2900    Ok(())
2901}
2902
2903/// Parsed CORS rule from bucket configuration XML.
2904pub(crate) struct CorsRule {
2905    allowed_origins: Vec<String>,
2906    allowed_methods: Vec<String>,
2907    allowed_headers: Vec<String>,
2908    expose_headers: Vec<String>,
2909    max_age_seconds: Option<u32>,
2910}
2911
2912/// Parse CORS configuration XML into rules.
2913pub(crate) fn parse_cors_config(xml: &str) -> Vec<CorsRule> {
2914    let mut rules = Vec::new();
2915    let mut remaining = xml;
2916    while let Some(start) = remaining.find("<CORSRule>") {
2917        let after = &remaining[start + 10..];
2918        if let Some(end) = after.find("</CORSRule>") {
2919            let block = &after[..end];
2920            let allowed_origins = extract_all_xml_values(block, "AllowedOrigin");
2921            let allowed_methods = extract_all_xml_values(block, "AllowedMethod");
2922            let allowed_headers = extract_all_xml_values(block, "AllowedHeader");
2923            let expose_headers = extract_all_xml_values(block, "ExposeHeader");
2924            let max_age_seconds =
2925                extract_xml_value(block, "MaxAgeSeconds").and_then(|s| s.parse().ok());
2926            rules.push(CorsRule {
2927                allowed_origins,
2928                allowed_methods,
2929                allowed_headers,
2930                expose_headers,
2931                max_age_seconds,
2932            });
2933            remaining = &after[end + 11..];
2934        } else {
2935            break;
2936        }
2937    }
2938    rules
2939}
2940
2941/// Match an origin against a CORS allowed origin pattern (supports "*" wildcard).
2942pub(crate) fn origin_matches(origin: &str, pattern: &str) -> bool {
2943    if pattern == "*" {
2944        return true;
2945    }
2946    // Simple wildcard: *.example.com
2947    if let Some(suffix) = pattern.strip_prefix('*') {
2948        return origin.ends_with(suffix);
2949    }
2950    origin == pattern
2951}
2952
2953/// Find the matching CORS rule for a given origin and method.
2954pub(crate) fn find_cors_rule<'a>(
2955    rules: &'a [CorsRule],
2956    origin: &str,
2957    method: Option<&str>,
2958) -> Option<&'a CorsRule> {
2959    rules.iter().find(|rule| {
2960        let origin_ok = rule
2961            .allowed_origins
2962            .iter()
2963            .any(|o| origin_matches(origin, o));
2964        let method_ok = match method {
2965            Some(m) => rule.allowed_methods.iter().any(|am| am == m),
2966            None => true,
2967        };
2968        origin_ok && method_ok
2969    })
2970}
2971
2972/// Check if an object is locked (retention or legal hold) and should block mutation.
2973/// Returns an error string if locked, None if allowed.
2974pub(crate) fn check_object_lock_for_overwrite(
2975    obj: &S3Object,
2976    req: &AwsRequest,
2977) -> Option<&'static str> {
2978    // Legal hold blocks overwrite
2979    if obj.lock_legal_hold.as_deref() == Some("ON") {
2980        return Some("AccessDenied");
2981    }
2982    // Retention check
2983    if let (Some(mode), Some(until)) = (&obj.lock_mode, &obj.lock_retain_until) {
2984        if *until > Utc::now() {
2985            if mode == "COMPLIANCE" {
2986                return Some("AccessDenied");
2987            }
2988            if mode == "GOVERNANCE" {
2989                let bypass = req
2990                    .headers
2991                    .get("x-amz-bypass-governance-retention")
2992                    .and_then(|v| v.to_str().ok())
2993                    .map(|s| s.eq_ignore_ascii_case("true"))
2994                    .unwrap_or(false);
2995                if !bypass {
2996                    return Some("AccessDenied");
2997                }
2998            }
2999        }
3000    }
3001    None
3002}
3003
3004#[cfg(test)]
3005mod tests;
3006
3007#[cfg(test)]
3008mod s3_detect_action_tests {
3009    //! Auth-mapping regressions for bug-hunt 2026-06-15 §5.3: ops that
3010    //! were unmapped (-> enforcement skipped) or mapped to the wrong
3011    //! action (-> the policy targeting them never applied).
3012    use super::s3_detect_action;
3013    use std::collections::HashMap;
3014
3015    fn q(pairs: &[(&str, &str)]) -> HashMap<String, String> {
3016        pairs
3017            .iter()
3018            .map(|(k, v)| (k.to_string(), v.to_string()))
3019            .collect()
3020    }
3021
3022    #[test]
3023    fn select_object_content_maps_to_get_object() {
3024        // POST /bucket/key?select&select-type=2 reads object data.
3025        let action = s3_detect_action(
3026            "POST",
3027            Some("bucket"),
3028            Some("key"),
3029            &q(&[("select", ""), ("select-type", "2")]),
3030        );
3031        assert_eq!(action, Some("GetObject"));
3032    }
3033
3034    #[test]
3035    fn write_get_object_response_is_detected() {
3036        // POST /WriteGetObjectResponse (no key) — Object Lambda data plane.
3037        let action = s3_detect_action("POST", Some("WriteGetObjectResponse"), None, &q(&[]));
3038        assert_eq!(action, Some("WriteGetObjectResponse"));
3039    }
3040
3041    #[test]
3042    fn list_directory_buckets_is_distinct_from_list_buckets() {
3043        // GET /?x-id=ListDirectoryBuckets -> s3express action, not ListBuckets.
3044        let dir = s3_detect_action("GET", None, None, &q(&[("x-id", "ListDirectoryBuckets")]));
3045        assert_eq!(dir, Some("ListAllMyDirectoryBuckets"));
3046        // Plain GET / is still ListBuckets.
3047        let plain = s3_detect_action("GET", None, None, &q(&[]));
3048        assert_eq!(plain, Some("ListBuckets"));
3049    }
3050
3051    #[test]
3052    fn rename_object_is_detected_with_key() {
3053        // PUT /bucket/newkey?renameObject carries a key — previously fell
3054        // through to PutObject so s3:RenameObject denies were ineffective.
3055        let action = s3_detect_action(
3056            "PUT",
3057            Some("bucket"),
3058            Some("newkey"),
3059            &q(&[("renameObject", "")]),
3060        );
3061        assert_eq!(action, Some("RenameObject"));
3062    }
3063
3064    #[test]
3065    fn update_object_encryption_maps_to_put_object() {
3066        // PUT /bucket/key?encryption changes object SSE; AWS gates this on
3067        // s3:PutObject (no distinct public action). Deliberate, not a
3068        // silent PutObject fall-through.
3069        let action = s3_detect_action(
3070            "PUT",
3071            Some("bucket"),
3072            Some("key"),
3073            &q(&[("encryption", "")]),
3074        );
3075        assert_eq!(action, Some("PutObject"));
3076    }
3077
3078    #[test]
3079    fn create_session_not_misdetected_as_list_objects() {
3080        // GET /bucket?session — previously fell through to ListObjects.
3081        let action = s3_detect_action("GET", Some("bucket"), None, &q(&[("session", "")]));
3082        assert_eq!(action, Some("CreateSession"));
3083    }
3084
3085    #[test]
3086    fn get_bucket_abac_not_misdetected_as_list_objects() {
3087        // GET /bucket?abac — previously fell through to ListObjects.
3088        let action = s3_detect_action("GET", Some("bucket"), None, &q(&[("abac", "")]));
3089        assert_eq!(action, Some("GetBucketAbacConfiguration"));
3090    }
3091
3092    #[test]
3093    fn put_bucket_abac_still_detected() {
3094        let action = s3_detect_action("PUT", Some("bucket"), None, &q(&[("abac", "")]));
3095        assert_eq!(action, Some("PutBucketAbacConfiguration"));
3096    }
3097}
3098
3099#[cfg(test)]
3100mod s3_iam_service_prefix_tests {
3101    //! §5.3: CreateSession and WriteGetObjectResponse are authorized under
3102    //! sibling IAM service prefixes (s3express / s3-object-lambda), not s3.
3103    use super::*;
3104    use parking_lot::RwLock;
3105    use std::sync::Arc;
3106
3107    fn make_service() -> S3Service {
3108        let state: SharedS3State = Arc::new(RwLock::new(
3109            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
3110        ));
3111        S3Service::new(state, Arc::new(DeliveryBus::new()))
3112    }
3113
3114    fn req(method: Method, path: &str, query: &[(&str, &str)]) -> AwsRequest {
3115        let path_segments: Vec<String> = path
3116            .split('/')
3117            .filter(|s| !s.is_empty())
3118            .map(|s| s.to_string())
3119            .collect();
3120        AwsRequest {
3121            service: "s3".to_string(),
3122            action: String::new(),
3123            region: "us-east-1".to_string(),
3124            account_id: "123456789012".to_string(),
3125            request_id: "rid".to_string(),
3126            headers: http::HeaderMap::new(),
3127            query_params: query
3128                .iter()
3129                .map(|(k, v)| (k.to_string(), v.to_string()))
3130                .collect(),
3131            body: bytes::Bytes::new(),
3132            body_stream: parking_lot::Mutex::new(None),
3133            path_segments,
3134            raw_path: path.to_string(),
3135            raw_query: String::new(),
3136            method,
3137            is_query_protocol: false,
3138            access_key_id: None,
3139            principal: None,
3140        }
3141    }
3142
3143    #[test]
3144    fn create_session_uses_s3express_prefix() {
3145        let svc = make_service();
3146        let action = svc
3147            .iam_action_for(&req(Method::GET, "/mybucket", &[("session", "")]))
3148            .expect("must be mapped");
3149        assert_eq!(action.service, "s3express");
3150        assert_eq!(action.action, "CreateSession");
3151        assert_eq!(action.action_string(), "s3express:CreateSession");
3152    }
3153
3154    #[test]
3155    fn write_get_object_response_uses_object_lambda_prefix() {
3156        let svc = make_service();
3157        let action = svc
3158            .iam_action_for(&req(Method::POST, "/WriteGetObjectResponse", &[]))
3159            .expect("must be mapped");
3160        assert_eq!(action.service, "s3-object-lambda");
3161        assert_eq!(action.action, "WriteGetObjectResponse");
3162        assert_eq!(
3163            action.action_string(),
3164            "s3-object-lambda:WriteGetObjectResponse"
3165        );
3166    }
3167
3168    #[test]
3169    fn get_object_torrent_maps_to_its_own_action() {
3170        // GET object with `?torrent` must authorize under s3:GetObjectTorrent,
3171        // not s3:GetObject (bug-audit 2026-06-20, 5.3).
3172        let svc = make_service();
3173        let action = svc
3174            .iam_action_for(&req(Method::GET, "/mybucket/key.txt", &[("torrent", "")]))
3175            .expect("must be mapped");
3176        assert_eq!(action.action, "GetObjectTorrent");
3177        // A plain GET still maps to GetObject.
3178        let plain = svc
3179            .iam_action_for(&req(Method::GET, "/mybucket/key.txt", &[]))
3180            .expect("must be mapped");
3181        assert_eq!(plain.action, "GetObject");
3182    }
3183}
3184
3185#[cfg(test)]
3186mod extract_xml_value_tests {
3187    use super::{decode_xml_entities, extract_xml_value};
3188
3189    #[test]
3190    fn returns_inner_value() {
3191        assert_eq!(
3192            extract_xml_value("<Root><Key>value</Key></Root>", "Key"),
3193            Some("value".to_string())
3194        );
3195    }
3196
3197    // The wire form of a key `a&b` is `<Key>a&amp;b</Key>`; the parsed value
3198    // must decode back to `a&b` so it matches the URL-decoded key that
3199    // arrives via the request path (otherwise DeleteObjects/Tagging silently
3200    // miss the object).
3201    #[test]
3202    fn decodes_named_and_numeric_entities() {
3203        assert_eq!(
3204            extract_xml_value("<Key>a&amp;b</Key>", "Key"),
3205            Some("a&b".to_string())
3206        );
3207        assert_eq!(
3208            extract_xml_value("<Key>&lt;x&gt; &quot;y&quot; &apos;z&apos;</Key>", "Key"),
3209            Some("<x> \"y\" 'z'".to_string())
3210        );
3211        assert_eq!(
3212            extract_xml_value("<Key>a&#38;b&#x26;c</Key>", "Key"),
3213            Some("a&b&c".to_string())
3214        );
3215    }
3216
3217    // A single left-to-right pass must resolve each entity exactly once:
3218    // `&amp;lt;` is the escaped literal text `&lt;`, so it decodes to `&lt;`,
3219    // NOT to `<` (which a chain of `str::replace` calls would wrongly produce).
3220    #[test]
3221    fn does_not_double_decode() {
3222        assert_eq!(decode_xml_entities("&amp;lt;"), "&lt;");
3223        assert_eq!(decode_xml_entities("&amp;amp;"), "&amp;");
3224    }
3225
3226    #[test]
3227    fn leaves_bare_ampersand_and_unknown_entities_verbatim() {
3228        assert_eq!(decode_xml_entities("a & b"), "a & b");
3229        assert_eq!(decode_xml_entities("50% &bogus; done"), "50% &bogus; done");
3230        assert_eq!(decode_xml_entities("no entities here"), "no entities here");
3231    }
3232
3233    #[test]
3234    fn missing_tag_is_none() {
3235        assert_eq!(extract_xml_value("<Root></Root>", "Key"), None);
3236    }
3237
3238    // bug-audit 2026-05-28, 2.2: a closing tag before the opening one used to
3239    // slice with end < start and panic; must return None instead.
3240    #[test]
3241    fn close_before_open_does_not_panic() {
3242        assert_eq!(extract_xml_value("</Key>oops<Key>value", "Key"), None);
3243    }
3244
3245    #[test]
3246    fn open_without_close_is_none() {
3247        assert_eq!(extract_xml_value("<Key>value", "Key"), None);
3248    }
3249}
3250
3251#[cfg(test)]
3252mod compute_checksum_tests {
3253    use super::{compute_checksum, compute_checksum_streaming};
3254
3255    // bug-audit 2026-06-15, 1.7: CopyObject / CompleteMultipartUpload routed
3256    // through the non-streaming compute_checksum, whose `_ =>` arm returned an
3257    // empty string for CRC32C / CRC64NVME -> GetObjectAttributes reported an
3258    // empty checksum and integrity checks silently broke.
3259    #[test]
3260    fn crc32c_is_non_empty_and_matches_known_vector() {
3261        // CRC32C of "123456789" is 0xE3069283; base64 of those 4 BE bytes.
3262        let out = compute_checksum("CRC32C", b"123456789");
3263        assert!(!out.is_empty());
3264        let bytes = base64::engine::general_purpose::STANDARD
3265            .decode(&out)
3266            .unwrap();
3267        assert_eq!(bytes, 0xE306_9283u32.to_be_bytes());
3268    }
3269
3270    #[test]
3271    fn crc64nvme_is_non_empty() {
3272        let out = compute_checksum("CRC64NVME", b"hello world");
3273        assert!(!out.is_empty());
3274        // 8 BE bytes of a u64.
3275        let bytes = base64::engine::general_purpose::STANDARD
3276            .decode(&out)
3277            .unwrap();
3278        assert_eq!(bytes.len(), 8);
3279    }
3280
3281    #[tokio::test]
3282    async fn non_streaming_matches_streaming_for_all_algorithms() {
3283        let data = b"the quick brown fox jumps over the lazy dog".repeat(100);
3284        let tmp = tempfile::NamedTempFile::new().unwrap();
3285        tokio::fs::write(tmp.path(), &data).await.unwrap();
3286
3287        for algo in ["CRC32", "CRC32C", "CRC64NVME", "SHA1", "SHA256"] {
3288            let direct = compute_checksum(algo, &data);
3289            let streamed = compute_checksum_streaming(algo, tmp.path()).await.unwrap();
3290            assert!(!direct.is_empty(), "{algo} direct empty");
3291            assert_eq!(direct, streamed, "{algo} direct != streaming");
3292        }
3293    }
3294
3295    use base64::Engine as _;
3296}
3297
3298#[cfg(test)]
3299mod complete_multipart_parse_tests {
3300    use super::parse_complete_multipart_xml;
3301
3302    // 2026-07-07: aws-sdk-go-v2 (the terraform AWS provider) serializes the
3303    // part ETag with the quote char as the numeric reference `&#34;`, e.g.
3304    // <ETag>&#34;<hex>&#34;</ETag>. The parser must decode that to the bare hex
3305    // so it matches the stored part ETag; otherwise every large go-SDK
3306    // multipart upload fails CompleteMultipartUpload with InvalidPart.
3307    #[test]
3308    fn parses_go_sdk_numeric_quote_entity_etag() {
3309        let xml = "<CompleteMultipartUpload>\
3310            <Part><PartNumber>1</PartNumber><ETag>&#34;100bffa249c1cbde1a66242835cdab03&#34;</ETag></Part>\
3311            <Part><PartNumber>2</PartNumber><ETag>&#x22;5df1aff8ceab081b20d677b2d4f7eab1&#x22;</ETag></Part>\
3312            </CompleteMultipartUpload>";
3313        let parts = parse_complete_multipart_xml(xml);
3314        assert_eq!(
3315            parts,
3316            vec![
3317                (1, "100bffa249c1cbde1a66242835cdab03".to_string()),
3318                (2, "5df1aff8ceab081b20d677b2d4f7eab1".to_string()),
3319            ]
3320        );
3321    }
3322
3323    // aws-cli / boto send the quote as a literal `"` or the named entity
3324    // `&quot;` — both must still parse to the same bare hex.
3325    #[test]
3326    fn parses_cli_literal_and_named_quote_etag() {
3327        let xml = "<CompleteMultipartUpload>\
3328            <Part><PartNumber>1</PartNumber><ETag>\"abc123\"</ETag></Part>\
3329            <Part><PartNumber>2</PartNumber><ETag>&quot;def456&quot;</ETag></Part>\
3330            </CompleteMultipartUpload>";
3331        let parts = parse_complete_multipart_xml(xml);
3332        assert_eq!(
3333            parts,
3334            vec![(1, "abc123".to_string()), (2, "def456".to_string())]
3335        );
3336    }
3337}