Strata: S3, Google Cloud Storage and Azure Blob Storage stores 8d80356371c5

[package]

name = "oxy-strata"

version.workspace = true

edition.workspace = true

license.workspace = true

description = "Oxygen Strata: repositories kept on S3, Google Cloud Storage and Azure Blob Storage"

[dependencies]

oxy-core = { path = "../oxy-core" }

# The cloud stores, signing with ring (the crypto stack oxy already

# links), not aws-lc-rs, which needs a C toolchain and cmake to build.

object_store = { version = "0.14", default-features = false, features = ["aws-base", "gcp-base", "azure-base", "reqwest", "ring"] }

# object_store's reqwest needs a TLS stack: rustls, with its provider

# installed at run time (`rustls::crypto::ring`). A dependency's feature

# list cannot name another crate's feature, so reqwest is named here.

reqwest = { version = "0.13", default-features = false, features = ["rustls-no-provider"] }

rustls = { version = "0.23", default-features = false, features = ["ring", "std"] }

tokio = { version = "1", default-features = false, features = ["rt"] }

futures = "0.3"

serde_json = "1"

[dev-dependencies]

tempfile = "3"

//! Amazon S3: `s3://BUCKET/PREFIX`.

//!

//! Credentials, in the order the AWS tools use them:

//! 1. `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` (and

//!    `AWS_SESSION_TOKEN`) in the environment, unless the URL names a

//!    profile;

//! 2. the profile (`?profile=NAME`, else `AWS_PROFILE`, else `default`):

//!    its keys from `~/.aws/credentials`;

//! 3. what the AWS CLI can produce for that profile

//!    (`aws configure export-credentials`): SSO, login sessions, assumed

//!    roles and credential processes, so anything that works for `aws`

//!    works for `oxy`;

//! 4. the instance's or container's role, through the provider's own

//!    lookup.

//!

//! The region comes from `?region=`, the environment, the profile, or is

//! asked of S3 itself; the endpoint (`?endpoint=`, for MinIO and other

//! S3-alikes) from the URL, the environment or the profile.

use crate::ini;

use crate::StoreUrl;

use object_store::aws::{AmazonS3, AmazonS3Builder, S3ConditionalPut};

use object_store::ClientOptions;

use oxy_core::strata::{Error, Result};

use std::path::PathBuf;

pub const HINT: &str = "set AWS_PROFILE (or add ?profile=NAME to the remote URL) to a profile in ~/.aws/credentials, run `aws sso login` or `aws login` for it, or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY";

struct Found {

    key_id: String,

    secret: String,

    token: Option<String>,

}

fn env(name: &str) -> Option<String> {

    std::env::var(name).ok().filter(|v| !v.is_empty())

}

fn home() -> Option<PathBuf> {

    env("HOME")

        .or_else(|| env("USERPROFILE"))

        .map(PathBuf::from)

}

fn read_ini(env_override: &str, default_rel: &str) -> ini::Sections {

    let path = env(env_override)

        .map(PathBuf::from)

        .or_else(|| home().map(|h| h.join(default_rel)));

    path.and_then(|p| std::fs::read_to_string(p).ok())

        .map(|t| ini::parse(&t))

        .unwrap_or_default()

}

/// A profile's section in `~/.aws/config`: `[profile NAME]`, or

/// `[default]`; `~/.aws/credentials` uses `[NAME]` for both.

fn config_section<'a>(

    config: &'a ini::Sections,

    profile: &str,

) -> Option<&'a std::collections::BTreeMap<String, String>> {

    if profile == "default" {

        config.get("default")

    } else {

        config

            .get(&format!("profile {profile}"))

            .or_else(|| config.get(profile))

    }

}

/// Ask the AWS CLI for the profile's credentials, if it is installed and

/// knows the profile.

fn from_cli(profile: &str) -> Option<Found> {

    let out = std::process::Command::new("aws")

        .args([

            "configure",

            "export-credentials",

            "--profile",

            profile,

            "--format",

            "process",

        ])

        .stdin(std::process::Stdio::null())

        .stderr(std::process::Stdio::null())

        .output()

        .ok()?;

    if !out.status.success() {

        return None;

    }

    let v: serde_json::Value = serde_json::from_slice(&out.stdout).ok()?;

    Some(Found {

        key_id: v.get("AccessKeyId")?.as_str()?.to_string(),

        secret: v.get("SecretAccessKey")?.as_str()?.to_string(),

        token: v

            .get("SessionToken")

            .and_then(|t| t.as_str())

            .map(str::to_string),

    })

}

