krabka_object_store/
worm.rs

1//! S3 control-plane checks required before enabling a WORM writer.
2
3use {
4    crate::{
5        GcsConfig, ObjectStoreError, S3Config,
6        build::{build_gcs_store, build_s3_store},
7        control_plane::{backend, read_ok_body, s3_bucket_url, s3_http_client, signed_s3_get_xml},
8    },
9    object_store::RetryConfig,
10    serde::{Deserialize, de::DeserializeOwned},
11};
12
13#[derive(Deserialize)]
14#[serde(rename_all = "PascalCase")]
15struct VersioningConfiguration {
16    status: Option<String>,
17}
18
19#[derive(Deserialize)]
20#[serde(rename_all = "PascalCase")]
21struct ObjectLockConfiguration {
22    object_lock_enabled: Option<String>,
23    rule: Option<ObjectLockRule>,
24}
25
26#[derive(Deserialize)]
27#[serde(rename_all = "PascalCase")]
28struct ObjectLockRule {
29    default_retention: Option<DefaultRetention>,
30}
31
32#[derive(Deserialize)]
33#[serde(rename_all = "PascalCase")]
34struct DefaultRetention {
35    mode: Option<String>,
36    days: Option<u64>,
37    years: Option<u64>,
38}
39
40#[derive(Deserialize)]
41#[serde(rename_all = "camelCase")]
42struct GcsBucketPolicy {
43    versioning: Option<GcsVersioning>,
44    retention_policy: Option<GcsRetentionPolicy>,
45}
46
47#[derive(Deserialize)]
48struct GcsVersioning {
49    enabled: Option<bool>,
50}
51
52#[derive(Deserialize)]
53#[serde(rename_all = "camelCase")]
54struct GcsRetentionPolicy {
55    retention_period: Option<String>,
56    is_locked: Option<bool>,
57}
58
59/// Confirms that an S3 bucket can protect multipart WORM objects.
60///
61/// # Errors
62///
63/// Names the failed control-plane request or the concrete missing policy.
64pub async fn verify_s3_worm_bucket(cfg: &S3Config) -> Result<(), ObjectStoreError> {
65    let versioning: VersioningConfiguration = get_bucket_xml(cfg, "versioning").await?;
66    let lock: ObjectLockConfiguration = get_bucket_xml(cfg, "object-lock").await?;
67    validate_s3_policy(cfg, &versioning, lock)
68}
69
70fn validate_s3_policy(
71    cfg: &S3Config,
72    versioning: &VersioningConfiguration,
73    lock: ObjectLockConfiguration,
74) -> Result<(), ObjectStoreError> {
75    if versioning.status.as_deref() != Some("Enabled") {
76        return Err(ObjectStoreError::InvalidConfig(format!(
77            "WORM bucket `{}` requires versioning Enabled, found {}",
78            cfg.bucket,
79            versioning.status.as_deref().unwrap_or("unset")
80        )));
81    }
82
83    if lock.object_lock_enabled.as_deref() != Some("Enabled") {
84        return Err(ObjectStoreError::InvalidConfig(format!(
85            "WORM bucket `{}` requires Object Lock Enabled",
86            cfg.bucket
87        )));
88    }
89    let retention = lock
90        .rule
91        .and_then(|rule| rule.default_retention)
92        .ok_or_else(|| {
93            ObjectStoreError::InvalidConfig(format!(
94                "WORM bucket `{}` has no Object Lock default retention",
95                cfg.bucket
96            ))
97        })?;
98    if retention.mode.as_deref() != Some("COMPLIANCE") {
99        return Err(ObjectStoreError::InvalidConfig(format!(
100            "WORM bucket `{}` default retention must use COMPLIANCE mode, found {}",
101            cfg.bucket,
102            retention.mode.as_deref().unwrap_or("unset")
103        )));
104    }
105    if retention.days.unwrap_or(0) == 0 && retention.years.unwrap_or(0) == 0 {
106        return Err(ObjectStoreError::InvalidConfig(format!(
107            "WORM bucket `{}` default retention has no positive Days or Years",
108            cfg.bucket
109        )));
110    }
111    Ok(())
112}
113
114/// Confirms that a GCS bucket can protect multipart WORM objects.
115///
116/// # Errors
117///
118/// Names the failed control-plane request or the concrete missing policy.
119pub async fn verify_gcs_worm_bucket(cfg: &GcsConfig) -> Result<(), ObjectStoreError> {
120    let store = build_gcs_store(cfg)?;
121    let credential = store.credentials().get_credential().await?;
122    let endpoint = cfg
123        .endpoint
124        .as_deref()
125        .unwrap_or("https://storage.googleapis.com");
126    let mut url = reqwest::Url::parse(endpoint)
127        .map_err(|error| ObjectStoreError::InvalidConfig(format!("GCS endpoint: {error}")))?;
128    url.path_segments_mut()
129        .map_err(|()| ObjectStoreError::InvalidConfig("GCS endpoint cannot be a base URL".into()))?
130        .pop_if_empty()
131        .extend(["storage", "v1", "b", &cfg.bucket]);
132    url.query_pairs_mut()
133        .append_pair("fields", "versioning,retentionPolicy");
134    let client = reqwest::Client::builder()
135        .timeout(cfg.request_timeout)
136        .connect_timeout(cfg.connect_timeout)
137        .build()
138        .map_err(backend)?;
139    let mut request = client.get(url);
140    if !credential.bearer.is_empty() {
141        request = request.bearer_auth(&credential.bearer);
142    }
143    let body = read_ok_body(request, "GCS Buckets.get", &RetryConfig {
144        max_retries: cfg.max_retries,
145        retry_timeout: cfg.retry_timeout,
146        ..RetryConfig::default()
147    })
148    .await?;
149    let policy: GcsBucketPolicy = serde_json::from_slice(&body)
150        .map_err(|error| ObjectStoreError::Backend(format!("GCS Buckets.get: {error}")))?;
151    validate_gcs_policy(cfg, policy)
152}
153
154fn validate_gcs_policy(cfg: &GcsConfig, policy: GcsBucketPolicy) -> Result<(), ObjectStoreError> {
155    if policy.versioning.and_then(|v| v.enabled) != Some(true) {
156        return Err(ObjectStoreError::InvalidConfig(format!(
157            "WORM bucket `{}` requires GCS versioning enabled",
158            cfg.bucket
159        )));
160    }
161    let retention = policy.retention_policy.ok_or_else(|| {
162        ObjectStoreError::InvalidConfig(format!(
163            "WORM bucket `{}` has no GCS retention policy",
164            cfg.bucket
165        ))
166    })?;
167    if retention.is_locked != Some(true) {
168        return Err(ObjectStoreError::InvalidConfig(format!(
169            "WORM bucket `{}` requires a locked GCS retention policy",
170            cfg.bucket
171        )));
172    }
173    let seconds = retention
174        .retention_period
175        .as_deref()
176        .and_then(|period| period.parse::<u64>().ok())
177        .unwrap_or(0);
178    if seconds == 0 {
179        return Err(ObjectStoreError::InvalidConfig(format!(
180            "WORM bucket `{}` GCS retention policy has no positive retention period",
181            cfg.bucket
182        )));
183    }
184    Ok(())
185}
186
187async fn get_bucket_xml<T: DeserializeOwned>(
188    cfg: &S3Config,
189    api: &str,
190) -> Result<T, ObjectStoreError> {
191    let operation = match api {
192        "versioning" => "GetBucketVersioning",
193        "object-lock" => "GetObjectLockConfiguration",
194        _ => "GetBucketConfiguration",
195    };
196    let store = build_s3_store(cfg)?;
197    let credential = store.credentials().get_credential().await?;
198    let mut url = s3_bucket_url(cfg)?;
199    url.query_pairs_mut().append_pair(api, "");
200    signed_s3_get_xml(&s3_http_client(cfg)?, &credential, cfg, url, operation).await
201}
202
203#[cfg(test)]
204mod tests {
205    use {
206        super::*,
207        crate::test_support::{
208            HttpListener, HttpPage, HttpServer, accept_http_request, http_listener, s3_config,
209            serve_http_pages,
210        },
211        assert2::check,
212        std::{future, time::Duration},
213        tokio::time,
214    };
215
216    const GCS_TEST_KEY: &str = r#"{
217        "private_key":"unused",
218        "private_key_id":"unused",
219        "client_email":"unused",
220        "disable_oauth":true
221    }"#;
222
223    #[derive(Clone, Copy)]
224    enum S3PolicyCase {
225        SuspendedVersioning,
226        MissingObjectLock,
227        MissingRule,
228        GovernanceRetention,
229        ZeroRetention,
230    }
231
232    fn versioning(status: Option<&str>) -> VersioningConfiguration {
233        VersioningConfiguration {
234            status: status.map(str::to_owned),
235        }
236    }
237
238    fn lock(
239        enabled: Option<&str>,
240        mode: Option<&str>,
241        days: Option<u64>,
242    ) -> ObjectLockConfiguration {
243        ObjectLockConfiguration {
244            object_lock_enabled: enabled.map(str::to_owned),
245            rule: Some(ObjectLockRule {
246                default_retention: Some(DefaultRetention {
247                    mode: mode.map(str::to_owned),
248                    days,
249                    years: None,
250                }),
251            }),
252        }
253    }
254
255    #[tokio::test]
256    async fn requires_versioning_and_positive_compliance_retention() {
257        let pages = [
258            (
259                "versioning",
260                "<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>",
261            ),
262            (
263                "object-lock",
264                "<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>COMPLIANCE</Mode><Days>30</Days></DefaultRetention></Rule></ObjectLockConfiguration>",
265            ),
266        ].into_iter().map(|(query, body)| HttpPage::new(format!("GET /bucket?{query}= HTTP/1.1"), "200 OK", body)).collect();
267        let HttpServer { endpoint, server } = serve_http_pages(pages, true).await;
268        let cfg = s3_config("bucket", endpoint);
269
270        verify_s3_worm_bucket(&cfg).await.unwrap();
271        server.await.unwrap();
272    }
273
274    #[yare::parameterized(
275        suspended_versioning = { S3PolicyCase::SuspendedVersioning },
276        missing_object_lock = { S3PolicyCase::MissingObjectLock },
277        missing_rule = { S3PolicyCase::MissingRule },
278        governance_retention = { S3PolicyCase::GovernanceRetention },
279        zero_retention = { S3PolicyCase::ZeroRetention },
280    )]
281    fn s3_policy_rejects_each_missing_protection(case: S3PolicyCase) {
282        let cfg = S3Config {
283            bucket: "bucket".into(),
284            ..Default::default()
285        };
286        let (versioning, lock, reason) = match case {
287            S3PolicyCase::SuspendedVersioning => (
288                versioning(Some("Suspended")),
289                lock(Some("Enabled"), Some("COMPLIANCE"), Some(1)),
290                "versioning Enabled",
291            ),
292            S3PolicyCase::MissingObjectLock => (
293                versioning(Some("Enabled")),
294                lock(None, Some("COMPLIANCE"), Some(1)),
295                "Object Lock Enabled",
296            ),
297            S3PolicyCase::MissingRule => (
298                versioning(Some("Enabled")),
299                ObjectLockConfiguration {
300                    object_lock_enabled: Some("Enabled".into()),
301                    rule: None,
302                },
303                "no Object Lock default retention",
304            ),
305            S3PolicyCase::GovernanceRetention => (
306                versioning(Some("Enabled")),
307                lock(Some("Enabled"), Some("GOVERNANCE"), Some(1)),
308                "COMPLIANCE",
309            ),
310            S3PolicyCase::ZeroRetention => (
311                versioning(Some("Enabled")),
312                lock(Some("Enabled"), Some("COMPLIANCE"), Some(0)),
313                "no positive Days or Years",
314            ),
315        };
316        let error = validate_s3_policy(&cfg, &versioning, lock).unwrap_err();
317        check!(error.to_string().contains(reason), "{reason}: {error}");
318    }
319
320    #[yare::parameterized(
321        versioning_disabled = {
322            r#"{"versioning":{"enabled":false},"retentionPolicy":{"retentionPeriod":"1","isLocked":true}}"#,
323            "versioning enabled",
324        },
325        missing_retention = { r#"{"versioning":{"enabled":true}}"#, "no GCS retention policy" },
326        unlocked_retention = {
327            r#"{"versioning":{"enabled":true},"retentionPolicy":{"retentionPeriod":"1","isLocked":false}}"#,
328            "locked GCS retention policy",
329        },
330        zero_retention = {
331            r#"{"versioning":{"enabled":true},"retentionPolicy":{"retentionPeriod":"0","isLocked":true}}"#,
332            "no positive retention period",
333        },
334    )]
335    fn gcs_policy_rejects_each_missing_protection(json: &str, reason: &str) {
336        let cfg = GcsConfig {
337            bucket: "bucket".into(),
338            ..Default::default()
339        };
340        let policy = serde_json::from_str(json).unwrap();
341        let error = validate_gcs_policy(&cfg, policy).unwrap_err();
342        check!(error.to_string().contains(reason), "{reason}: {error}");
343    }
344
345    /// Serves one canned `Buckets.get` body to one client, checking on the
346    /// way past that the request is the control-plane call this module claims
347    /// to make. Returns the endpoint to point a `GcsConfig` at, and the task
348    /// to join once the check has run.
349    async fn serve_bucket_policy(body: &'static str) -> HttpServer {
350        serve_http_pages(
351            vec![HttpPage::new(
352                "GET /storage/v1/b/bucket?fields=",
353                "200 OK",
354                body,
355            )],
356            false,
357        )
358        .await
359    }
360
361    fn gcs_cfg(endpoint: String) -> GcsConfig {
362        GcsConfig {
363            bucket: "bucket".into(),
364            endpoint: Some(endpoint),
365            service_account_key: Some(GCS_TEST_KEY.into()),
366            allow_http: true,
367            ..Default::default()
368        }
369    }
370
371    #[tokio::test]
372    async fn confirms_gcs_versioning_and_locked_retention() {
373        let HttpServer { endpoint, server } = serve_bucket_policy(
374            r#"{"versioning":{"enabled":true},"retentionPolicy":{"retentionPeriod":"86400","isLocked":true}}"#,
375        )
376        .await;
377
378        verify_gcs_worm_bucket(&gcs_cfg(endpoint)).await.unwrap();
379        server.await.unwrap();
380    }
381
382    /// The refusal half of the same control-plane call. `validate_gcs_policy`
383    /// covers the decision; this covers the whole `Buckets.get` round trip, so
384    /// a policy a real bucket reports as non-compliant fails archive startup
385    /// rather than being lost between the request and the check.
386    #[yare::parameterized(
387        versioning_disabled = {
388            r#"{"versioning":{"enabled":false},"retentionPolicy":{"retentionPeriod":"86400","isLocked":true}}"#,
389            "versioning enabled",
390        },
391        missing_retention = { r#"{"versioning":{"enabled":true}}"#, "no GCS retention policy" },
392        unlocked_retention = {
393            r#"{"versioning":{"enabled":true},"retentionPolicy":{"retentionPeriod":"86400","isLocked":false}}"#,
394            "locked GCS retention policy",
395        },
396    )]
397    #[test_macro(tokio::test)]
398    async fn refuses_a_gcs_bucket_whose_policy_is_not_worm(body: &str, reason: &str) {
399        let HttpServer { endpoint, server } = serve_bucket_policy(body).await;
400
401        let error = verify_gcs_worm_bucket(&gcs_cfg(endpoint))
402            .await
403            .unwrap_err();
404        check!(error.to_string().contains(reason), "{reason}: {error}");
405        server.await.unwrap();
406    }
407
408    #[yare::parameterized(
409        s3_request_deadline = { false, false },
410        gcs_request_deadline = { true, false },
411        s3_retry_deadline = { false, true },
412        gcs_retry_deadline = { true, true },
413    )]
414    #[test_macro(tokio::test)]
415    async fn bucket_checks_obey_request_and_retry_deadlines(gcs: bool, retry_deadline: bool) {
416        let HttpListener { listener, endpoint } = http_listener().await;
417        let server = tokio::spawn(async move {
418            let _socket = accept_http_request(listener).await;
419            future::pending::<()>().await;
420        });
421        let request_timeout = if retry_deadline {
422            Duration::from_secs(2)
423        } else {
424            Duration::from_millis(25)
425        };
426        let retry_timeout = if retry_deadline {
427            Duration::from_millis(25)
428        } else {
429            Duration::from_secs(2)
430        };
431        let result = time::timeout(Duration::from_secs(1), async {
432            if gcs {
433                let mut cfg = gcs_cfg(endpoint);
434                cfg.request_timeout = request_timeout;
435                cfg.retry_timeout = retry_timeout;
436                cfg.max_retries = 0;
437                verify_gcs_worm_bucket(&cfg).await
438            } else {
439                let mut cfg = s3_config("bucket", endpoint);
440                cfg.request_timeout = request_timeout;
441                cfg.retry_timeout = retry_timeout;
442                cfg.max_retries = 0;
443                verify_s3_worm_bucket(&cfg).await
444            }
445        })
446        .await;
447        server.abort();
448        let error = result
449            .expect("configured deadline must expire before the test guard")
450            .unwrap_err();
451        check!(matches!(error, ObjectStoreError::Backend(_)));
452        if retry_deadline {
453            check!(error.to_string().contains("retry deadline exceeded"));
454        }
455    }
456}
krabka_object_store/worm.rs

krabka_object_store/
worm.rs
