bucketwarden-server 0.1.0

BucketWarden storage server runtime.
Documentation
use super::*;

fn generated_counter(value: &str, prefix: char) -> Option<u64> {
    value.strip_prefix(prefix)?.parse::<u64>().ok()
}

impl BucketWarden {
    pub fn list_multipart_uploads(
        &mut self,
        principal: &str,
        bucket: &str,
        prefix: Option<&str>,
        key_marker: Option<&str>,
        upload_id_marker: Option<&str>,
        max_uploads: Option<usize>,
    ) -> Result<ListMultipartUploadsResult, RuntimeError> {
        self.authorize(principal, S3Action::ListMultipartUploads, bucket)?;
        self.require_bucket(bucket)?;
        let max_uploads = max_uploads.unwrap_or(1000);
        if upload_id_marker.is_some() && key_marker.is_none() {
            return Err(RuntimeError::InvalidListParameter {
                name: "upload-id-marker".to_string(),
                value: upload_id_marker.unwrap_or_default().to_string(),
            });
        }
        let prefix_value = prefix.unwrap_or_default();
        let mut uploads = self
            .multipart_uploads
            .iter()
            .filter(|(_, upload)| upload.bucket == bucket)
            .filter(|(_, upload)| upload.key.starts_with(prefix_value))
            .map(|(upload_id, upload)| ListedMultipartUpload {
                key: upload.key.clone(),
                upload_id: upload_id.clone(),
                initiated_epoch_seconds: upload.initiated_epoch_seconds,
            })
            .collect::<Vec<_>>();
        uploads.sort_by(|left, right| {
            left.key
                .cmp(&right.key)
                .then_with(|| match (
                    generated_counter(&left.upload_id, 'u'),
                    generated_counter(&right.upload_id, 'u'),
                ) {
                    (Some(left), Some(right)) => left.cmp(&right),
                    _ => left.upload_id.cmp(&right.upload_id),
                })
        });
        if let Some(key_marker) = key_marker {
            let upload_id_marker = upload_id_marker.unwrap_or_default();
            let mut resumed = upload_id_marker.is_empty();
            uploads = uploads
                .into_iter()
                .filter_map(|upload| match upload.key.as_str().cmp(key_marker) {
                    std::cmp::Ordering::Less => None,
                    std::cmp::Ordering::Greater => Some(upload),
                    std::cmp::Ordering::Equal if upload_id_marker.is_empty() => None,
                    std::cmp::Ordering::Equal if resumed => Some(upload),
                    std::cmp::Ordering::Equal if upload.upload_id == upload_id_marker => {
                        resumed = true;
                        None
                    }
                    std::cmp::Ordering::Equal => None,
                })
                .collect();
        }
        let is_truncated = uploads.len() > max_uploads;
        let (next_key_marker, next_upload_id_marker) = if is_truncated {
            let marker_index = if max_uploads == 0 { 0 } else { max_uploads - 1 };
            uploads
                .get(marker_index)
                .map(|upload| (Some(upload.key.clone()), Some(upload.upload_id.clone())))
                .unwrap_or((None, None))
        } else {
            (None, None)
        };
        uploads.truncate(max_uploads);
        self.audit_allowed(
            principal,
            S3Action::ListMultipartUploads,
            bucket,
            Some(uploads.len().to_string()),
        );
        Ok(ListMultipartUploadsResult {
            bucket: bucket.to_string(),
            prefix: prefix.map(str::to_string),
            key_marker: key_marker.map(str::to_string),
            upload_id_marker: upload_id_marker.map(str::to_string),
            max_uploads,
            is_truncated,
            next_key_marker,
            next_upload_id_marker,
            uploads,
        })
    }

    pub fn list_parts(
        &mut self,
        principal: &str,
        bucket: &str,
        key: &str,
        upload_id: &str,
        part_number_marker: Option<u16>,
        max_parts: Option<usize>,
    ) -> Result<ListPartsResult, RuntimeError> {
        let resource = object_resource(bucket, key);
        self.authorize(principal, S3Action::ListParts, &resource)?;
        let upload = self
            .multipart_uploads
            .get(upload_id)
            .ok_or_else(|| RuntimeError::NoSuchUpload(upload_id.to_string()))?;
        if upload.bucket != bucket || upload.key != key {
            return Err(RuntimeError::NoSuchUpload(upload_id.to_string()));
        }
        let max_parts = max_parts.unwrap_or(1000);
        let mut parts = upload
            .parts
            .iter()
            .filter(|(part_number, _)| part_number_marker.is_none_or(|marker| **part_number > marker))
            .map(|(part_number, part)| ListedPart {
                part_number: *part_number,
                etag: part.etag.clone(),
                size: part.body.len(),
            })
            .collect::<Vec<_>>();
        let is_truncated = parts.len() > max_parts;
        let next_part_number_marker = if is_truncated {
            let marker_index = if max_parts == 0 { 0 } else { max_parts - 1 };
            parts.get(marker_index).map(|part| part.part_number)
        } else {
            None
        };
        parts.truncate(max_parts);
        self.audit_allowed(
            principal,
            S3Action::ListParts,
            &resource,
            Some(parts.len().to_string()),
        );
        Ok(ListPartsResult {
            bucket: bucket.to_string(),
            key: key.to_string(),
            upload_id: upload_id.to_string(),
            part_number_marker,
            max_parts,
            is_truncated,
            next_part_number_marker,
            parts,
        })
    }
}