pub fn open(url: &StoreUrl, runtime: &tokio::runtime::Runtime) -> Result<AmazonS3> {

    let bucket = url.host.clone();

    let explicit_profile = url.param("profile").map(str::to_string);

    let profile = explicit_profile

        .clone()

        .or_else(|| env("AWS_PROFILE"))

        .unwrap_or_else(|| "default".into());

    let config = read_ini("AWS_CONFIG_FILE", ".aws/config");

    let creds = read_ini("AWS_SHARED_CREDENTIALS_FILE", ".aws/credentials");

    let section = config_section(&config, &profile);

    let known_profile = section.is_some() || creds.contains_key(&profile);

    // from_env picks up the environment's keys, region, endpoint, and the

    // container and instance roles; what follows fills in or overrides.

    let mut b = AmazonS3Builder::from_env()

        .with_bucket_name(&bucket)

        .with_conditional_put(S3ConditionalPut::ETagMatch);

    let env_keys = match (env("AWS_ACCESS_KEY_ID"), env("AWS_SECRET_ACCESS_KEY")) {

        (Some(k), Some(s)) if explicit_profile.is_none() => Some(Found {

            key_id: k,

            secret: s,

            token: env("AWS_SESSION_TOKEN"),

        }),

        _ => None,

    };

    let found = env_keys

        .or_else(|| {

            let c = creds.get(&profile)?;

            Some(Found {

                key_id: c.get("aws_access_key_id")?.clone(),

                secret: c.get("aws_secret_access_key")?.clone(),

                token: c.get("aws_session_token").cloned(),

            })

        })

        .or_else(|| {

            if known_profile {

                from_cli(&profile)

            } else {

                None

            }

        });

    match found {

        Some(f) => {

            b = b

                .with_access_key_id(f.key_id)

                .with_secret_access_key(f.secret);

            if let Some(t) = f.token {

                b = b.with_token(t);

            }

        }

        None if explicit_profile.is_some() && !known_profile => {

            return Err(Error::Backend(format!(

                "{}: profile {profile:?} is not in ~/.aws/config or ~/.aws/credentials: {HINT}",

                url.plain()

            )));

        }

        None => {}

    }

    let endpoint = url

        .param("endpoint")

        .map(str::to_string)

        .or_else(|| env("AWS_ENDPOINT_URL_S3"))

        .or_else(|| env("AWS_ENDPOINT_URL"))

        .or_else(|| env("AWS_ENDPOINT"))

        .or_else(|| section.and_then(|s| s.get("endpoint_url").cloned()));

    if let Some(e) = &endpoint {

        b = b

            .with_endpoint(e.clone())

            .with_allow_http(e.starts_with("http://"));

        // S3-alikes serve buckets by path, not by host name.

        b = b.with_virtual_hosted_style_request(false);

    }

    let region = url

        .param("region")

        .map(str::to_string)

        .or_else(|| env("AWS_REGION"))

        .or_else(|| env("AWS_DEFAULT_REGION"))

        .or_else(|| section.and_then(|s| s.get("region").cloned()));

    let region = match region {

        Some(r) => r,

        None if endpoint.is_some() => "us-east-1".into(),

        None => runtime

            .block_on(object_store::aws::resolve_bucket_region(&bucket, &ClientOptions::new()))

            .map_err(|e| {

                Error::Backend(format!(

                    "{}: no region was given and S3 did not say which region holds the bucket ({e}): add ?region=NAME to the remote URL, set AWS_REGION, or put region in the profile",

                    url.plain()

                ))

            })?,

    };

    b = b.with_region(region);

    b.build()

        .map_err(|e| Error::Backend(format!("{}: {e}: {HINT}", url.plain())))

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn profile_sections_are_found_under_either_heading() {

        let config =

            ini::parse("[default]\nregion = a\n[profile work]\nregion = b\n[plain]\nregion = c\n");

        assert_eq!(config_section(&config, "default").unwrap()["region"], "a");

        assert_eq!(config_section(&config, "work").unwrap()["region"], "b");

        assert_eq!(config_section(&config, "plain").unwrap()["region"], "c");

        assert!(config_section(&config, "none").is_none());

    }

}

//! Azure Blob Storage: `az://ACCOUNT/CONTAINER/PREFIX`.

//!

//! Credentials: a service principal (`AZURE_CLIENT_ID`,

//! `AZURE_CLIENT_SECRET`, `AZURE_TENANT_ID`, as the Azure SDKs name them,

//! or the `AZURE_STORAGE_*` spellings), an account key

//! (`AZURE_STORAGE_ACCOUNT_KEY`), a SAS token (`AZURE_STORAGE_SAS_TOKEN`),

//! else what `az login` left behind, else the machine's managed

//! identity. `?endpoint=URL` points at an emulator.

use crate::StoreUrl;

use object_store::azure::{MicrosoftAzure, MicrosoftAzureBuilder};

use oxy_core::strata::{Error, Result};

pub const HINT: &str = "run `az login`, or set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET and AZURE_TENANT_ID for a service principal, or AZURE_STORAGE_ACCOUNT_KEY";

fn env(name: &str) -> Option<String> {

    std::env::var(name).ok().filter(|v| !v.is_empty())

}

/// The store and the prefix under the container.

