faucet_common_delta/
credentials.rs

1//! Object-store credentials for Delta tables.
2//!
3//! delta-rs authenticates through a `storage_options: HashMap<String,String>`
4//! map whose keys mirror the underlying `object_store` environment variables
5//! (`AWS_ACCESS_KEY_ID`, `AZURE_STORAGE_ACCOUNT_NAME`, `GOOGLE_SERVICE_ACCOUNT`,
6//! …). [`DeltaCredentials`] is a typed, self-documenting front for the common
7//! cases; anything it does not cover can still be set verbatim via the
8//! connection's `storage_options`. Explicit `storage_options` win on key
9//! collision (see [`DeltaCredentials::apply`]).
10
11use std::collections::HashMap;
12
13use schemars::JsonSchema;
14use serde::{Deserialize, Serialize};
15
16/// Object-store credentials for reading/writing a Delta table.
17///
18/// Uses the project-wide adjacently-tagged `{ type, config }` shape
19/// (snake_case discriminators): e.g. `credentials: { type: aws, config: { … } }`.
20/// The default is [`DeltaCredentials::Default`], which resolves credentials from
21/// the ambient environment / instance metadata via the object-store default
22/// provider chain — nothing is injected into `storage_options`.
23#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
24#[serde(tag = "type", content = "config", rename_all = "snake_case")]
25pub enum DeltaCredentials {
26    /// Resolve credentials from the ambient environment / default provider
27    /// chain (AWS instance profile, `AZURE_*` env, GCP ADC, …). Injects no
28    /// keys of its own.
29    #[default]
30    Default,
31    /// Static AWS credentials (or an S3-compatible endpoint such as MinIO).
32    Aws(AwsCredentials),
33    /// Azure Blob / ADLS Gen2 credentials.
34    Azure(AzureCredentials),
35    /// Google Cloud Storage credentials.
36    Gcp(GcpCredentials),
37}
38
39impl DeltaCredentials {
40    /// Merge the credential-derived keys into `options`, **without** overwriting
41    /// any key the user set explicitly. This makes explicit `storage_options`
42    /// authoritative — an escape hatch for keys this enum does not model.
43    pub fn apply(&self, options: &mut HashMap<String, String>) {
44        let derived = self.storage_options();
45        for (k, v) in derived {
46            options.entry(k).or_insert(v);
47        }
48    }
49
50    /// The `storage_options` entries implied by these credentials.
51    pub fn storage_options(&self) -> HashMap<String, String> {
52        let mut m = HashMap::new();
53        match self {
54            DeltaCredentials::Default => {}
55            DeltaCredentials::Aws(a) => {
56                if let Some(v) = &a.access_key_id {
57                    m.insert("AWS_ACCESS_KEY_ID".into(), v.clone());
58                }
59                if let Some(v) = &a.secret_access_key {
60                    m.insert("AWS_SECRET_ACCESS_KEY".into(), v.clone());
61                }
62                if let Some(v) = &a.session_token {
63                    m.insert("AWS_SESSION_TOKEN".into(), v.clone());
64                }
65                if let Some(v) = &a.region {
66                    m.insert("AWS_REGION".into(), v.clone());
67                }
68                if let Some(v) = &a.endpoint_url {
69                    m.insert("AWS_ENDPOINT_URL".into(), v.clone());
70                }
71                if let Some(v) = a.allow_http {
72                    m.insert("AWS_ALLOW_HTTP".into(), v.to_string());
73                }
74            }
75            DeltaCredentials::Azure(a) => {
76                if let Some(v) = &a.account_name {
77                    m.insert("AZURE_STORAGE_ACCOUNT_NAME".into(), v.clone());
78                }
79                if let Some(v) = &a.access_key {
80                    m.insert("AZURE_STORAGE_ACCESS_KEY".into(), v.clone());
81                }
82                if let Some(v) = &a.sas_token {
83                    m.insert("AZURE_STORAGE_SAS_KEY".into(), v.clone());
84                }
85            }
86            DeltaCredentials::Gcp(g) => {
87                if let Some(v) = &g.service_account_path {
88                    m.insert("GOOGLE_SERVICE_ACCOUNT".into(), v.clone());
89                }
90                if let Some(v) = &g.service_account_key {
91                    m.insert("GOOGLE_SERVICE_ACCOUNT_KEY".into(), v.clone());
92                }
93            }
94        }
95        m
96    }
97}
98
99/// Static AWS credentials and optional S3-compatible endpoint overrides.
100///
101/// Leaving a field unset lets the object-store default chain resolve it (so
102/// `region` alone with an instance profile is valid). `endpoint_url` +
103/// `allow_http: true` targets MinIO / LocalStack.
104#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
105pub struct AwsCredentials {
106    /// `AWS_ACCESS_KEY_ID`.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub access_key_id: Option<String>,
109    /// `AWS_SECRET_ACCESS_KEY`.
110    #[serde(default, skip_serializing_if = "Option::is_none")]
111    pub secret_access_key: Option<String>,
112    /// `AWS_SESSION_TOKEN` (temporary credentials).
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub session_token: Option<String>,
115    /// `AWS_REGION`.
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub region: Option<String>,
118    /// `AWS_ENDPOINT_URL` — override for S3-compatible stores (MinIO, etc.).
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub endpoint_url: Option<String>,
121    /// `AWS_ALLOW_HTTP` — permit plaintext HTTP to the endpoint (MinIO in dev).
122    #[serde(default, skip_serializing_if = "Option::is_none")]
123    pub allow_http: Option<bool>,
124}
125
126/// Azure Blob / ADLS Gen2 credentials.
127#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
128pub struct AzureCredentials {
129    /// `AZURE_STORAGE_ACCOUNT_NAME`.
130    #[serde(default, skip_serializing_if = "Option::is_none")]
131    pub account_name: Option<String>,
132    /// `AZURE_STORAGE_ACCESS_KEY` (shared-key auth).
133    #[serde(default, skip_serializing_if = "Option::is_none")]
134    pub access_key: Option<String>,
135    /// `AZURE_STORAGE_SAS_KEY` (SAS-token auth).
136    #[serde(default, skip_serializing_if = "Option::is_none")]
137    pub sas_token: Option<String>,
138}
139
140/// Google Cloud Storage credentials.
141#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
142pub struct GcpCredentials {
143    /// `GOOGLE_SERVICE_ACCOUNT` — path to a service-account JSON key file.
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub service_account_path: Option<String>,
146    /// `GOOGLE_SERVICE_ACCOUNT_KEY` — inline service-account JSON key.
147    #[serde(default, skip_serializing_if = "Option::is_none")]
148    pub service_account_key: Option<String>,
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154    use serde_json::json;
155
156    #[test]
157    fn default_injects_nothing() {
158        let c = DeltaCredentials::default();
159        assert!(c.storage_options().is_empty());
160    }
161
162    #[test]
163    fn aws_maps_all_fields() {
164        let c = DeltaCredentials::Aws(AwsCredentials {
165            access_key_id: Some("AK".into()),
166            secret_access_key: Some("SK".into()),
167            session_token: Some("ST".into()),
168            region: Some("us-east-1".into()),
169            endpoint_url: Some("http://localhost:9000".into()),
170            allow_http: Some(true),
171        });
172        let m = c.storage_options();
173        assert_eq!(m["AWS_ACCESS_KEY_ID"], "AK");
174        assert_eq!(m["AWS_SECRET_ACCESS_KEY"], "SK");
175        assert_eq!(m["AWS_SESSION_TOKEN"], "ST");
176        assert_eq!(m["AWS_REGION"], "us-east-1");
177        assert_eq!(m["AWS_ENDPOINT_URL"], "http://localhost:9000");
178        assert_eq!(m["AWS_ALLOW_HTTP"], "true");
179    }
180
181    #[test]
182    fn azure_and_gcp_partial() {
183        let az = DeltaCredentials::Azure(AzureCredentials {
184            account_name: Some("acct".into()),
185            access_key: None,
186            sas_token: Some("sas".into()),
187        });
188        let m = az.storage_options();
189        assert_eq!(m["AZURE_STORAGE_ACCOUNT_NAME"], "acct");
190        assert_eq!(m["AZURE_STORAGE_SAS_KEY"], "sas");
191        assert!(!m.contains_key("AZURE_STORAGE_ACCESS_KEY"));
192
193        let gcp = DeltaCredentials::Gcp(GcpCredentials {
194            service_account_path: Some("/keys/sa.json".into()),
195            service_account_key: None,
196        });
197        assert_eq!(
198            gcp.storage_options()["GOOGLE_SERVICE_ACCOUNT"],
199            "/keys/sa.json"
200        );
201
202        // Azure shared-key and GCP inline-key branches.
203        let az_key = DeltaCredentials::Azure(AzureCredentials {
204            account_name: None,
205            access_key: Some("shared-key".into()),
206            sas_token: None,
207        });
208        assert_eq!(
209            az_key.storage_options()["AZURE_STORAGE_ACCESS_KEY"],
210            "shared-key"
211        );
212        let gcp_key = DeltaCredentials::Gcp(GcpCredentials {
213            service_account_path: None,
214            service_account_key: Some("{\"type\":\"service_account\"}".into()),
215        });
216        assert_eq!(
217            gcp_key.storage_options()["GOOGLE_SERVICE_ACCOUNT_KEY"],
218            "{\"type\":\"service_account\"}"
219        );
220    }
221
222    #[test]
223    fn explicit_options_win_over_credentials() {
224        let c = DeltaCredentials::Aws(AwsCredentials {
225            region: Some("us-east-1".into()),
226            ..Default::default()
227        });
228        let mut opts = HashMap::new();
229        opts.insert("AWS_REGION".to_string(), "eu-west-1".to_string());
230        c.apply(&mut opts);
231        // Explicit value is preserved.
232        assert_eq!(opts["AWS_REGION"], "eu-west-1");
233    }
234
235    #[test]
236    fn tagged_shape_round_trips() {
237        let v = json!({ "type": "aws", "config": { "region": "us-east-1" } });
238        let c: DeltaCredentials = serde_json::from_value(v).unwrap();
239        assert_eq!(
240            c,
241            DeltaCredentials::Aws(AwsCredentials {
242                region: Some("us-east-1".into()),
243                ..Default::default()
244            })
245        );
246        // Default variant is just `{ "type": "default" }`.
247        let d: DeltaCredentials = serde_json::from_value(json!({ "type": "default" })).unwrap();
248        assert_eq!(d, DeltaCredentials::Default);
249    }
250}
faucet_common_delta/credentials.rs

faucet_common_delta/
credentials.rs
