rc_core/
transfer_options.rs

1//! Backend-neutral options for faithful object reads, writes, and copies.
2
3use std::collections::HashMap;
4use std::fmt;
5
6use base64::Engine;
7use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
8use http::header::{HeaderName, HeaderValue};
9use jiff::Timestamp;
10use zeroize::Zeroizing;
11
12use crate::encryption::ObjectEncryptionRequest;
13use crate::object_lock::{LegalHoldStatus, ObjectRetention};
14use crate::traits::{CopyObjectOptions, ObjectReadOptions};
15use crate::{Error, Result};
16
17const S3_MULTIPART_CHECKSUM_MAX_PARTS: u32 = 10_000;
18
19/// Standard HTTP object attributes and user-defined metadata.
20#[derive(Debug, Clone, Default, PartialEq, Eq)]
21pub struct ObjectAttributes {
22    /// Media type stored with the object.
23    pub content_type: Option<String>,
24    /// Cache policy stored with the object.
25    pub cache_control: Option<String>,
26    /// Content disposition stored with the object.
27    pub content_disposition: Option<String>,
28    /// Content encoding stored with the object.
29    pub content_encoding: Option<String>,
30    /// Content language stored with the object.
31    pub content_language: Option<String>,
32    /// Optional expiry timestamp stored with the object.
33    pub expires: Option<Timestamp>,
34    /// User-defined metadata without the protocol header prefix.
35    pub user_metadata: HashMap<String, String>,
36}
37
38impl ObjectAttributes {
39    /// Validate values that would otherwise be ambiguous at the S3 boundary.
40    pub fn validate(&self) -> Result<()> {
41        for (name, value) in [
42            ("Content-Type", self.content_type.as_deref()),
43            ("Cache-Control", self.cache_control.as_deref()),
44            ("Content-Disposition", self.content_disposition.as_deref()),
45            ("Content-Encoding", self.content_encoding.as_deref()),
46            ("Content-Language", self.content_language.as_deref()),
47        ] {
48            if let Some(value) = value {
49                if value.trim().is_empty() {
50                    return Err(Error::InvalidPath(format!("{name} cannot be empty")));
51                }
52                HeaderValue::from_str(value).map_err(|_| {
53                    Error::InvalidPath(format!("{name} contains invalid HTTP header characters"))
54                })?;
55            }
56        }
57        for (key, value) in &self.user_metadata {
58            if key.trim().is_empty() {
59                return Err(Error::InvalidPath(
60                    "User metadata keys cannot be empty".to_string(),
61                ));
62            }
63            let header_name = format!("x-amz-meta-{key}");
64            HeaderName::from_bytes(header_name.as_bytes()).map_err(|_| {
65                Error::InvalidPath(format!(
66                    "User metadata key '{key}' cannot be represented as an HTTP header"
67                ))
68            })?;
69            HeaderValue::from_str(value).map_err(|_| {
70                Error::InvalidPath(format!(
71                    "User metadata value for '{key}' contains invalid HTTP header characters"
72                ))
73            })?;
74        }
75        Ok(())
76    }
77
78    fn contains_only_content_type(&self) -> bool {
79        self.cache_control.is_none()
80            && self.content_disposition.is_none()
81            && self.content_encoding.is_none()
82            && self.content_language.is_none()
83            && self.expires.is_none()
84            && self.user_metadata.is_empty()
85    }
86}
87
88/// How CopyObject handles source metadata.
89#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub enum MetadataDirective {
91    /// Copy metadata from the selected source object.
92    Copy,
93    /// Replace source metadata with the provided destination attributes.
94    Replace,
95}
96
97/// How CopyObject handles source tags.
98#[derive(Debug, Clone, Copy, PartialEq, Eq)]
99pub enum TaggingDirective {
100    /// Copy tags from the selected source object.
101    Copy,
102    /// Replace source tags with the provided destination tags.
103    Replace,
104}
105
106/// Checksum algorithms understood by S3-compatible object APIs.
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108pub enum ChecksumAlgorithm {
109    /// CRC-32.
110    Crc32,
111    /// CRC-32C.
112    Crc32c,
113    /// CRC-64/NVME.
114    Crc64Nvme,
115    /// SHA-1.
116    Sha1,
117    /// SHA-256.
118    Sha256,
119}
120
121/// An encoded checksum paired with its algorithm.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct ObjectChecksum {
124    /// Algorithm used to calculate the checksum.
125    pub algorithm: ChecksumAlgorithm,
126    /// Protocol-encoded checksum value.
127    pub value: String,
128}
129
130/// How a destination checksum should be supplied.
131#[derive(Debug, Clone, PartialEq, Eq)]
132pub enum ChecksumRequest {
133    /// Ask the adapter and service to calculate the selected algorithm.
134    Calculate(ChecksumAlgorithm),
135    /// Send a checksum that the caller already calculated.
136    Precomputed(ObjectChecksum),
137}
138
139impl ChecksumRequest {
140    fn validate(&self) -> Result<()> {
141        match self {
142            Self::Calculate(_) => Ok(()),
143            Self::Precomputed(checksum) => checksum.validate(),
144        }
145    }
146}
147
148/// Fidelity metadata returned independently of the stable ObjectInfo output contract.
149#[derive(Debug, Clone, Default, PartialEq, Eq)]
150pub struct ObjectTransferMetadata {
151    /// Standard HTTP attributes and user-defined metadata.
152    pub attributes: ObjectAttributes,
153    /// Storage class reported by the service.
154    pub storage_class: Option<String>,
155    /// Persisted checksums reported by checksum-mode metadata reads.
156    pub checksums: Vec<ObjectChecksum>,
157}
158
159impl ObjectChecksum {
160    /// Build a checksum while rejecting an ambiguous empty value.
161    pub fn new(algorithm: ChecksumAlgorithm, value: impl Into<String>) -> Result<Self> {
162        let checksum = Self {
163            algorithm,
164            value: value.into(),
165        };
166        checksum.validate()?;
167        Ok(checksum)
168    }
169
170    /// Build a checksum reported by a metadata read.
171    ///
172    /// Multipart SHA and CRC checksums use the S3 composite form
173    /// `<base64-digest>-<part-count>`, which is not valid as a precomputed
174    /// full-object write checksum.
175    pub fn new_persisted(algorithm: ChecksumAlgorithm, value: impl Into<String>) -> Result<Self> {
176        let checksum = Self {
177            algorithm,
178            value: value.into(),
179        };
180        if checksum.validate().is_ok() {
181            return Ok(checksum);
182        }
183
184        let (digest, part_count) = checksum.value.rsplit_once('-').ok_or_else(|| {
185            Error::InvalidPath("Persisted object checksum is not valid Base64".to_string())
186        })?;
187        let part_count = part_count.parse::<u32>().map_err(|_| {
188            Error::InvalidPath(
189                "Composite checksum part count must be a positive integer".to_string(),
190            )
191        })?;
192        if part_count == 0
193            || part_count > S3_MULTIPART_CHECKSUM_MAX_PARTS
194            || checksum.algorithm == ChecksumAlgorithm::Crc64Nvme
195        {
196            return Err(Error::InvalidPath(
197                "Persisted composite checksum is not valid for this algorithm".to_string(),
198            ));
199        }
200        let decoded = BASE64_STANDARD.decode(digest).map_err(|_| {
201            Error::InvalidPath("Composite checksum digest must be valid Base64".to_string())
202        })?;
203        let expected_length = match checksum.algorithm {
204            ChecksumAlgorithm::Crc32 | ChecksumAlgorithm::Crc32c => 4,
205            ChecksumAlgorithm::Sha1 => 20,
206            ChecksumAlgorithm::Sha256 => 32,
207            ChecksumAlgorithm::Crc64Nvme => 8,
208        };
209        if decoded.len() != expected_length {
210            return Err(Error::InvalidPath(format!(
211                "Composite checksum for {:?} must decode to {expected_length} bytes",
212                checksum.algorithm
213            )));
214        }
215        Ok(checksum)
216    }
217
218    /// Validate the checksum value before a request is attempted.
219    pub fn validate(&self) -> Result<()> {
220        if self.value.trim().is_empty() {
221            return Err(Error::InvalidPath(
222                "Object checksum value cannot be empty".to_string(),
223            ));
224        }
225        let decoded = BASE64_STANDARD
226            .decode(&self.value)
227            .map_err(|_| Error::InvalidPath("Object checksum must be valid Base64".to_string()))?;
228        let expected_length = match self.algorithm {
229            ChecksumAlgorithm::Crc32 | ChecksumAlgorithm::Crc32c => 4,
230            ChecksumAlgorithm::Crc64Nvme => 8,
231            ChecksumAlgorithm::Sha1 => 20,
232            ChecksumAlgorithm::Sha256 => 32,
233        };
234        if decoded.len() != expected_length {
235            return Err(Error::InvalidPath(format!(
236                "Object checksum for {:?} must decode to {expected_length} bytes",
237                self.algorithm
238            )));
239        }
240        Ok(())
241    }
242}
243
244/// A 256-bit SSE-C key whose formatting never reveals key material.
245#[derive(Clone, PartialEq, Eq)]
246pub struct SseCustomerKey(Zeroizing<Vec<u8>>);
247
248impl SseCustomerKey {
249    /// Store an exact 256-bit key in zeroizing memory.
250    pub fn new(key: Vec<u8>) -> Result<Self> {
251        let key = Zeroizing::new(key);
252        if key.len() != 32 {
253            return Err(Error::InvalidPath(
254                "SSE-C keys must contain exactly 32 bytes".to_string(),
255            ));
256        }
257        Ok(Self(key))
258    }
259
260    /// Borrow key bytes for protocol adapters.
261    ///
262    /// Callers must not log, serialize, or include this value in errors.
263    pub fn expose_secret(&self) -> &[u8] {
264        self.0.as_slice()
265    }
266}
267
268impl fmt::Debug for SseCustomerKey {
269    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
270        formatter.write_str("SseCustomerKey([REDACTED])")
271    }
272}
273
274/// Encryption requested for a destination object.
275#[derive(Debug, Clone, PartialEq, Eq)]
276pub enum ObjectWriteEncryption {
277    /// Existing S3-managed or KMS-managed encryption request.
278    Managed(ObjectEncryptionRequest),
279    /// Customer-provided key used only for this transfer.
280    SseCustomer {
281        /// Redacted, zeroizing customer key.
282        key: SseCustomerKey,
283    },
284}
285
286impl From<ObjectEncryptionRequest> for ObjectWriteEncryption {
287    fn from(value: ObjectEncryptionRequest) -> Self {
288        Self::Managed(value)
289    }
290}
291
292/// Complete destination options for an object write or copy.
293#[derive(Debug, Clone, Default, PartialEq, Eq)]
294pub struct ObjectWriteOptions {
295    /// Attributes to store. `Some(Default::default())` represents an explicit empty replacement.
296    pub attributes: Option<ObjectAttributes>,
297    /// Tags to store. `Some(HashMap::new())` represents an explicit empty replacement.
298    pub tags: Option<HashMap<String, String>>,
299    /// Requested destination storage class.
300    pub storage_class: Option<String>,
301    /// Checksum supplied or selected for the write.
302    pub checksum: Option<ChecksumRequest>,
303    /// Destination encryption policy.
304    pub encryption: Option<ObjectWriteEncryption>,
305    /// Retention applied atomically with object creation.
306    pub retention: Option<ObjectRetention>,
307    /// Legal-hold state applied atomically with object creation.
308    pub legal_hold: Option<LegalHoldStatus>,
309}
310
311impl ObjectWriteOptions {
312    /// Validate fields before any backend mutation.
313    pub fn validate(&self) -> Result<()> {
314        if let Some(attributes) = &self.attributes {
315            attributes.validate()?;
316        }
317        if let Some(tags) = &self.tags {
318            if tags.len() > 10 {
319                return Err(Error::InvalidPath(
320                    "Objects can have at most 10 tags".to_string(),
321                ));
322            }
323            for (key, value) in tags {
324                let key_length = key.chars().count();
325                if !(1..=128).contains(&key_length) {
326                    return Err(Error::InvalidPath(
327                        "Object tag keys must contain between 1 and 128 characters".to_string(),
328                    ));
329                }
330                if value.chars().count() > 256 {
331                    return Err(Error::InvalidPath(
332                        "Object tag values cannot exceed 256 characters".to_string(),
333                    ));
334                }
335            }
336        }
337        if self
338            .storage_class
339            .as_deref()
340            .is_some_and(|value| value.trim().is_empty())
341        {
342            return Err(Error::InvalidPath(
343                "Storage class cannot be empty".to_string(),
344            ));
345        }
346        if let Some(checksum) = &self.checksum {
347            checksum.validate()?;
348        }
349        if self
350            .retention
351            .as_ref()
352            .is_some_and(|retention| retention.retain_until <= jiff::Timestamp::now())
353        {
354            return Err(Error::InvalidPath(
355                "Object retention date must be in the future".to_string(),
356            ));
357        }
358        if matches!(
359            self.encryption.as_ref(),
360            Some(ObjectWriteEncryption::Managed(
361                ObjectEncryptionRequest::SseKms { key_id }
362            )) if key_id.trim().is_empty()
363        ) {
364            return Err(Error::InvalidPath("KMS key ID cannot be empty".to_string()));
365        }
366        Ok(())
367    }
368
369    /// Translate the subset supported by the original ObjectStore put API.
370    ///
371    /// Backends use this to preserve compatibility without silently dropping advanced fields.
372    pub fn legacy_put_arguments(&self) -> Result<(Option<&str>, Option<&ObjectEncryptionRequest>)> {
373        self.validate()?;
374        let content_type = match &self.attributes {
375            Some(attributes) if attributes.contains_only_content_type() => {
376                attributes.content_type.as_deref()
377            }
378            Some(_) => {
379                return Err(Error::UnsupportedFeature(
380                    "Object attributes other than Content-Type require advanced write support"
381                        .to_string(),
382                ));
383            }
384            None => None,
385        };
386        if self.tags.is_some()
387            || self.storage_class.is_some()
388            || self.checksum.is_some()
389            || self.retention.is_some()
390            || self.legal_hold.is_some()
391        {
392            return Err(Error::UnsupportedFeature(
393                "Transfer fidelity options are not implemented by this object store".to_string(),
394            ));
395        }
396        let encryption = match self.encryption.as_ref() {
397            Some(ObjectWriteEncryption::Managed(request)) => Some(request),
398            Some(ObjectWriteEncryption::SseCustomer { .. }) => {
399                return Err(Error::UnsupportedFeature(
400                    "SSE-C writes are not implemented by this object store".to_string(),
401                ));
402            }
403            None => None,
404        };
405        Ok((content_type, encryption))
406    }
407
408    fn legacy_copy_encryption(&self) -> Result<Option<&ObjectEncryptionRequest>> {
409        self.validate()?;
410        if self.attributes.is_some()
411            || self.tags.is_some()
412            || self.storage_class.is_some()
413            || self.checksum.is_some()
414            || self.retention.is_some()
415            || self.legal_hold.is_some()
416        {
417            return Err(Error::UnsupportedFeature(
418                "Advanced copy destination options are not implemented by this object store"
419                    .to_string(),
420            ));
421        }
422        match self.encryption.as_ref() {
423            Some(ObjectWriteEncryption::Managed(request)) => Ok(Some(request)),
424            Some(ObjectWriteEncryption::SseCustomer { .. }) => Err(Error::UnsupportedFeature(
425                "Destination SSE-C is not implemented by this object store".to_string(),
426            )),
427            None => Ok(None),
428        }
429    }
430}
431
432/// Advanced read options that do not change the legacy ObjectReadOptions contract.
433#[derive(Debug, Clone, Default, PartialEq, Eq)]
434pub struct TransferReadOptions {
435    /// Exact object version to select.
436    pub version_id: Option<String>,
437    /// Request persisted checksum fields from the backend.
438    pub checksum_mode: bool,
439    /// Customer key needed to read an SSE-C source object.
440    pub customer_key: Option<SseCustomerKey>,
441}
442
443impl TransferReadOptions {
444    /// Validate read selection before issuing a request.
445    pub fn validate(&self) -> Result<()> {
446        if self.version_id.as_deref().is_some_and(str::is_empty) {
447            return Err(Error::InvalidPath("Version ID cannot be empty".to_string()));
448        }
449        Ok(())
450    }
451
452    pub(crate) fn legacy_read_options(&self) -> Result<ObjectReadOptions> {
453        self.validate()?;
454        if self.checksum_mode || self.customer_key.is_some() {
455            return Err(Error::UnsupportedFeature(
456                "Checksum-mode or SSE-C reads are not implemented by this object store".to_string(),
457            ));
458        }
459        ObjectReadOptions::for_version(self.version_id.clone())
460    }
461}
462
463impl From<ObjectReadOptions> for TransferReadOptions {
464    fn from(value: ObjectReadOptions) -> Self {
465        Self {
466            version_id: value.version_id,
467            ..Self::default()
468        }
469    }
470}
471
472/// Complete options for a faithful server-side object copy.
473#[derive(Debug, Clone, Default, PartialEq, Eq)]
474pub struct TransferCopyOptions {
475    /// Source version, checksum, and SSE-C selection.
476    pub source: TransferReadOptions,
477    /// Source metadata handling requested for the destination.
478    pub metadata_directive: Option<MetadataDirective>,
479    /// Source tag handling requested for the destination.
480    pub tagging_directive: Option<TaggingDirective>,
481    /// Destination attributes, tags, checksum, encryption, and lock state.
482    pub destination: ObjectWriteOptions,
483}
484
485impl TransferCopyOptions {
486    /// Validate directives and their replacement payloads before mutation.
487    pub fn validate(&self) -> Result<()> {
488        self.source.validate()?;
489        self.destination.validate()?;
490        match self.metadata_directive {
491            None | Some(MetadataDirective::Copy) if self.destination.attributes.is_some() => {
492                return Err(Error::InvalidPath(
493                    "Destination attributes require an explicit metadata REPLACE directive"
494                        .to_string(),
495                ));
496            }
497            Some(MetadataDirective::Replace) if self.destination.attributes.is_none() => {
498                return Err(Error::InvalidPath(
499                    "Metadata REPLACE requires an explicit attributes value".to_string(),
500                ));
501            }
502            _ => {}
503        }
504        match self.tagging_directive {
505            None | Some(TaggingDirective::Copy) if self.destination.tags.is_some() => {
506                return Err(Error::InvalidPath(
507                    "Destination tags require an explicit tagging REPLACE directive".to_string(),
508                ));
509            }
510            Some(TaggingDirective::Replace) if self.destination.tags.is_none() => {
511                return Err(Error::InvalidPath(
512                    "Tagging REPLACE requires an explicit tags value".to_string(),
513                ));
514            }
515            _ => {}
516        }
517        Ok(())
518    }
519
520    pub(crate) fn legacy_copy_arguments(
521        &self,
522    ) -> Result<(CopyObjectOptions, Option<&ObjectEncryptionRequest>)> {
523        self.validate()?;
524        if self.metadata_directive.is_some() || self.tagging_directive.is_some() {
525            return Err(Error::UnsupportedFeature(
526                "Metadata or tagging directives are not implemented by this object store"
527                    .to_string(),
528            ));
529        }
530        let source = self.source.legacy_read_options()?;
531        let encryption = self.destination.legacy_copy_encryption()?;
532        let copy = CopyObjectOptions::for_source_version(source.version_id)?;
533        Ok((copy, encryption))
534    }
535
536    pub(crate) fn validate_multipart_source_version(
537        &self,
538        multipart_source_version_id: Option<&str>,
539    ) -> Result<()> {
540        if self.source.version_id.is_some()
541            && self.source.version_id.as_deref() != multipart_source_version_id
542        {
543            return Err(Error::InvalidPath(
544                "Transfer and multipart source version IDs must match".to_string(),
545            ));
546        }
547        Ok(())
548    }
549}
550
551#[cfg(test)]
552mod tests {
553    use super::*;
554
555    #[test]
556    fn persisted_checksums_accept_valid_composite_values_only() {
557        let digest = BASE64_STANDARD.encode([7_u8; 32]);
558        let checksum =
559            ObjectChecksum::new_persisted(ChecksumAlgorithm::Sha256, format!("{digest}-3"))
560                .expect("valid composite checksum");
561        assert_eq!(checksum.value, format!("{digest}-3"));
562
563        for value in [
564            format!("{digest}-0"),
565            format!("{digest}-10001"),
566            format!("{digest}-not-a-number"),
567        ] {
568            assert!(matches!(
569                ObjectChecksum::new_persisted(ChecksumAlgorithm::Sha256, value),
570                Err(Error::InvalidPath(_))
571            ));
572        }
573        assert!(matches!(
574            ObjectChecksum::new_persisted(
575                ChecksumAlgorithm::Crc64Nvme,
576                format!("{}-2", BASE64_STANDARD.encode([3_u8; 8]))
577            ),
578            Err(Error::InvalidPath(_))
579        ));
580    }
581
582    #[test]
583    fn advanced_reads_cannot_fall_through_to_legacy_backends() {
584        let checksum_read = TransferReadOptions {
585            checksum_mode: true,
586            ..TransferReadOptions::default()
587        };
588        assert!(matches!(
589            checksum_read.legacy_read_options(),
590            Err(Error::UnsupportedFeature(_))
591        ));
592
593        let customer_key = SseCustomerKey::new(vec![3; 32]).expect("valid customer key");
594        let encrypted_read = TransferReadOptions {
595            customer_key: Some(customer_key),
596            ..TransferReadOptions::default()
597        };
598        assert!(matches!(
599            encrypted_read.legacy_read_options(),
600            Err(Error::UnsupportedFeature(_))
601        ));
602    }
603
604    #[test]
605    fn explicit_copy_directives_cannot_fall_through_to_legacy_backends() {
606        for options in [
607            TransferCopyOptions {
608                metadata_directive: Some(MetadataDirective::Copy),
609                ..TransferCopyOptions::default()
610            },
611            TransferCopyOptions {
612                tagging_directive: Some(TaggingDirective::Copy),
613                ..TransferCopyOptions::default()
614            },
615        ] {
616            assert!(matches!(
617                options.legacy_copy_arguments(),
618                Err(Error::UnsupportedFeature(_))
619            ));
620        }
621    }
622
623    #[test]
624    fn multipart_source_versions_must_match_when_transfer_selects_one() {
625        let options = TransferCopyOptions {
626            source: TransferReadOptions {
627                version_id: Some("source-v1".to_string()),
628                ..TransferReadOptions::default()
629            },
630            ..TransferCopyOptions::default()
631        };
632
633        options
634            .validate_multipart_source_version(Some("source-v1"))
635            .expect("matching source versions should be accepted");
636        assert!(matches!(
637            options.validate_multipart_source_version(Some("source-v2")),
638            Err(Error::InvalidPath(_))
639        ));
640        assert!(matches!(
641            options.validate_multipart_source_version(None),
642            Err(Error::InvalidPath(_))
643        ));
644    }
645}
rc_core/transfer_options.rs

rc_core/
transfer_options.rs