pub fn open(url: &StoreUrl) -> Result<(MicrosoftAzure, String)> {

    let (container, prefix) = match url.path.split_once('/') {

        Some((c, p)) => (c.to_string(), p.to_string()),

        None if !url.path.is_empty() => (url.path.clone(), String::new()),

        None => {

            return Err(Error::Backend(format!(

                "{}: an Azure store URL names the account and the container: az://ACCOUNT/CONTAINER/PREFIX",

                url.plain()

            )))

        }

    };

    let mut b = MicrosoftAzureBuilder::from_env()

        .with_account(&url.host)

        .with_container_name(&container);

    let principal = (

        env("AZURE_CLIENT_ID").or_else(|| env("AZURE_STORAGE_CLIENT_ID")),

        env("AZURE_CLIENT_SECRET").or_else(|| env("AZURE_STORAGE_CLIENT_SECRET")),

        env("AZURE_TENANT_ID").or_else(|| env("AZURE_STORAGE_TENANT_ID")),

    );

    let mut explicit = false;

    if let (Some(id), Some(secret), Some(tenant)) = principal {

        b = b.with_client_secret_authorization(id, secret, tenant);

        explicit = true;

    }

    if env("AZURE_STORAGE_ACCOUNT_KEY").is_some()

        || env("AZURE_STORAGE_ACCESS_KEY").is_some()

        || env("AZURE_STORAGE_SAS_TOKEN").is_some()

        || env("AZURE_STORAGE_SAS_KEY").is_some()

    {

        explicit = true;

    }

    if let Some(endpoint) = url.param("endpoint") {

        b = b

            .with_endpoint(endpoint.to_string())

            .with_allow_http(endpoint.starts_with("http://"));

    }

    if !explicit {

        // What `az login` left behind; the managed identity comes after.

        b = b.with_use_azure_cli(true);

    }

    let store = b

        .build()

        .map_err(|e| Error::Backend(format!("{}: {e}: {HINT}", url.plain())))?;

    Ok((store, prefix))

}

//! Google Cloud Storage: `gs://BUCKET/PREFIX`.

//!

//! Credentials, in Google's own order: a service account key file named

//! by `?credentials=PATH` or `GOOGLE_APPLICATION_CREDENTIALS` (which may

//! also hold application default credentials), the key or path in

//! `GOOGLE_SERVICE_ACCOUNT*`, then the application default credentials

//! `gcloud auth application-default login` wrote, then the instance's

//! service account. `?endpoint=URL` points at an emulator, and

//! `?anonymous=1` skips signing for one.

use crate::StoreUrl;

use object_store::gcp::{GoogleCloudStorage, GoogleCloudStorageBuilder};

use oxy_core::strata::{Error, Result};

pub const HINT: &str = "run `gcloud auth application-default login`, or set GOOGLE_APPLICATION_CREDENTIALS (or add ?credentials=PATH to the remote URL) to a service account key file";

/// Whether a credentials file is a service account key, as opposed to

/// application default credentials.

fn is_service_account_key(path: &str) -> bool {

    std::fs::read_to_string(path)

        .ok()

        .and_then(|t| serde_json::from_str::<serde_json::Value>(&t).ok())

        .and_then(|v| {

            v.get("type")

                .and_then(|t| t.as_str())

                .map(|t| t == "service_account")

        })

        .unwrap_or(false)

}

pub fn open(url: &StoreUrl) -> Result<GoogleCloudStorage> {

    let mut b = GoogleCloudStorageBuilder::from_env().with_bucket_name(&url.host);

    let file = url.param("credentials").map(str::to_string).or_else(|| {

        std::env::var("GOOGLE_APPLICATION_CREDENTIALS")

            .ok()

            .filter(|v| !v.is_empty())

    });

    if let Some(path) = file {

        if !std::path::Path::new(&path).is_file() {

            return Err(Error::Backend(format!(

                "{}: the credentials file {path} does not exist: {HINT}",

                url.plain()

            )));

        }

        b = if is_service_account_key(&path) {

            b.with_service_account_path(path)

        } else {

            b.with_application_credentials(path)

        };

    }

    if let Some(endpoint) = url.param("endpoint") {

        b = b.with_base_url(endpoint);

    }

    if url.param("anonymous").is_some_and(|v| v != "0") {

        // An emulator checks no token, but the client would still ask the

        // metadata server for one: give it a token to send.

        b = b.with_bearer_token("anonymous").with_skip_signature(true);

    }

    b.build()

        .map_err(|e| Error::Backend(format!("{}: {e}: {HINT}", url.plain())))

}

//! The INI files the AWS tools keep (`~/.aws/credentials`, `~/.aws/config`):

//! `[section]` headings and `key = value` lines, nothing more.

use std::collections::BTreeMap;

pub type Sections = BTreeMap<String, BTreeMap<String, String>>;

