minco_plugin_object_storage/
uploads.rs

1use crate::{
2    MemoryObjectStore, ObjectAccessSigner, ObjectKey, ObjectMetadata, ObjectStoragePlugin,
3    ObjectStore, ObjectStoreError, PresignedObjectRequest,
4};
5use async_trait::async_trait;
6use chrono::{DateTime, TimeDelta, Utc};
7use minco_core::{CapabilityProvision, Plugin, PluginContext, PluginDescriptor, PluginError};
8use semver::Version;
9use serde::{Deserialize, Serialize};
10use std::{
11    collections::{BTreeMap, BTreeSet},
12    sync::Arc,
13};
14use uuid::Uuid;
15
16const MAX_UPLOAD_EXPIRY_SECONDS: i64 = 24 * 60 * 60;
17const DEFAULT_UPLOAD_EXPIRY_SECONDS: i64 = 15 * 60;
18const UPLOAD_ID_ATTRIBUTE: &str = "minco.upload_id";
19const RESERVED_ATTRIBUTE_PREFIX: &str = "minco.";
20
21/// Metadata returned without loading an object's bytes into application memory.
22#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23pub struct ObjectHead {
24    pub key: ObjectKey,
25    pub content_type: String,
26    pub size_bytes: u64,
27    /// SHA-256 is optional because a provider can report metadata for legacy
28    /// objects that were not uploaded with a provider checksum.
29    pub sha256: Option<String>,
30    pub created_at: DateTime<Utc>,
31    #[serde(default)]
32    pub attributes: BTreeMap<String, String>,
33}
34
35impl ObjectHead {
36    fn from_metadata(key: ObjectKey, metadata: ObjectMetadata) -> Self {
37        Self {
38            key,
39            content_type: metadata.content_type,
40            size_bytes: metadata.size_bytes,
41            sha256: Some(metadata.sha256),
42            created_at: metadata.created_at,
43            attributes: metadata.attributes,
44        }
45    }
46}
47
48/// Provider port for an inexpensive metadata lookup such as S3 `HeadObject`.
49#[async_trait]
50pub trait ObjectMetadataReader: Send + Sync + std::fmt::Debug {
51    async fn head(&self, key: &ObjectKey) -> Result<Option<ObjectHead>, ObjectStoreError>;
52}
53
54#[async_trait]
55impl ObjectMetadataReader for MemoryObjectStore {
56    async fn head(&self, key: &ObjectKey) -> Result<Option<ObjectHead>, ObjectStoreError> {
57        ObjectStore::get(self, key).await.map(|object| {
58            object.map(|stored| ObjectHead::from_metadata(stored.key, stored.metadata))
59        })
60    }
61}
62
63/// Typed metadata service injected by a managed object-storage plugin.
64#[derive(Clone)]
65pub struct ObjectMetadataService(Arc<dyn ObjectMetadataReader>);
66
67impl std::fmt::Debug for ObjectMetadataService {
68    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
69        formatter.debug_tuple("ObjectMetadataService").finish()
70    }
71}
72
73impl ObjectMetadataService {
74    pub fn new(reader: Arc<dyn ObjectMetadataReader>) -> Self {
75        Self(reader)
76    }
77
78    pub async fn head(&self, key: &ObjectKey) -> Result<Option<ObjectHead>, ObjectStoreError> {
79        self.0.head(key).await
80    }
81}
82
83/// Exact provider request for one checksummed direct upload.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct SignObjectUpload {
86    pub key: ObjectKey,
87    pub content_type: String,
88    pub size_bytes: u64,
89    /// Lowercase hexadecimal SHA-256 of the complete object body.
90    pub sha256: String,
91    pub expires_in: TimeDelta,
92    pub attributes: BTreeMap<String, String>,
93}
94
95/// Provider adapter that can bind an upload capability to exact bytes.
96///
97/// This is separate from [`ObjectAccessSigner`] so the compatibility-preserving
98/// `PresignPutObject` contract can remain available while managed uploads require
99/// an exact size and SHA-256 checksum.
100#[async_trait]
101pub trait ObjectUploadSigner: Send + Sync + std::fmt::Debug {
102    async fn sign_upload(
103        &self,
104        request: SignObjectUpload,
105    ) -> Result<PresignedObjectRequest, ObjectUploadError>;
106}
107
108/// Closed upload policy owned by the application composition root.
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct ObjectUploadPolicy {
111    key_prefix: ObjectKey,
112    allowed_content_types: BTreeSet<String>,
113    maximum_size_bytes: u64,
114    expires_in: TimeDelta,
115}
116
117impl ObjectUploadPolicy {
118    pub fn new<I, S>(
119        key_prefix: ObjectKey,
120        maximum_size_bytes: u64,
121        allowed_content_types: I,
122    ) -> Result<Self, ObjectUploadError>
123    where
124        I: IntoIterator<Item = S>,
125        S: AsRef<str>,
126    {
127        if maximum_size_bytes == 0 {
128            return Err(ObjectUploadError::InvalidMaximumSize);
129        }
130        let allowed_content_types = allowed_content_types
131            .into_iter()
132            .map(|value| normalize_content_type(value.as_ref()))
133            .collect::<Result<BTreeSet<_>, _>>()?;
134        if allowed_content_types.is_empty() {
135            return Err(ObjectUploadError::EmptyContentTypeAllowlist);
136        }
137        Ok(Self {
138            key_prefix,
139            allowed_content_types,
140            maximum_size_bytes,
141            expires_in: TimeDelta::seconds(DEFAULT_UPLOAD_EXPIRY_SECONDS),
142        })
143    }
144
145    pub fn with_expiry(mut self, expires_in: TimeDelta) -> Result<Self, ObjectUploadError> {
146        validate_expiry(expires_in)?;
147        self.expires_in = expires_in;
148        Ok(self)
149    }
150
151    pub const fn key_prefix(&self) -> &ObjectKey {
152        &self.key_prefix
153    }
154
155    pub const fn maximum_size_bytes(&self) -> u64 {
156        self.maximum_size_bytes
157    }
158
159    pub const fn expires_in(&self) -> TimeDelta {
160        self.expires_in
161    }
162
163    pub fn allows_content_type(&self, content_type: &str) -> bool {
164        normalize_content_type(content_type)
165            .is_ok_and(|value| self.allowed_content_types.contains(&value))
166    }
167}
168
169/// Application-authorized request for one direct object upload.
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct IssueObjectUpload {
172    pub content_type: String,
173    pub size_bytes: u64,
174    /// Lowercase or uppercase hexadecimal SHA-256 of the complete file.
175    pub sha256: String,
176    #[serde(default)]
177    pub attributes: BTreeMap<String, String>,
178}
179
180/// Client-facing bearer capability. Send this value to the authorized client,
181/// but never persist or log it as the trusted upload record.
182#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
183pub struct ObjectUploadGrant {
184    pub key: ObjectKey,
185    pub request: PresignedObjectRequest,
186}
187
188/// Trusted server-side record for one issued upload capability.
189///
190/// Persist this value in application-owned state. Do not accept a replacement
191/// from an untrusted client as authorization.
192#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
193pub struct PendingObjectUpload {
194    pub key: ObjectKey,
195    pub expected_content_type: String,
196    pub expected_size_bytes: u64,
197    pub expected_sha256: String,
198    pub expected_attributes: BTreeMap<String, String>,
199    /// Expiry of the bearer upload capability, not of the pending record.
200    ///
201    /// Verification may happen after this timestamp when the provider accepted
202    /// the upload before the capability expired. Pending-record retention and
203    /// cleanup remain application-owned.
204    pub capability_expires_at: DateTime<Utc>,
205}
206
207/// Split result that prevents the bearer request from becoming trusted state by
208/// accident. Return only [`Self::grant`] to the client and retain
209/// [`Self::pending`] on the server.
210#[derive(Debug, Clone, PartialEq, Eq)]
211pub struct IssuedObjectUpload {
212    pub grant: ObjectUploadGrant,
213    pub pending: PendingObjectUpload,
214}
215
216/// Metadata accepted after the provider confirms the issued upload contract.
217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
218pub struct VerifiedObjectUpload {
219    pub key: ObjectKey,
220    pub metadata: ObjectHead,
221}
222
223/// Issues unique direct-upload capabilities and verifies their provider metadata.
224#[derive(Clone)]
225pub struct ObjectUploadService {
226    signer: Arc<dyn ObjectUploadSigner>,
227    metadata: ObjectMetadataService,
228    policy: ObjectUploadPolicy,
229}
230
231impl std::fmt::Debug for ObjectUploadService {
232    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
233        formatter
234            .debug_struct("ObjectUploadService")
235            .field("policy", &self.policy)
236            .finish_non_exhaustive()
237    }
238}
239
240impl ObjectUploadService {
241    pub const fn new(
242        signer: Arc<dyn ObjectUploadSigner>,
243        metadata: ObjectMetadataService,
244        policy: ObjectUploadPolicy,
245    ) -> Self {
246        Self {
247            signer,
248            metadata,
249            policy,
250        }
251    }
252
253    pub async fn issue(
254        &self,
255        request: IssueObjectUpload,
256    ) -> Result<IssuedObjectUpload, ObjectUploadError> {
257        let IssueObjectUpload {
258            content_type,
259            size_bytes,
260            sha256,
261            attributes,
262        } = request;
263        let content_type = normalize_content_type(&content_type)?;
264        if !self.policy.allowed_content_types.contains(&content_type) {
265            return Err(ObjectUploadError::UnsupportedContentType(content_type));
266        }
267        if size_bytes == 0 {
268            return Err(ObjectUploadError::EmptyObject);
269        }
270        if size_bytes > self.policy.maximum_size_bytes {
271            return Err(ObjectUploadError::ObjectTooLarge {
272                actual: size_bytes,
273                maximum: self.policy.maximum_size_bytes,
274            });
275        }
276        let sha256 = normalize_sha256(&sha256)?;
277        let mut attributes = validate_attributes(attributes)?;
278        let upload_id = Uuid::now_v7().to_string();
279        attributes.insert(UPLOAD_ID_ATTRIBUTE.to_owned(), upload_id.clone());
280        let key = generated_key(&self.policy.key_prefix, &upload_id)?;
281        let signed = self
282            .signer
283            .sign_upload(SignObjectUpload {
284                key: key.clone(),
285                content_type: content_type.clone(),
286                size_bytes,
287                sha256: sha256.clone(),
288                expires_in: self.policy.expires_in,
289                attributes: attributes.clone(),
290            })
291            .await?;
292        let capability_expires_at = signed.expires_at;
293        Ok(IssuedObjectUpload {
294            grant: ObjectUploadGrant {
295                key: key.clone(),
296                request: signed,
297            },
298            pending: PendingObjectUpload {
299                key,
300                expected_content_type: content_type,
301                expected_size_bytes: size_bytes,
302                expected_sha256: sha256,
303                expected_attributes: attributes,
304                capability_expires_at,
305            },
306        })
307    }
308
309    pub async fn verify(
310        &self,
311        pending: &PendingObjectUpload,
312    ) -> Result<VerifiedObjectUpload, ObjectUploadError> {
313        let Some(metadata) = self.metadata.head(&pending.key).await? else {
314            return Err(ObjectUploadError::MissingObject);
315        };
316        if metadata.key != pending.key {
317            return Err(ObjectUploadError::ObjectKeyMismatch);
318        }
319        let actual_content_type = normalize_content_type(&metadata.content_type)
320            .map_err(|_| ObjectUploadError::ContentTypeMismatch)?;
321        if actual_content_type != pending.expected_content_type {
322            return Err(ObjectUploadError::ContentTypeMismatch);
323        }
324        if metadata.size_bytes != pending.expected_size_bytes {
325            return Err(ObjectUploadError::ObjectSizeMismatch {
326                actual: metadata.size_bytes,
327                expected: pending.expected_size_bytes,
328            });
329        }
330        let actual_sha256 = metadata
331            .sha256
332            .as_deref()
333            .map(normalize_sha256)
334            .transpose()
335            .map_err(|_| ObjectUploadError::ChecksumMismatch)?;
336        if actual_sha256.as_deref() != Some(pending.expected_sha256.as_str()) {
337            return Err(ObjectUploadError::ChecksumMismatch);
338        }
339        if metadata.attributes != pending.expected_attributes {
340            return Err(ObjectUploadError::AttributeMismatch);
341        }
342        Ok(VerifiedObjectUpload {
343            key: pending.key.clone(),
344            metadata,
345        })
346    }
347}
348
349/// Object-storage plugin with the direct-upload and metadata lifecycle installed.
350#[derive(Debug, Clone)]
351pub struct ManagedObjectStoragePlugin {
352    storage: ObjectStoragePlugin,
353    metadata: ObjectMetadataService,
354    uploads: ObjectUploadService,
355}
356
357impl ManagedObjectStoragePlugin {
358    pub fn new<S>(
359        store: Arc<dyn ObjectStore>,
360        signer: Arc<S>,
361        metadata_reader: Arc<dyn ObjectMetadataReader>,
362        policy: ObjectUploadPolicy,
363    ) -> Self
364    where
365        S: ObjectAccessSigner + ObjectUploadSigner + 'static,
366    {
367        let access_signer: Arc<dyn ObjectAccessSigner> = signer.clone();
368        let upload_signer: Arc<dyn ObjectUploadSigner> = signer;
369        Self::new_with_signers(store, access_signer, upload_signer, metadata_reader, policy)
370    }
371
372    /// Construct managed storage with independent private-download and upload
373    /// signers while retaining static, typed composition.
374    pub fn new_with_signers(
375        store: Arc<dyn ObjectStore>,
376        access_signer: Arc<dyn ObjectAccessSigner>,
377        upload_signer: Arc<dyn ObjectUploadSigner>,
378        metadata_reader: Arc<dyn ObjectMetadataReader>,
379        policy: ObjectUploadPolicy,
380    ) -> Self {
381        let metadata = ObjectMetadataService::new(metadata_reader);
382        let uploads = ObjectUploadService::new(upload_signer, metadata.clone(), policy);
383        Self {
384            storage: ObjectStoragePlugin::new(store).with_access_signer(access_signer),
385            metadata,
386            uploads,
387        }
388    }
389}
390
391impl Plugin for ManagedObjectStoragePlugin {
392    fn descriptor(&self) -> PluginDescriptor {
393        let mut descriptor = self.storage.descriptor();
394        descriptor.provides.extend([
395            CapabilityProvision {
396                name: "storage.object.metadata".into(),
397                version: Version::new(1, 0, 0),
398            },
399            CapabilityProvision {
400                name: "storage.object.upload".into(),
401                version: Version::new(1, 0, 0),
402            },
403        ]);
404        descriptor
405    }
406
407    fn install(&self, context: &mut PluginContext<'_>) -> Result<(), PluginError> {
408        self.storage.install(context)?;
409        context.services().insert(Arc::new(self.metadata.clone()))?;
410        context.services().insert(Arc::new(self.uploads.clone()))?;
411        Ok(())
412    }
413}
414
415#[non_exhaustive]
416#[derive(Debug, thiserror::Error)]
417pub enum ObjectUploadError {
418    #[error("upload policy maximum size must be greater than zero")]
419    InvalidMaximumSize,
420    #[error("upload policy must allow at least one exact content type")]
421    EmptyContentTypeAllowlist,
422    #[error("upload content type is invalid")]
423    InvalidContentType,
424    #[error("upload content type is not allowed: {0}")]
425    UnsupportedContentType(String),
426    #[error("upload body must not be empty")]
427    EmptyObject,
428    #[error("upload SHA-256 must be exactly 64 hexadecimal characters")]
429    InvalidSha256,
430    #[error("upload attributes are invalid or use a reserved Minco key")]
431    InvalidAttributes,
432    #[error("upload expiry must be greater than zero and no more than 24 hours")]
433    InvalidExpiry,
434    #[error("the uploaded object does not exist")]
435    MissingObject,
436    #[error("the provider reported metadata for a different object key")]
437    ObjectKeyMismatch,
438    #[error("the uploaded object's content type does not match the issued capability")]
439    ContentTypeMismatch,
440    #[error("the uploaded object's signed attributes do not match the issued capability")]
441    AttributeMismatch,
442    #[error("the uploaded object's SHA-256 does not match the issued capability")]
443    ChecksumMismatch,
444    #[error("the requested upload is {actual} bytes; the policy maximum is {maximum} bytes")]
445    ObjectTooLarge { actual: u64, maximum: u64 },
446    #[error("the uploaded object is {actual} bytes; the issued size is {expected} bytes")]
447    ObjectSizeMismatch { actual: u64, expected: u64 },
448    #[error("the provider endpoint for the upload capability could not be resolved")]
449    EndpointResolution,
450    #[error("the signing credentials have an invalid expiration time")]
451    InvalidCredentialExpiry,
452    #[error("the signing credentials expire too soon to issue an upload capability")]
453    CredentialLifetimeTooShort,
454    #[error(transparent)]
455    ObjectStore(#[from] ObjectStoreError),
456}
457
458fn validate_expiry(expires_in: TimeDelta) -> Result<(), ObjectUploadError> {
459    if expires_in <= TimeDelta::zero() || expires_in > TimeDelta::seconds(MAX_UPLOAD_EXPIRY_SECONDS)
460    {
461        Err(ObjectUploadError::InvalidExpiry)
462    } else {
463        Ok(())
464    }
465}
466
467fn normalize_content_type(value: &str) -> Result<String, ObjectUploadError> {
468    let value = value.trim().to_ascii_lowercase();
469    let Some((top_level, subtype)) = value.split_once('/') else {
470        return Err(ObjectUploadError::InvalidContentType);
471    };
472    if value.len() > 255
473        || subtype.contains('/')
474        || !valid_media_token(top_level)
475        || !valid_media_token(subtype)
476    {
477        return Err(ObjectUploadError::InvalidContentType);
478    }
479    Ok(value)
480}
481
482fn valid_media_token(value: &str) -> bool {
483    !value.is_empty()
484        && value.bytes().all(|byte| {
485            byte.is_ascii_alphanumeric()
486                || matches!(
487                    byte,
488                    b'!' | b'#' | b'$' | b'&' | b'^' | b'_' | b'.' | b'+' | b'-'
489                )
490        })
491}
492
493fn normalize_sha256(value: &str) -> Result<String, ObjectUploadError> {
494    let value = value.trim();
495    if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
496        return Err(ObjectUploadError::InvalidSha256);
497    }
498    Ok(value.to_ascii_lowercase())
499}
500
501fn validate_attributes(
502    attributes: BTreeMap<String, String>,
503) -> Result<BTreeMap<String, String>, ObjectUploadError> {
504    if attributes.len() > 31
505        || attributes.iter().any(|(key, value)| {
506            key.trim().is_empty()
507                || key.starts_with(RESERVED_ATTRIBUTE_PREFIX)
508                || key.len() > 128
509                || value.len() > 1024
510                || key.chars().any(char::is_control)
511                || value.chars().any(char::is_control)
512        })
513    {
514        return Err(ObjectUploadError::InvalidAttributes);
515    }
516    Ok(attributes)
517}
518
519fn generated_key(prefix: &ObjectKey, upload_id: &str) -> Result<ObjectKey, ObjectUploadError> {
520    ObjectKey::parse(format!("{}/{upload_id}", prefix.as_str())).map_err(ObjectUploadError::from)
521}
522
523#[cfg(test)]
524mod tests {
525    use super::*;
526    use crate::{PresignGetObject, PresignPutObject, PresignedMethod, PutObject};
527    use minco_core::{PluginId, PluginManager, PluginSelection};
528    use sha2::{Digest, Sha256};
529
530    #[derive(Debug)]
531    struct TestSigner;
532
533    #[async_trait]
534    impl ObjectAccessSigner for TestSigner {
535        async fn sign_put(
536            &self,
537            request: PresignPutObject,
538        ) -> Result<PresignedObjectRequest, ObjectStoreError> {
539            Ok(PresignedObjectRequest {
540                method: PresignedMethod::Post,
541                url: "https://objects.example/upload".into(),
542                headers: BTreeMap::new(),
543                form_fields: BTreeMap::from([
544                    ("key".into(), request.key.as_str().to_owned()),
545                    ("content-type".into(), request.content_type),
546                    (
547                        "maximum-size".into(),
548                        request.maximum_size_bytes.to_string(),
549                    ),
550                ]),
551                expires_at: Utc::now() + request.expires_in,
552            })
553        }
554
555        async fn sign_get(
556            &self,
557            request: PresignGetObject,
558        ) -> Result<PresignedObjectRequest, ObjectStoreError> {
559            Ok(PresignedObjectRequest {
560                method: PresignedMethod::Get,
561                url: format!("https://objects.example/{}", request.key.as_str()),
562                headers: BTreeMap::new(),
563                form_fields: BTreeMap::new(),
564                expires_at: Utc::now() + request.expires_in,
565            })
566        }
567    }
568
569    #[async_trait]
570    impl ObjectUploadSigner for TestSigner {
571        async fn sign_upload(
572            &self,
573            request: SignObjectUpload,
574        ) -> Result<PresignedObjectRequest, ObjectUploadError> {
575            Ok(PresignedObjectRequest {
576                method: PresignedMethod::Post,
577                url: "https://objects.example/upload".into(),
578                headers: BTreeMap::new(),
579                form_fields: BTreeMap::from([
580                    ("key".into(), request.key.as_str().to_owned()),
581                    ("content-type".into(), request.content_type),
582                    ("size-bytes".into(), request.size_bytes.to_string()),
583                    ("sha256".into(), request.sha256),
584                ]),
585                expires_at: Utc::now() + request.expires_in,
586            })
587        }
588    }
589
590    fn sha256(bytes: &[u8]) -> String {
591        format!("{:x}", Sha256::digest(bytes))
592    }
593
594    fn policy() -> ObjectUploadPolicy {
595        ObjectUploadPolicy::new(
596            ObjectKey::parse("uploads/images").unwrap(),
597            1_024,
598            ["image/png", "image/jpeg"],
599        )
600        .unwrap()
601    }
602
603    fn service(store: Arc<MemoryObjectStore>) -> ObjectUploadService {
604        let metadata: Arc<dyn ObjectMetadataReader> = store;
605        ObjectUploadService::new(
606            Arc::new(TestSigner),
607            ObjectMetadataService::new(metadata),
608            policy(),
609        )
610    }
611
612    fn request() -> IssueObjectUpload {
613        IssueObjectUpload {
614            content_type: "IMAGE/PNG".into(),
615            size_bytes: 3,
616            sha256: sha256(b"png"),
617            attributes: BTreeMap::from([("tenant".into(), "acme".into())]),
618        }
619    }
620
621    #[test]
622    fn policy_is_closed_and_bounded() {
623        assert!(
624            ObjectUploadPolicy::new(ObjectKey::parse("uploads").unwrap(), 0, ["image/png"])
625                .is_err()
626        );
627        assert!(
628            ObjectUploadPolicy::new(
629                ObjectKey::parse("uploads").unwrap(),
630                1,
631                std::iter::empty::<&str>()
632            )
633            .is_err()
634        );
635        assert!(policy().with_expiry(TimeDelta::hours(25)).is_err());
636        assert!(policy().allows_content_type("IMAGE/PNG"));
637        assert!(!policy().allows_content_type("text/html"));
638    }
639
640    #[tokio::test]
641    async fn issuing_an_upload_separates_the_bearer_grant_from_trusted_state() {
642        let service = service(Arc::new(MemoryObjectStore::default()));
643        let first = service.issue(request()).await.unwrap();
644        let second = service.issue(request()).await.unwrap();
645        assert_ne!(first.grant.key, second.grant.key);
646        assert_eq!(first.grant.key, first.pending.key);
647        assert!(first.grant.key.as_str().starts_with("uploads/images/"));
648        assert!(
649            std::path::Path::new(first.grant.key.as_str())
650                .extension()
651                .is_none()
652        );
653        assert_eq!(first.pending.expected_content_type, "image/png");
654        assert_eq!(first.pending.expected_size_bytes, 3);
655        assert_eq!(first.pending.expected_sha256, sha256(b"png"));
656        assert!(
657            first
658                .pending
659                .expected_attributes
660                .contains_key(UPLOAD_ID_ATTRIBUTE)
661        );
662        assert_eq!(
663            first
664                .grant
665                .request
666                .form_fields
667                .get("size-bytes")
668                .map(String::as_str),
669            Some("3")
670        );
671        assert_eq!(
672            first.grant.request.expires_at,
673            first.pending.capability_expires_at
674        );
675    }
676
677    #[tokio::test]
678    async fn issuance_rejects_empty_oversized_or_unchecksummed_objects() {
679        let service = service(Arc::new(MemoryObjectStore::default()));
680
681        let mut empty = request();
682        empty.size_bytes = 0;
683        assert!(matches!(
684            service.issue(empty).await,
685            Err(ObjectUploadError::EmptyObject)
686        ));
687
688        let mut oversized = request();
689        oversized.size_bytes = 1_025;
690        assert!(matches!(
691            service.issue(oversized).await,
692            Err(ObjectUploadError::ObjectTooLarge {
693                actual: 1_025,
694                maximum: 1_024
695            })
696        ));
697
698        let mut invalid_checksum = request();
699        invalid_checksum.sha256 = "not-a-sha256".into();
700        assert!(matches!(
701            service.issue(invalid_checksum).await,
702            Err(ObjectUploadError::InvalidSha256)
703        ));
704    }
705
706    #[tokio::test]
707    async fn verification_accepts_only_the_issued_metadata_contract() {
708        let store = Arc::new(MemoryObjectStore::default());
709        let service = service(Arc::clone(&store));
710        let issued = service.issue(request()).await.unwrap();
711        ObjectStore::put(
712            store.as_ref(),
713            PutObject {
714                key: issued.pending.key.clone(),
715                bytes: b"png".to_vec(),
716                content_type: issued.pending.expected_content_type.clone(),
717                attributes: issued.pending.expected_attributes.clone(),
718            },
719        )
720        .await
721        .unwrap();
722        let verified = service.verify(&issued.pending).await.unwrap();
723        assert_eq!(verified.metadata.size_bytes, 3);
724        assert_eq!(
725            verified.metadata.sha256.as_deref(),
726            Some(sha256(b"png").as_str())
727        );
728
729        let mut wrong_size = issued.pending.clone();
730        wrong_size.expected_size_bytes = 4;
731        assert!(matches!(
732            service.verify(&wrong_size).await,
733            Err(ObjectUploadError::ObjectSizeMismatch {
734                actual: 3,
735                expected: 4
736            })
737        ));
738
739        let mut wrong_checksum = issued.pending.clone();
740        wrong_checksum.expected_sha256 = sha256(b"jpg");
741        assert!(matches!(
742            service.verify(&wrong_checksum).await,
743            Err(ObjectUploadError::ChecksumMismatch)
744        ));
745
746        let mut wrong_attributes = issued.pending;
747        wrong_attributes.expected_attributes.clear();
748        assert!(matches!(
749            service.verify(&wrong_attributes).await,
750            Err(ObjectUploadError::AttributeMismatch)
751        ));
752    }
753
754    #[derive(Debug)]
755    struct WrongKeyMetadataReader;
756
757    #[async_trait]
758    impl ObjectMetadataReader for WrongKeyMetadataReader {
759        async fn head(&self, _key: &ObjectKey) -> Result<Option<ObjectHead>, ObjectStoreError> {
760            Ok(Some(ObjectHead {
761                key: ObjectKey::parse("uploads/images/different").unwrap(),
762                content_type: "image/png".into(),
763                size_bytes: 3,
764                sha256: Some(sha256(b"png")),
765                created_at: Utc::now(),
766                attributes: BTreeMap::new(),
767            }))
768        }
769    }
770
771    #[tokio::test]
772    async fn verification_rejects_metadata_for_a_different_logical_key() {
773        let service = ObjectUploadService::new(
774            Arc::new(TestSigner),
775            ObjectMetadataService::new(Arc::new(WrongKeyMetadataReader)),
776            policy(),
777        );
778        let issued = service.issue(request()).await.unwrap();
779        assert!(matches!(
780            service.verify(&issued.pending).await,
781            Err(ObjectUploadError::ObjectKeyMismatch)
782        ));
783    }
784
785    #[test]
786    fn managed_plugin_advertises_and_installs_the_lifecycle() {
787        let store = Arc::new(MemoryObjectStore::default());
788        let object_store: Arc<dyn ObjectStore> = store.clone();
789        let metadata: Arc<dyn ObjectMetadataReader> = store;
790        let mut manager = PluginManager::default();
791        manager
792            .register(ManagedObjectStoragePlugin::new_with_signers(
793                object_store,
794                Arc::new(TestSigner),
795                Arc::new(TestSigner),
796                metadata,
797                policy(),
798            ))
799            .unwrap();
800        let id = PluginId::new("object-storage").unwrap();
801        let mut selection = PluginSelection::default();
802        selection.enabled.insert(id);
803        let application = manager.compose(&selection).unwrap();
804        assert!(
805            application
806                .graph
807                .capabilities
808                .contains_key("storage.object.upload")
809        );
810        assert!(application.services.get::<ObjectUploadService>().is_ok());
811        assert!(application.services.get::<ObjectMetadataService>().is_ok());
812    }
813}
minco_plugin_object_storage/uploads.rs

minco_plugin_object_storage/
uploads.rs
