coven_storage/
s3_detection.rs

1//! What S3 detection observed, the settings derived from it, and the setup
2//! safety check of effective settings against it (§20.8). The probes themselves
3//! run in the adapter; this is the pure decision over their results.
4use crate::{
5    S3BucketVersioning, S3Checksums, S3CopyCondition, S3CopyPublication, S3Listing,
6    S3MultipartTimestamp, S3Profile, S3ResponseHeaders, S3Revision, S3Settings, S3Signing,
7    S3SinglePut, StorageCheck, StorageCheckError, StorageConfig, StorageError, StorageFailure,
8};
9use std::sync::Arc;
10
11/// What detection observed at one location (§20.7).
12#[derive(Clone, Debug)]
13pub struct S3Detection {
14    /// Bucket versioning, or the native refusal that left it unknown.
15    pub versioning: Result<S3BucketVersioning, Arc<StorageError>>,
16    /// Settings derived from the results; prefill advanced settings with them.
17    pub settings: S3Settings,
18    /// One observation per probe, in the order run.
19    pub results: Vec<S3ProbeResult>,
20}
21
22/// The detection runs one detection cleans up after (§20.7): its own,
23/// which the caller recorded before detection writes its first probe, and
24/// the earlier runs at the same bucket whose probes and uploads no cleanup
25/// has confirmed gone, such as one a crash cut.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub struct S3DetectionRuns {
28    /// This detection's run: its probes go under `.coven-check/<run>/`.
29    pub run: uuid::Uuid,
30    /// The earlier runs whose leftovers it removes after its own.
31    pub earlier: Vec<uuid::Uuid>,
32}
33
34/// What one detection did (§20.7).
35#[derive(Debug)]
36pub struct S3DetectionRun {
37    /// What it observed, or why it failed; a failed cleanup fails it.
38    pub detection: Result<S3Detection, StorageError>,
39    /// The runs whose probes and uploads it confirmed gone: its own when it
40    /// wrote nothing or its cleanup succeeded, and each earlier run whose
41    /// leftovers it removed.
42    pub cleaned: Vec<uuid::Uuid>,
43}
44
45impl S3DetectionRun {
46    /// A detection that wrote nothing: its own run left nothing, and it
47    /// removed no earlier run's leftovers.
48    pub fn unwritten(runs: &S3DetectionRuns, detection: Result<S3Detection, StorageError>) -> Self {
49        Self { detection, cleaned: vec![runs.run] }
50    }
51}
52
53/// One probe's outcome.
54#[derive(Clone, Debug, PartialEq, Eq)]
55pub struct S3ProbeResult {
56    /// The behavior probed.
57    pub probe: S3Probe,
58    /// Whether the behavior held, as described on each probe.
59    pub held: bool,
60    /// Native outcome; names no key, endpoint or object.
61    pub detail: String,
62}
63
64/// A behavior detection probes.
65#[derive(Clone, Copy, Debug, PartialEq, Eq)]
66pub enum S3Probe {
67    /// A read-only listing is accepted with this signer.
68    Signing(S3Signing),
69    /// The condition refuses an occupied name, leaving its bytes and time.
70    SinglePut(S3SinglePut),
71    /// An unfinished multipart upload is absent from HEAD, GET and listing.
72    PartialUploadHidden,
73    /// Conditional completion refuses an occupied name, leaving bytes, ETag and time.
74    ConditionalCompletion,
75    /// Native multipart `Last-Modified` dates completion; otherwise it dates start.
76    CompletionTimestamp,
77    /// A copy with this condition refuses an occupied name, leaving bytes and time.
78    CopyPublication(S3CopyCondition),
79    /// A copy is dated later than its source.
80    CopyTimestamp,
81    /// This protocol lists every probe object; V2 must also agree with V1.
82    Listing(S3Listing),
83    /// HEAD with checksum mode is accepted.
84    ChecksumMode,
85    /// A checksum arrives without checksum mode.
86    PassiveChecksum,
87    /// Some probe returned a checksum.
88    ChecksumReturned,
89    /// Every returned checksum stated its scope.
90    ChecksumScopeStated,
91    /// GCS's `x-goog-generation` arrived.
92    GoogleGeneration,
93    /// GCS's `x-goog-hash` or generation arrived.
94    GoogleHeaders,
95    /// A zero-entry listing at a missing object's key is accepted; status
96    /// uses it to tell a missing bucket from a missing object.
97    BucketList,
98}
99
100/// Detection's preference orders (Appendix E.7): the first enforced condition is chosen.
101pub(crate) const SIGNERS: [S3Signing; 2] = [S3Signing::GoogleHmac, S3Signing::AwsSigV4];
102pub(crate) const SINGLE_PUTS: [S3SinglePut; 2] =
103    [S3SinglePut::IfNoneMatch, S3SinglePut::GenerationMatch];
104pub(crate) const COPY_CONDITIONS: [S3CopyCondition; 3] = [
105    S3CopyCondition::GenerationMatch,
106    S3CopyCondition::IfNoneMatch,
107    S3CopyCondition::CloudflareIfNoneMatch,
108];
109
110impl S3Detection {
111    /// Derive each field by Appendix E.7. Only signer and single-PUT selection
112    /// retain their starting values when neither candidate holds; other fields
113    /// use the listed fallbacks. [`Self::check`] refuses unsupported choices.
114    pub(crate) fn derive(
115        results: Vec<S3ProbeResult>,
116        plain: S3Settings,
117        versioning: Result<S3BucketVersioning, Arc<StorageError>>,
118    ) -> Self {
119        let detection = Self { versioning, settings: plain, results };
120        let held = |probe| detection.held(probe);
121        let first = |candidates: &[S3Probe]| candidates.iter().copied().find(|probe| held(*probe));
122        let mut settings = plain;
123        if let Some(S3Probe::Signing(signing)) = first(&SIGNERS.map(S3Probe::Signing)) {
124            settings.signing = signing;
125        }
126        if let Some(S3Probe::SinglePut(condition)) = first(&SINGLE_PUTS.map(S3Probe::SinglePut)) {
127            settings.single_put = condition;
128        }
129        settings.multipart_timestamp = if held(S3Probe::CompletionTimestamp) {
130            S3MultipartTimestamp::Completion
131        } else {
132            S3MultipartTimestamp::Start
133        };
134        // Sequential probes cannot show that racing PUTs and copies need
135        // every object published by copy, so detection never chooses `Every`.
136        settings.copy_publication = if held(S3Probe::ConditionalCompletion)
137            && held(S3Probe::PartialUploadHidden)
138            && held(S3Probe::CompletionTimestamp)
139        {
140            S3CopyPublication::Never
141        } else if let Some(S3Probe::CopyPublication(condition)) =
142            first(&COPY_CONDITIONS.map(S3Probe::CopyPublication))
143        {
144            S3CopyPublication::Multipart(condition)
145        } else {
146            S3CopyPublication::Never
147        };
148        settings.listing =
149            if held(S3Probe::Listing(S3Listing::V2)) { S3Listing::V2 } else { S3Listing::V1 };
150        settings.checksums = if !held(S3Probe::ChecksumReturned) {
151            S3Checksums::Absent
152        } else if !held(S3Probe::ChecksumScopeStated) {
153            S3Checksums::UnstatedIgnored
154        } else {
155            S3Checksums::ScopeStated
156        };
157        // Request the checksum extension only when the response needs it.
158        settings.request_checksums = held(S3Probe::ChecksumMode)
159            && !held(S3Probe::PassiveChecksum)
160            && settings.checksums != S3Checksums::Absent;
161        settings.revision =
162            if held(S3Probe::GoogleGeneration) { S3Revision::Generation } else { S3Revision::ETag };
163        settings.response_headers = if held(S3Probe::GoogleHeaders) {
164            S3ResponseHeaders::Google
165        } else {
166            S3ResponseHeaders::S3
167        };
168        Self { settings, ..detection }
169    }
170
171    /// Whether this probe's behavior held; an unrun probe did not.
172    pub fn held(&self, probe: S3Probe) -> bool {
173        self.results.iter().any(|observation| observation.probe == probe && observation.held)
174    }
175
176    /// Refuses settings that contradict the results, naming the first
177    /// failed check as setup would: CreateOnce, MultipartCreateOnce,
178    /// PublicationTime, then Behavior.
179    pub fn check(&self, settings: &S3Settings) -> Result<(), StorageCheckError> {
180        let fail = |check, reason: &'static str| {
181            Err(StorageCheckError::ProviderCheck {
182                check,
183                source: StorageFailure::Protocol.with_source(reason),
184            })
185        };
186        let held = |probe| self.held(probe);
187        if !held(S3Probe::SinglePut(settings.single_put)) {
188            return fail(
189                StorageCheck::CreateOnce,
190                "the single-PUT condition did not refuse an occupied name with its bytes and time unchanged",
191            );
192        }
193        let (create_once, dated) = match settings.copy_publication.multipart() {
194            None => (
195                held(S3Probe::ConditionalCompletion) && held(S3Probe::PartialUploadHidden),
196                held(S3Probe::CompletionTimestamp),
197            ),
198            Some(condition) => {
199                (held(S3Probe::CopyPublication(condition)), held(S3Probe::CopyTimestamp))
200            }
201        };
202        if !create_once {
203            return fail(
204                StorageCheck::MultipartCreateOnce,
205                "multipart publication did not refuse an occupied name with its bytes and time unchanged, or showed an unfinished upload",
206            );
207        }
208        if !dated {
209            return fail(
210                StorageCheck::PublicationTime,
211                "multipart publication is not dated by complete publication",
212            );
213        }
214        let observed = if held(S3Probe::CompletionTimestamp) {
215            S3MultipartTimestamp::Completion
216        } else {
217            S3MultipartTimestamp::Start
218        };
219        if settings.multipart_timestamp != observed {
220            return fail(
221                StorageCheck::PublicationTime,
222                "the multipart timestamp setting contradicts the observed timestamp",
223            );
224        }
225        let behavior = [
226            (held(S3Probe::Signing(settings.signing)), "the service refused this signer"),
227            (
228                held(S3Probe::Listing(settings.listing)),
229                "this listing protocol failed or disagreed with V1",
230            ),
231            (
232                match settings.checksums {
233                    S3Checksums::Absent => !held(S3Probe::ChecksumReturned),
234                    S3Checksums::ScopeStated => held(S3Probe::ChecksumScopeStated),
235                    S3Checksums::UnstatedIgnored => true,
236                },
237                "the checksum policy contradicts the returned checksums",
238            ),
239            (
240                !settings.request_checksums || held(S3Probe::ChecksumMode),
241                "the service refused checksum mode",
242            ),
243            (
244                settings.revision == S3Revision::ETag || held(S3Probe::GoogleGeneration),
245                "generation revisions need GCS's generation header",
246            ),
247            (
248                (settings.response_headers == S3ResponseHeaders::Google)
249                    == held(S3Probe::GoogleHeaders),
250                "the response-header setting contradicts the headers returned",
251            ),
252            (held(S3Probe::BucketList), "the service refused the zero-entry bucket listing"),
253        ];
254        match behavior.into_iter().find(|(agrees, _)| !agrees) {
255            Some((_, reason)) => fail(StorageCheck::Behavior, reason),
256            None => Ok(()),
257        }
258    }
259
260    /// The location setup records, after checking versioning support and its effective settings:
261    /// an uncatalogued endpoint naming no profile gets these detected
262    /// settings as its custom profile; any other location is kept as given.
263    pub fn resolve(&self, location: StorageConfig) -> Result<StorageConfig, StorageCheckError> {
264        let StorageConfig::S3 { bucket, region, endpoint, profile } = location else {
265            return Err(StorageFailure::InvalidConfiguration
266                .with_source("expected S3 configuration")
267                .into());
268        };
269        let (profile, settings) =
270            match crate::providers::s3_settings(endpoint.as_ref(), profile.as_ref())? {
271                Some(settings) => (profile, settings),
272                None => (Some(S3Profile::Custom(self.settings)), self.settings),
273            };
274        crate::providers::check_s3_versioning(
275            endpoint.as_ref(),
276            profile.as_ref(),
277            &self.versioning,
278        )?;
279        self.check(&settings)?;
280        Ok(StorageConfig::S3 { bucket, region, endpoint, profile })
281    }
282}
283
284#[cfg(test)]
285#[path = "s3_detection_tests.rs"]
286mod tests;
coven_storage/s3_detection.rs

coven_storage/
s3_detection.rs