pub fn parse(text: &str) -> Sections {

    let mut out = Sections::new();

    let mut current: Option<String> = None;

    for raw in text.lines() {

        let line = raw.trim();

        if line.is_empty() || line.starts_with('#') || line.starts_with(';') {

            continue;

        }

        if let Some(name) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) {

            current = Some(name.trim().to_string());

            out.entry(name.trim().to_string()).or_default();

            continue;

        }

        if let (Some(section), Some((k, v))) = (&current, line.split_once('=')) {

            out.entry(section.clone())

                .or_default()

                .insert(k.trim().to_string(), v.trim().to_string());

        }

    }

    out

}

#[cfg(test)]

mod tests {

    use super::parse;

    #[test]

    fn sections_and_keys() {

        let s = parse("[default]\nregion = eu-west-2\n\n# a comment\n[profile work]\nregion=us-east-1\nsso_start_url = https://x\n");

        assert_eq!(s["default"]["region"], "eu-west-2");

        assert_eq!(s["profile work"]["region"], "us-east-1");

        assert_eq!(s["profile work"]["sso_start_url"], "https://x");

        assert!(parse("").is_empty());

    }

}

//! Oxygen Strata: repositories kept on cloud object storage (docs/stores.md).

//!

//! This crate gives `oxy` the stores that need an HTTP client and the

//! providers' signing: Amazon S3 (`s3://BUCKET/PREFIX`), Google Cloud

//! Storage (`gs://BUCKET/PREFIX`) and Azure Blob Storage

//! (`az://ACCOUNT/CONTAINER/PREFIX`), each as a [`Store`] over the

//! `object_store` crate. Credentials come from each provider's standard

//! places, with nothing oxy-specific to set up; a URL may carry settings

//! as query parameters (`?profile=work&region=eu-west-2`), which is where

//! `oxy remote add --profile` keeps them.

//!

//! [`register`] hands these stores to oxy-core's remote opener; a program

//! that links this crate calls it once at start.

mod aws;

mod azure;

mod gcp;

mod ini;

use futures::TryStreamExt;

use object_store::path::Path as ObjectPath;

use object_store::{

    GetOptions, ObjectStore, ObjectStoreExt, PutMode, PutOptions, PutPayload, UpdateVersion,

};

use oxy_core::strata::{check_key, Entry, Error, Result, Store, Version};

use std::collections::BTreeMap;

use std::future::Future;

use std::ops::Range;

use std::sync::{Arc, Mutex};

/// Make the cloud stores available to `oxy_core::protocol::open_remote`.

/// Safe to call more than once.

pub fn register() {

    // object_store's reqwest is built with rustls and no bundled crypto

    // provider; oxy links ring already, so that is the one to install.

    let _ = rustls::crypto::ring::default_provider().install_default();

    oxy_core::protocol::register_store_opener(open_url);

}

fn open_url(

    url: &str,

    _options: &oxy_core::protocol::RemoteOptions,

) -> Option<Result<Box<dyn Store>>> {

    let scheme = url.split_once("://")?.0;

    if !matches!(scheme, "s3" | "gs" | "az") {

        return None;

    }

    Some(CloudStore::open(url).map(|s| Box::new(s) as Box<dyn Store>))

}

/// A parsed store URL: `SCHEME://HOST/PATH?PARAMS`.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct StoreUrl {

    pub scheme: String,

    /// The bucket (S3, GCS) or the storage account (Azure).

    pub host: String,

    /// The path under the host, without leading or trailing slashes; for

    /// Azure the container is its first segment.

    pub path: String,

    pub params: BTreeMap<String, String>,

}

impl StoreUrl {

    pub fn parse(url: &str) -> Result<StoreUrl> {

        let bad = |why: &str| Error::Backend(format!("{url}: {why}"));

        let (scheme, rest) = url.split_once("://").ok_or_else(|| bad("not a URL"))?;

        let (rest, query) = match rest.split_once('?') {

            Some((r, q)) => (r, Some(q)),

            None => (rest, None),

        };

        let (host, path) = match rest.split_once('/') {

            Some((h, p)) => (h, p.trim_matches('/')),

            None => (rest, ""),

        };

        if host.is_empty() {

            return Err(bad("names no bucket"));

        }

        let mut params = BTreeMap::new();

        for pair in query.unwrap_or("").split('&').filter(|p| !p.is_empty()) {

            let (k, v) = pair.split_once('=').unwrap_or((pair, ""));

            params.insert(k.to_string(), percent_decode(v));

        }

        Ok(StoreUrl {

            scheme: scheme.to_string(),

            host: host.to_string(),

            path: path.to_string(),

            params,

        })

    }

    /// The URL without its parameters, for messages.

    pub fn plain(&self) -> String {

        if self.path.is_empty() {

            format!("{}://{}", self.scheme, self.host)

        } else {

            format!("{}://{}/{}", self.scheme, self.host, self.path)

        }

    }

