laminar_connectors/storage/validation/
mod.rs

1//! Per-cloud-provider configuration validation.
2//!
3//! [`CloudConfigValidator`] checks [`ResolvedStorageOptions`] for missing
4//! or invalid credentials at connector `open()` time, producing clear,
5//! actionable error messages that include both the config key name and
6//! the fallback environment variable.
7
8use super::provider::StorageProvider;
9use super::resolver::{AuthSource, ResolvedStorageOptions};
10
11/// Result of cloud configuration validation.
12#[derive(Debug, Clone)]
13pub struct CloudValidationResult {
14    /// Hard errors that prevent the connector from opening.
15    pub errors: Vec<CloudValidationError>,
16    /// Soft warnings (may still work with instance metadata / default creds).
17    pub warnings: Vec<CloudValidationWarning>,
18}
19
20impl CloudValidationResult {
21    /// Returns true if there are no hard errors.
22    #[must_use]
23    pub fn is_valid(&self) -> bool {
24        self.errors.is_empty()
25    }
26
27    /// Formats all errors into a single string for `ConnectorError`.
28    #[must_use]
29    pub fn error_message(&self) -> String {
30        self.errors
31            .iter()
32            .map(|error| match &error.env_var {
33                Some(env_var) if !error.message.contains(env_var) => {
34                    format!("{} (environment: {env_var})", error.message)
35                }
36                _ => error.message.clone(),
37            })
38            .collect::<Vec<_>>()
39            .join("; ")
40    }
41}
42
43/// A hard validation error.
44#[derive(Debug, Clone)]
45pub struct CloudValidationError {
46    /// The missing or invalid configuration key.
47    pub key: String,
48    /// The fallback environment variable (if applicable).
49    pub env_var: Option<String>,
50    /// Human-readable error message.
51    pub message: String,
52}
53
54/// A soft validation warning.
55#[derive(Debug, Clone)]
56pub struct CloudValidationWarning {
57    /// The configuration key this warning relates to.
58    pub key: String,
59    /// Human-readable warning message.
60    pub message: String,
61}
62
63/// Validates resolved storage options for a given provider.
64pub struct CloudConfigValidator;
65
66impl CloudConfigValidator {
67    /// Validates the resolved storage options for the detected provider.
68    ///
69    /// Returns a [`CloudValidationResult`] with any errors or warnings.
70    /// Hard errors indicate the connector cannot open. Warnings indicate
71    /// missing credentials that may still be resolved by instance metadata
72    /// or default credential providers.
73    #[must_use]
74    pub fn validate(resolved: &ResolvedStorageOptions) -> CloudValidationResult {
75        let mut result = match resolved.provider {
76            StorageProvider::AwsS3 => Self::validate_s3(resolved),
77            StorageProvider::AzureAdls => Self::validate_azure(resolved),
78            StorageProvider::Gcs => Self::validate_gcs(resolved),
79            StorageProvider::Local => CloudValidationResult {
80                errors: Vec::new(),
81                warnings: Vec::new(),
82            },
83        };
84        validate_endpoint_options(resolved, &mut result.errors);
85        result
86    }
87
88    fn validate_s3(resolved: &ResolvedStorageOptions) -> CloudValidationResult {
89        let mut errors = Vec::new();
90        let mut warnings = Vec::new();
91
92        if !has_any(resolved, &["aws_region", "aws_default_region", "region"]) {
93            warnings.push(CloudValidationWarning {
94                key: "aws_region".into(),
95                message: "No explicit AWS region is configured; the downstream AWS region chain remains active"
96                    .into(),
97            });
98        }
99
100        // If access key is provided, secret key must also be provided.
101        let access_key = has_any(resolved, &["aws_access_key_id", "access_key_id"]);
102        let secret_key = has_any(resolved, &["aws_secret_access_key", "secret_access_key"]);
103        if access_key && !secret_key {
104            errors.push(CloudValidationError {
105                key: "aws_secret_access_key".into(),
106                env_var: Some("AWS_SECRET_ACCESS_KEY".into()),
107                message: "'storage.aws_access_key_id' provided without \
108                          'storage.aws_secret_access_key'"
109                    .into(),
110            });
111        }
112
113        // If secret key is provided, access key must also be provided.
114        if secret_key && !access_key {
115            errors.push(CloudValidationError {
116                key: "aws_access_key_id".into(),
117                env_var: Some("AWS_ACCESS_KEY_ID".into()),
118                message: "'storage.aws_secret_access_key' provided without \
119                          'storage.aws_access_key_id'"
120                    .into(),
121            });
122        }
123
124        let web_identity = has_any(
125            resolved,
126            &["aws_web_identity_token_file", "web_identity_token_file"],
127        );
128        let role = has_any(resolved, &["aws_role_arn", "role_arn"]);
129        if web_identity != role {
130            let key = if web_identity {
131                "aws_role_arn"
132            } else {
133                "aws_web_identity_token_file"
134            };
135            errors.push(CloudValidationError {
136                key: key.into(),
137                env_var: Some(if web_identity {
138                    "AWS_ROLE_ARN".into()
139                } else {
140                    "AWS_WEB_IDENTITY_TOKEN_FILE".into()
141                }),
142                message: format!("AWS web identity configuration is missing '{key}'"),
143            });
144        }
145
146        CloudValidationResult { errors, warnings }
147    }
148
149    fn validate_azure(resolved: &ResolvedStorageOptions) -> CloudValidationResult {
150        let mut errors = Vec::new();
151        let warnings = Vec::new();
152
153        // Account name is always required for Azure.
154        if !has_any(
155            resolved,
156            &["azure_storage_account_name", "azure_account_name"],
157        ) {
158            errors.push(CloudValidationError {
159                key: "azure_storage_account_name".into(),
160                env_var: Some("AZURE_STORAGE_ACCOUNT_NAME".into()),
161                message: "Azure paths require 'storage.azure_storage_account_name' \
162                          in config or AZURE_STORAGE_ACCOUNT_NAME environment variable"
163                    .into(),
164            });
165        }
166
167        let client_secret = has_any(
168            resolved,
169            &[
170                "azure_storage_client_secret",
171                "azure_client_secret",
172                "client_secret",
173            ],
174        );
175        let client_id = has_any(
176            resolved,
177            &["azure_storage_client_id", "azure_client_id", "client_id"],
178        );
179        let tenant_id = has_any(
180            resolved,
181            &["azure_storage_tenant_id", "azure_tenant_id", "tenant_id"],
182        );
183        if client_secret && !client_id {
184            errors.push(missing_field(
185                "azure_storage_client_id",
186                "AZURE_CLIENT_ID",
187                "Azure client-secret authentication",
188            ));
189        }
190        if client_secret && !tenant_id {
191            errors.push(missing_field(
192                "azure_storage_tenant_id",
193                "AZURE_TENANT_ID",
194                "Azure client-secret authentication",
195            ));
196        }
197        if resolved.auth_source == AuthSource::WorkloadIdentity {
198            if !client_id {
199                errors.push(missing_field(
200                    "azure_storage_client_id",
201                    "AZURE_CLIENT_ID",
202                    "Azure workload identity",
203                ));
204            }
205            if !tenant_id {
206                errors.push(missing_field(
207                    "azure_storage_tenant_id",
208                    "AZURE_TENANT_ID",
209                    "Azure workload identity",
210                ));
211            }
212        }
213
214        CloudValidationResult { errors, warnings }
215    }
216
217    fn validate_gcs(resolved: &ResolvedStorageOptions) -> CloudValidationResult {
218        let mut errors = Vec::new();
219        if has_any(resolved, &["gcs.token", "google_token"]) {
220            errors.push(CloudValidationError {
221                key: "gcs.token".into(),
222                env_var: None,
223                message: "the pinned Delta/object_store GCS backend does not accept a direct access token; configure a service-account path/key or supported Application Default Credentials"
224                    .into(),
225            });
226        }
227        CloudValidationResult {
228            errors,
229            warnings: Vec::new(),
230        }
231    }
232}
233
234fn has_any(resolved: &ResolvedStorageOptions, keys: &[&str]) -> bool {
235    resolved.options.iter().any(|(candidate, value)| {
236        keys.iter().any(|key| candidate.eq_ignore_ascii_case(key)) && !value.trim().is_empty()
237    }) || resolved
238        .env_resolved_keys
239        .iter()
240        .any(|candidate| keys.iter().any(|key| candidate.eq_ignore_ascii_case(key)))
241}
242
243fn missing_field(key: &str, env_var: &str, context: &str) -> CloudValidationError {
244    CloudValidationError {
245        key: key.into(),
246        env_var: Some(env_var.into()),
247        message: format!("{context} is missing '{key}'"),
248    }
249}
250
251fn validate_endpoint_options(
252    resolved: &ResolvedStorageOptions,
253    errors: &mut Vec<CloudValidationError>,
254) {
255    let endpoints = resolved.options.iter().filter(|(key, value)| {
256        matches!(
257            key.to_ascii_lowercase().as_str(),
258            "endpoint"
259                | "endpoint_url"
260                | "aws_endpoint"
261                | "aws_endpoint_url"
262                | "aws_endpoint_url_s3"
263                | "azure_storage_endpoint"
264                | "azure_endpoint"
265                | "google_base_url"
266                | "base_url"
267        ) && !value.trim().is_empty()
268    });
269    if resolved.provider != StorageProvider::Local
270        && resolved.auth_source == AuthSource::Anonymous
271        && !resolved.endpoint_override_configured
272    {
273        errors.push(CloudValidationError {
274            key: "storage.endpoint".into(),
275            env_var: None,
276            message:
277                "anonymous object-store access requires an explicit compatibility or test endpoint"
278                    .into(),
279        });
280    }
281    for (key, value) in endpoints {
282        let parsed = match url::Url::parse(value) {
283            Ok(parsed)
284                if matches!(parsed.scheme(), "http" | "https")
285                    && parsed.host_str().is_some()
286                    && parsed.username().is_empty()
287                    && parsed.password().is_none()
288                    && parsed.query().is_none()
289                    && parsed.fragment().is_none() =>
290            {
291                parsed
292            }
293            _ => {
294                errors.push(CloudValidationError {
295                    key: key.clone(),
296                    env_var: None,
297                    message: format!(
298                        "storage endpoint option '{key}' must be an HTTP(S) URL without credentials, query parameters, or fragments"
299                    ),
300                });
301                continue;
302            }
303        };
304        if parsed.scheme() == "http"
305            && !has_true(
306                resolved,
307                &[
308                    "aws_allow_http",
309                    "allow_http",
310                    "google_allow_http",
311                    "azure_allow_http",
312                    "azure_storage_use_emulator",
313                    "use_emulator",
314                ],
315            )
316        {
317            errors.push(CloudValidationError {
318                key: "allow_http".into(),
319                env_var: None,
320                message:
321                    "an HTTP object-store endpoint requires an explicit allow-http or emulator option"
322                        .into(),
323            });
324        }
325    }
326}
327
328fn has_true(resolved: &ResolvedStorageOptions, keys: &[&str]) -> bool {
329    resolved.options.iter().any(|(candidate, value)| {
330        keys.iter().any(|key| candidate.eq_ignore_ascii_case(key))
331            && value.trim().eq_ignore_ascii_case("true")
332    }) || resolved
333        .env_resolved_keys
334        .iter()
335        .any(|candidate| keys.iter().any(|key| candidate.eq_ignore_ascii_case(key)))
336}
337
338#[cfg(test)]
339#[allow(clippy::disallowed_types)] // cold path: storage configuration
340mod tests;
laminar_connectors/storage/validation/mod.rs

laminar_connectors/storage/validation/
mod.rs