    pub fn param(&self, name: &str) -> Option<&str> {

        self.params.get(name).map(String::as_str)

    }

}

fn percent_decode(s: &str) -> String {

    let bytes = s.as_bytes();

    let mut out = Vec::with_capacity(bytes.len());

    let mut i = 0;

    while i < bytes.len() {

        if bytes[i] == b'%' && i + 2 < bytes.len() {

            if let Ok(v) = u8::from_str_radix(&s[i + 1..i + 3], 16) {

                out.push(v);

                i += 3;

                continue;

            }

        }

        out.push(bytes[i]);

        i += 1;

    }

    String::from_utf8_lossy(&out).into_owned()

}

/// A store on a cloud provider: the provider's client, the prefix every

/// key goes under, and a runtime to run the client on.

pub struct CloudStore {

    inner: Arc<dyn ObjectStore>,

    prefix: String,

    name: String,

    hint: &'static str,

    runtime: tokio::runtime::Runtime,

    /// One call at a time: the runtime runs on the calling thread.

    turn: Mutex<()>,

}

impl CloudStore {

    /// Open the store a URL names. Nothing is sent yet: credentials are

    /// resolved here, but the bucket is first spoken to on the first call.

    pub fn open(url: &str) -> Result<CloudStore> {

        let parsed = StoreUrl::parse(url)?;

        let runtime = tokio::runtime::Builder::new_current_thread()

            .enable_all()

            .build()

            .map_err(|e| Error::Backend(format!("starting the HTTP runtime: {e}")))?;

        let (inner, prefix, hint): (Arc<dyn ObjectStore>, String, &'static str) =

            match parsed.scheme.as_str() {

                "s3" => (

                    Arc::new(aws::open(&parsed, &runtime)?),

                    parsed.path.clone(),

                    aws::HINT,

                ),

                "gs" => (

                    Arc::new(gcp::open(&parsed)?),

                    parsed.path.clone(),

                    gcp::HINT,

                ),

                "az" => {

                    let (store, prefix) = azure::open(&parsed)?;

                    (Arc::new(store), prefix, azure::HINT)

                }

                other => {

                    return Err(Error::Backend(format!(

                        "{url}: {other}:// is not a store oxy knows (s3://, gs://, az://, dir://)"

                    )))

                }

            };

        Ok(CloudStore {

            inner,

            prefix,

            name: parsed.plain(),

            hint,

            runtime,

            turn: Mutex::new(()),

        })

    }

    fn block<F: Future>(&self, f: F) -> F::Output {

        let _turn = self.turn.lock().unwrap_or_else(|p| p.into_inner());

        self.runtime.block_on(f)

    }

    fn path(&self, key: &str) -> Result<ObjectPath> {

        check_key(key)?;

        let full = if self.prefix.is_empty() {

            key.to_string()

        } else {

            format!("{}/{key}", self.prefix)

        };

        ObjectPath::parse(&full).map_err(|e| Error::Backend(format!("{full}: {e}")))

    }

    /// Our error for the provider's, with what to do next where it helps.

    fn map(&self, e: object_store::Error, key: &str) -> Error {

        use object_store::Error as E;

        match e {

            E::NotFound { .. } => Error::NotFound(key.to_string()),

            E::AlreadyExists { .. } => Error::Exists(key.to_string()),

            E::Precondition { .. } | E::NotModified { .. } => Error::Conflict(key.to_string()),

            other => {

                let text = other.to_string();

                let lower = text.to_ascii_lowercase();

                let advice = if lower.contains("403")

                    || lower.contains("accessdenied")

                    || lower.contains("forbidden")

                    || lower.contains("authorizationpermissionmismatch")

                {

                    format!(

                        ": access denied; the credentials in use may not do this on {}: {}",

                        self.name, self.hint

                    )

                } else if lower.contains("credential") || lower.contains("no valid credentials") {

                    format!(": no usable credentials: {}", self.hint)

                } else if lower.contains("nosuchbucket")

                    || lower.contains("bucketnotfound")

                    || lower.contains("containernotfound")

                {

                    format!(

                        ": {} does not exist, or the credentials cannot see it",

                        self.name

                    )

                } else if lower.contains("dns")

                    || lower.contains("connect")

                    || lower.contains("timed out")

                {

                    ": cannot reach the service; check the network, the endpoint and the region"

                        .to_string()

                } else {

                    String::new()

                };

                Error::Backend(format!("{}: {text}{advice}", self.name))

            }

        }

    }

    fn version_of(e_tag: Option<String>, version: Option<String>) -> Version {

        Version(format!(

            "{}\t{}",

            e_tag.unwrap_or_default(),

            version.unwrap_or_default()

        ))

    }

    fn update_version(v: &Version) -> UpdateVersion {

        let (e_tag, version) = v.0.split_once('\t').unwrap_or((v.0.as_str(), ""));

        UpdateVersion {

            e_tag: (!e_tag.is_empty()).then(|| e_tag.to_string()),

            version: (!version.is_empty()).then(|| version.to_string()),

        }

    }

}

impl Store for CloudStore {

    fn describe(&self) -> String {

        self.name.clone()

    }

    fn has(&self, key: &str) -> Result<bool> {

        let path = self.path(key)?;

        match self.block(self.inner.head(&path)) {

            Ok(_) => Ok(true),

            Err(object_store::Error::NotFound { .. }) => Ok(false),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        self.block(async {

            let r = self.inner.get(&path).await?;

            r.bytes().await

        })

        .map(|b| b.to_vec())

        .map_err(|e| self.map(e, key))

    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        if range.start > range.end {

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} is not a range",

                range.start, range.end

            )));

        }

        if range.is_empty() {

            return Ok(Vec::new());

        }

        let want = (range.end - range.start) as usize;

        let bytes = self

            .block(self.inner.get_range(&path, range.clone()))

            .map_err(|e| self.map(e, key))?;

        if bytes.len() != want {

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} asked, {} returned: the object is shorter",

                range.start,

                range.end,

                bytes.len()

            )));

        }

        Ok(bytes.to_vec())

    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {

        let path = self.path(key)?;

        let opts = PutOptions {

            mode: PutMode::Create,

            ..Default::default()

        };

        self.block(

            self.inner

                .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),

        )

        .map(|_| ())

        .map_err(|e| self.map(e, key))

    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {

        // Providers list by whole path segments; the asked prefix may end

        // mid-segment, so list its directory and filter here.

        let dir = prefix.rsplit_once('/').map(|(d, _)| d).unwrap_or("");

        let full_dir = match (self.prefix.as_str(), dir) {

            ("", "") => None,

            ("", d) => Some(d.to_string()),

            (p, "") => Some(p.to_string()),

            (p, d) => Some(format!("{p}/{d}")),

        };

        let full_dir = full_dir

            .map(|d| ObjectPath::parse(&d).map_err(|e| Error::Backend(format!("{d}: {e}"))))

            .transpose()?;

        let metas: Vec<object_store::ObjectMeta> = self

            .block(self.inner.list(full_dir.as_ref()).try_collect())

            .map_err(|e| self.map(e, prefix))?;

        let strip = if self.prefix.is_empty() {

            String::new()

        } else {

            format!("{}/", self.prefix)

        };

        let mut out: Vec<Entry> = metas

            .into_iter()

            .filter_map(|m| {

                let loc = m.location.to_string();

                let key = loc.strip_prefix(&strip)?;

                key.starts_with(prefix).then(|| Entry {

                    key: key.to_string(),

                    size: m.size,

                })

            })

            .collect();

        out.sort_by(|a, b| a.key.cmp(&b.key));

        Ok(out)

    }

    fn delete(&self, key: &str) -> Result<()> {

        let path = self.path(key)?;

        match self.block(self.inner.delete(&path)) {

            Ok(()) | Err(object_store::Error::NotFound { .. }) => Ok(()),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {

        let path = self.path(key)?;

        let got = self.block(async {

            let r = self.inner.get_opts(&path, GetOptions::default()).await?;

            let meta = r.meta.clone();

            let bytes = r.bytes().await?;

            Ok::<_, object_store::Error>((bytes, meta))

        });

        match got {

            Ok((bytes, meta)) => Ok(Some((

                bytes.to_vec(),

                Self::version_of(meta.e_tag, meta.version),

            ))),

            Err(object_store::Error::NotFound { .. }) => Ok(None),

            Err(e) => Err(self.map(e, key)),

        }

    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {

        let path = self.path(key)?;

        let mode = match expected {

            None => PutMode::Create,

            Some(v) => PutMode::Update(Self::update_version(v)),

        };

        let opts = PutOptions {

            mode,

            ..Default::default()

        };

        let put = self

            .block(

                self.inner

                    .put_opts(&path, PutPayload::from(bytes.to_vec()), opts),

            )

            .map_err(|e| match self.map(e, key) {

                // A record that exists when none was expected is a lost

                // race too, not a misuse of put_new.

                Error::Exists(k) => Error::Conflict(k),

                other => other,

            })?;

        Ok(Self::version_of(put.e_tag, put.version))

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn urls_parse_with_their_parameters() {

        let u =

            StoreUrl::parse("s3://my-bucket/repos/oxygen?profile=work&region=eu-west-2").unwrap();

        assert_eq!(u.scheme, "s3");

        assert_eq!(u.host, "my-bucket");

        assert_eq!(u.path, "repos/oxygen");

        assert_eq!(u.param("profile"), Some("work"));

        assert_eq!(u.param("region"), Some("eu-west-2"));

        assert_eq!(u.plain(), "s3://my-bucket/repos/oxygen");

        let u = StoreUrl::parse("gs://bucket").unwrap();

        assert_eq!((u.host.as_str(), u.path.as_str()), ("bucket", ""));

        assert_eq!(u.plain(), "gs://bucket");

        let u = StoreUrl::parse("az://acct/container/a/b/?endpoint=http%3A%2F%2F127.0.0.1%3A10000")

            .unwrap();

        assert_eq!(u.path, "container/a/b");

        assert_eq!(u.param("endpoint"), Some("http://127.0.0.1:10000"));

        assert!(StoreUrl::parse("s3://").is_err());

        assert!(StoreUrl::parse("nonsense").is_err());

    }

    #[test]

    fn versions_carry_both_an_etag_and_a_generation() {

        let v = CloudStore::version_of(Some("\"abc\"".into()), Some("17".into()));

        let u = CloudStore::update_version(&v);

        assert_eq!(u.e_tag.as_deref(), Some("\"abc\""));

        assert_eq!(u.version.as_deref(), Some("17"));

        let u = CloudStore::update_version(&CloudStore::version_of(None, Some("3".into())));

        assert_eq!((u.e_tag, u.version.as_deref()), (None, Some("3")));

    }

}

//! The cloud stores against real services or emulators, when the

//! environment names them; otherwise each test says so and passes.

//!

//! - `OXY_TEST_S3=s3://BUCKET/PREFIX[?params]`, with `OXY_TEST_S3_PROFILE`

//!   added as `?profile=`;

//! - `OXY_TEST_GCS=gs://BUCKET/PREFIX[?params]`, with

//!   `OXY_TEST_GCS_CREDENTIALS` added as `?credentials=`;

//! - `OXY_TEST_AZURE=az://ACCOUNT/CONTAINER/PREFIX[?params]`, with the

//!   `AZURE_*` variables read from the file `OXY_TEST_AZURE_CREDENTIALS`.

//!

//! Every test works under its own prefix below the one given, and deletes

//! what it wrote.

use oxy_core::repo::Repo;

use oxy_core::strata::{conformance, Store};

use oxy_strata::CloudStore;

use std::path::Path;

fn configured(scheme: &str) -> Option<String> {

    let (var, extra, key) = match scheme {

        "s3" => ("OXY_TEST_S3", "OXY_TEST_S3_PROFILE", "profile"),

        "gs" => ("OXY_TEST_GCS", "OXY_TEST_GCS_CREDENTIALS", "credentials"),

        "az" => ("OXY_TEST_AZURE", "OXY_TEST_AZURE_CREDENTIALS", ""),

        _ => unreachable!(),

    };

    let url = std::env::var(var).ok().filter(|u| !u.is_empty())?;

    let extra_value = std::env::var(extra).ok().filter(|v| !v.is_empty());

    if scheme == "az" {

        if let Some(file) = &extra_value {

            for line in std::fs::read_to_string(file).unwrap_or_default().lines() {

                if let Some((k, v)) = line.split_once('=') {

                    std::env::set_var(k.trim(), v.trim());

                }

            }

        }

        return Some(url);

    }

    Some(match extra_value {

        Some(v) if !url.contains(&format!("{key}=")) => {

            let sep = if url.contains('?') { '&' } else { '?' };

            format!("{url}{sep}{key}={v}")

        }

        _ => url,

    })

}

fn each_store() -> Vec<(String, CloudStore)> {

    oxy_strata::register();

    ["s3", "gs", "az"]

        .into_iter()

        .filter_map(|s| {

            let Some(url) = configured(s) else {

                eprintln!("{s}: not configured, skipped");

                return None;

            };

            let store = CloudStore::open(&url).unwrap_or_else(|e| panic!("{url}: {e}"));

            Some((url, store))

        })

        .collect()

}

fn remove_all(store: &dyn Store, prefix: &str) {

    for e in store.list(prefix).unwrap() {

        store.delete(&e.key).unwrap();

    }

}

/// What earlier runs that failed may have left under the test prefix:

/// other processes' `probe-*` and `repo-*` keys.

fn remove_stale(store: &dyn Store) {

    let mine = format!("-{}", std::process::id());

    for e in store.list("").unwrap() {

        let first = e.key.split('/').next().unwrap_or("");

        if (first.starts_with("probe-") || first.starts_with("repo-")) && !first.ends_with(&mine) {

            store.delete(&e.key).unwrap();

        }

    }

}

#[test]

fn every_configured_store_conforms() {

    for (url, store) in each_store() {

        remove_stale(&store);

        let prefix = format!("probe-{}", std::process::id());

        let passed = conformance(&store, &prefix).unwrap_or_else(|e| {

            remove_all(&store, &prefix);

            panic!("{url}: {e}")

        });

        eprintln!("{url}: {} checks passed", passed.len());

        assert_eq!(passed.len(), 8);

    }

}

/// A small repository pushed to the store, cloned back, and pulled, by

/// URL: what a user does. Then every key the test wrote is removed.

#[test]

fn a_repository_round_trips_through_every_configured_store() {

    for (url, base_store) in each_store() {

        remove_stale(&base_store);

        let dir = tempfile::tempdir().unwrap();

        let root = dir.path();

        std::fs::create_dir_all(root.join("home")).unwrap();

        std::env::set_var("OXY_HOME", root.join("home"));

        let (base, query) = match url.split_once('?') {

            Some((b, q)) => (b.to_string(), format!("?{q}")),

            None => (url.clone(), String::new()),

        };

        let remote_url = format!("{base}/repo-{}{query}", std::process::id());

        let plain = format!("{base}/repo-{}", std::process::id());

        let store = CloudStore::open(&remote_url).unwrap();

        let mut a = Repo::init(&root.join("a")).unwrap();

        a.config.set("remote.cloud.url", &remote_url).unwrap();

        a.save_config().unwrap();

        write(root, "a/f.txt", b"hello from a\n");

        let h1 = a.commit("first", &[]).unwrap();

        let r = a.push("cloud", false).unwrap();

        assert!(r.iter().all(|x| x.ok), "{plain}: {r:?}");

        assert!(store.has("oxygen.toml").unwrap());

        assert_eq!(store.list("bundles/").unwrap().len(), 1, "{plain}");

        let b = Repo::clone(&remote_url, &root.join("b"))

            .unwrap_or_else(|e| panic!("{plain}: clone: {e}"));

        assert_eq!(

            std::fs::read(root.join("b/f.txt")).unwrap(),

            b"hello from a\n"

        );

        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);

        write(root, "b/f.txt", b"hello from a\nand b\n");

        b.commit("second", &[]).unwrap();

        assert!(b.push("origin", false).unwrap()[0].ok);

        let pr = a.pull("cloud").unwrap();

        assert!(pr.changed, "{plain}");

        assert_eq!(

            std::fs::read(root.join("a/f.txt")).unwrap(),

            b"hello from a\nand b\n"

        );

        let total: u64 = store.list("").unwrap().iter().map(|e| e.size).sum();

        eprintln!("{plain}: round trip ok, {total} bytes in the store before cleanup");

        remove_all(&store, "");

        assert!(store.list("").unwrap().is_empty(), "{plain}: cleanup");

    }

}

fn write(root: &Path, rel: &str, content: &[u8]) {

    let p = root.join(rel);

    std::fs::create_dir_all(p.parent().unwrap()).unwrap();

    std::fs::write(p, content).unwrap();

}

name = "android_system_properties"

version = "0.1.6"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc"

dependencies = [

 "libc",

]

[[package]]

]

[[package]]

name = "async-trait"

version = "0.1.92"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 3.0.6",

name = "chacha20"

version = "0.10.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"

dependencies = [

 "cfg-if",

 "cpufeatures",

 "rand_core 0.10.1",

]

[[package]]

name = "chrono"

version = "0.4.45"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"

dependencies = [

 "iana-time-zone",

 "num-traits",

 "serde",

 "windows-link",

]

[[package]]

name = "combine"

version = "4.6.8"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e"

dependencies = [

 "bytes",

 "memchr",

]

[[package]]

name = "core-foundation"

version = "0.10.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"

dependencies = [

 "core-foundation-sys",

 "libc",

]

[[package]]

name = "core-foundation-sys"

version = "0.8.7"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"

[[package]]

]

[[package]]

name = "crc-fast"

version = "1.10.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5"

dependencies = [

 "digest 0.10.7",

 "spin",

[[package]]

name = "crypto-common"

version = "0.1.7"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"

dependencies = [

 "generic-array",

 "typenum",

]

 "digest 0.11.3",

[[package]]

name = "digest"

version = "0.10.7"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"

dependencies = [

 "crypto-common 0.1.7",

]

 "crypto-common 0.2.2",

]

[[package]]

name = "futures"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"

dependencies = [

 "futures-channel",

 "futures-core",

 "futures-executor",

 "futures-io",

 "futures-sink",

 "futures-task",

 "futures-util",

 "futures-sink",

name = "futures-executor"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432"

dependencies = [

 "futures-core",

 "futures-task",

 "futures-util",

]

[[package]]

name = "futures-io"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"

[[package]]

name = "futures-macro"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 3.0.6",

]

[[package]]

 "futures-channel",

 "futures-io",

 "futures-macro",

 "futures-sink",

 "memchr",

name = "generic-array"

version = "0.14.7"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"

dependencies = [

 "typenum",

 "version_check",

]

[[package]]

[[package]]

name = "humantime"

version = "2.4.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15"

 "want",

name = "hyper-rustls"

version = "0.27.10"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "dfa8e654703247911e29c23fbeaa261834bd9bb74efba2f9acddc37bfb127f53"

dependencies = [

 "http",

 "hyper",

 "hyper-util",

 "rustls",

 "tokio",

 "tokio-rustls",

 "tower-service",

]

[[package]]

 "base64 0.23.1",

 "futures-channel",

 "futures-util",

 "httparse",