faucet_common_redshift/
config.rs

1//! Shared Amazon Redshift connection + credentials configuration.
2//!
3//! Redshift speaks the PostgreSQL wire protocol, so both the source and the
4//! sink connect through `sqlx`'s Postgres driver. This module holds the
5//! connection block (host / port / database / user + a TLS toggle) and the
6//! credentials enum that both connectors flatten into their own configs.
7
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10
11/// Default Redshift port.
12pub const DEFAULT_PORT: u16 = 5439;
13
14fn default_port() -> u16 {
15    DEFAULT_PORT
16}
17
18fn default_tls() -> bool {
19    true
20}
21
22/// How to authenticate with Amazon Redshift.
23///
24/// Serializes as `{ type: <method>, config: { … } }` (adjacent tagging,
25/// snake_case discriminators) — the consistent auth wire shape shared by every
26/// faucet connector.
27///
28/// - [`Password`](Self::Password) — a database user and password over the
29///   PostgreSQL wire protocol.
30/// - [`Iam`](Self::Iam) — temporary database credentials from
31///   `redshift:GetClusterCredentials` (provisioned cluster) or
32///   `redshift-serverless:GetCredentials` (workgroup), refreshed before they
33///   expire, over the PostgreSQL wire protocol.
34/// - [`RedshiftDataApi`](Self::RedshiftDataApi) — every statement runs through
35///   the Redshift Data API over HTTPS; no network path to the cluster is
36///   needed.
37#[derive(Clone, Serialize, Deserialize, JsonSchema)]
38#[serde(tag = "type", content = "config", rename_all = "snake_case")]
39pub enum RedshiftCredentials {
40    /// Username/password authentication (the user comes from
41    /// [`RedshiftConnection::user`]).
42    Password {
43        /// The password (use `${env:…}` / `${vault:…}` to inject it, never a
44        /// literal).
45        password: String,
46    },
47    /// IAM authentication with temporary database credentials. Set exactly
48    /// one of `cluster_identifier` (provisioned, `GetClusterCredentials`) or
49    /// `workgroup_name` (serverless, `GetCredentials`). The connection then
50    /// uses the returned user and password over the PostgreSQL wire protocol
51    /// (`host`, `port`, `database` and TLS as configured), and new pool
52    /// connections get fresh credentials before the old ones expire.
53    Iam {
54        /// AWS region of the cluster or workgroup. `None` uses the SDK
55        /// default chain (`AWS_REGION`, profile).
56        #[serde(default, skip_serializing_if = "Option::is_none")]
57        region: Option<String>,
58        /// Provisioned cluster identifier (`GetClusterCredentials`).
59        #[serde(default, skip_serializing_if = "Option::is_none")]
60        cluster_identifier: Option<String>,
61        /// Database user to request credentials for (provisioned only).
62        /// Defaults to [`RedshiftConnection::user`].
63        #[serde(default, skip_serializing_if = "Option::is_none")]
64        db_user: Option<String>,
65        /// Serverless workgroup name (`redshift-serverless:GetCredentials`);
66        /// the database user is derived from the IAM identity.
67        #[serde(default, skip_serializing_if = "Option::is_none")]
68        workgroup_name: Option<String>,
69        /// Create `db_user` when it does not exist (provisioned only).
70        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
71        auto_create: bool,
72        /// Database groups the user joins for the session (provisioned only).
73        #[serde(default, skip_serializing_if = "Vec::is_empty")]
74        db_groups: Vec<String>,
75        /// Lifetime of the temporary credentials, 900–3600 seconds. `None`
76        /// uses the service default (900).
77        #[serde(default, skip_serializing_if = "Option::is_none")]
78        duration_secs: Option<u32>,
79        /// AWS credentials used to call the credentials API. Defaults to the
80        /// SDK default chain.
81        #[serde(default, skip_serializing_if = "RedshiftAwsCredentials::is_default")]
82        aws_credentials: RedshiftAwsCredentials,
83        /// Endpoint override for the credentials API (VPC endpoints, tests).
84        #[serde(default, skip_serializing_if = "Option::is_none")]
85        #[schemars(extend("x-faucet-resource" = "url"))]
86        endpoint_url: Option<String>,
87    },
88    /// Every statement runs through the Redshift Data API over HTTPS
89    /// (`ExecuteStatement`, `DescribeStatement`, `GetStatementResult`), not the
90    /// PostgreSQL wire protocol, so `host`, `port`, `user` and the TLS settings
91    /// are not used. Set exactly one of `cluster_identifier` or
92    /// `workgroup_name`; a cluster also needs `db_user` (temporary
93    /// credentials) or `secret_arn`.
94    RedshiftDataApi {
95        /// AWS region. `None` uses the SDK default chain.
96        #[serde(default, skip_serializing_if = "Option::is_none")]
97        region: Option<String>,
98        /// Provisioned cluster identifier.
99        #[serde(default, skip_serializing_if = "Option::is_none")]
100        cluster_identifier: Option<String>,
101        /// Serverless workgroup name (alternative to `cluster_identifier`).
102        #[serde(default, skip_serializing_if = "Option::is_none")]
103        workgroup_name: Option<String>,
104        /// Secrets Manager ARN holding the database credentials.
105        #[serde(default, skip_serializing_if = "Option::is_none")]
106        secret_arn: Option<String>,
107        /// Database user to authenticate as with temporary credentials
108        /// (provisioned clusters only).
109        #[serde(default, skip_serializing_if = "Option::is_none")]
110        db_user: Option<String>,
111        /// AWS credentials used to call the Data API. Defaults to the SDK
112        /// default chain.
113        #[serde(default, skip_serializing_if = "RedshiftAwsCredentials::is_default")]
114        aws_credentials: RedshiftAwsCredentials,
115        /// Endpoint override for the Data API (VPC endpoints, tests).
116        #[serde(default, skip_serializing_if = "Option::is_none")]
117        #[schemars(extend("x-faucet-resource" = "url"))]
118        endpoint_url: Option<String>,
119    },
120}
121
122/// The AWS credentials the `iam` and `redshift_data_api` methods call AWS
123/// with. The same shape as the other AWS connectors' `credentials`.
124#[derive(Clone, Default, Serialize, Deserialize, JsonSchema)]
125#[serde(tag = "type", content = "config", rename_all = "snake_case")]
126pub enum RedshiftAwsCredentials {
127    /// The AWS SDK default provider chain: environment variables, shared
128    /// config/credentials files, container credentials, web-identity tokens
129    /// and instance profiles, refreshed automatically.
130    #[default]
131    Default,
132    /// A named profile from the shared AWS config/credentials files.
133    Profile {
134        /// Profile name.
135        name: String,
136    },
137    /// Static access keys. Prefer `${env:…}` / secrets-manager interpolation
138    /// over literals.
139    AccessKey {
140        /// AWS access key id.
141        access_key_id: String,
142        /// AWS secret access key.
143        secret_access_key: String,
144        /// Optional session token (temporary credentials).
145        #[serde(default, skip_serializing_if = "Option::is_none")]
146        session_token: Option<String>,
147    },
148    /// Assume an IAM role via STS on top of the default provider chain.
149    AssumeRole {
150        /// ARN of the role to assume.
151        role_arn: String,
152        /// Session name recorded in CloudTrail. Defaults to `faucet-stream`.
153        #[serde(default, skip_serializing_if = "Option::is_none")]
154        session_name: Option<String>,
155        /// Optional external id for cross-account trust policies.
156        #[serde(default, skip_serializing_if = "Option::is_none")]
157        external_id: Option<String>,
158    },
159    /// Web-identity federation (`AWS_WEB_IDENTITY_TOKEN_FILE` +
160    /// `AWS_ROLE_ARN`, e.g. EKS IRSA), through the default chain.
161    WebIdentity,
162}
163
164impl RedshiftAwsCredentials {
165    /// Whether this is the default provider chain.
166    pub fn is_default(&self) -> bool {
167        matches!(self, Self::Default)
168    }
169}
170
171impl std::fmt::Debug for RedshiftAwsCredentials {
172    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173        match self {
174            Self::Default => write!(f, "Default"),
175            Self::Profile { name } => f.debug_struct("Profile").field("name", name).finish(),
176            Self::AccessKey { access_key_id, .. } => f
177                .debug_struct("AccessKey")
178                .field("access_key_id", access_key_id)
179                .field("secret_access_key", &"***")
180                .finish(),
181            Self::AssumeRole { role_arn, .. } => f
182                .debug_struct("AssumeRole")
183                .field("role_arn", role_arn)
184                .finish(),
185            Self::WebIdentity => write!(f, "WebIdentity"),
186        }
187    }
188}
189
190impl std::fmt::Debug for RedshiftCredentials {
191    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
192        match self {
193            Self::Password { .. } => write!(f, "Password(***)"),
194            Self::Iam {
195                region,
196                cluster_identifier,
197                db_user,
198                workgroup_name,
199                aws_credentials,
200                ..
201            } => f
202                .debug_struct("Iam")
203                .field("region", region)
204                .field("cluster_identifier", cluster_identifier)
205                .field("db_user", db_user)
206                .field("workgroup_name", workgroup_name)
207                .field("aws_credentials", aws_credentials)
208                .finish_non_exhaustive(),
209            Self::RedshiftDataApi {
210                region,
211                cluster_identifier,
212                workgroup_name,
213                secret_arn,
214                db_user,
215                aws_credentials,
216                ..
217            } => f
218                .debug_struct("RedshiftDataApi")
219                .field("region", region)
220                .field("cluster_identifier", cluster_identifier)
221                .field("workgroup_name", workgroup_name)
222                .field("secret_arn", secret_arn)
223                .field("db_user", db_user)
224                .field("aws_credentials", aws_credentials)
225                .finish_non_exhaustive(),
226        }
227    }
228}
229
230/// A Redshift connection: endpoint, database, user, credentials, and a TLS
231/// toggle. Flattened (`#[serde(flatten)]`) into both the source and sink
232/// configs so the connection fields appear at the config top level.
233#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
234pub struct RedshiftConnection {
235    /// Cluster / endpoint host (e.g. `my-cluster.abc123.us-east-1.redshift.amazonaws.com`).
236    /// Required except with `credentials: { type: redshift_data_api }`.
237    #[serde(default)]
238    #[schemars(extend("x-faucet-resource" = "url"))]
239    pub host: String,
240    /// Port. Defaults to [`DEFAULT_PORT`] (5439).
241    #[serde(default = "default_port")]
242    pub port: u16,
243    /// Database name.
244    pub database: String,
245    /// Database user. Required for `password`; for a provisioned `iam`
246    /// cluster it is the default `db_user`; unused by `redshift_data_api`.
247    #[serde(default)]
248    pub user: String,
249    /// Authentication credentials.
250    pub credentials: RedshiftCredentials,
251    /// Whether to require TLS. Defaults to `true` (Redshift clusters require SSL
252    /// by default). `true` maps to `sslmode=require` (encrypted, but the server
253    /// certificate is **not** verified); `false` maps to `sslmode=prefer`
254    /// (opportunistic TLS with plaintext fallback). Ignored when `tls_mode` is
255    /// set.
256    #[serde(default = "default_tls")]
257    pub tls: bool,
258    /// Explicit TLS mode; overrides `tls`. `verify_ca` / `verify_full` check
259    /// the server certificate against `ssl_root_cert` (or, without one, the
260    /// Mozilla root store bundled into the binary — not the operating
261    /// system's), the latter also its host name.
262    #[serde(default, skip_serializing_if = "Option::is_none")]
263    pub tls_mode: Option<RedshiftTlsMode>,
264    /// PEM file of the CA that signed the server certificate (the Redshift CA
265    /// bundle), for `tls_mode: verify_ca | verify_full`.
266    #[serde(default, skip_serializing_if = "Option::is_none")]
267    #[schemars(extend("x-faucet-resource" = "path"))]
268    pub ssl_root_cert: Option<String>,
269}
270
271/// How a Redshift connection uses TLS.
272#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
273#[serde(rename_all = "snake_case")]
274pub enum RedshiftTlsMode {
275    /// Plaintext only.
276    Disable,
277    /// TLS when the server offers it, else plaintext.
278    Prefer,
279    /// TLS required; the certificate is not verified.
280    Require,
281    /// TLS required and the certificate chain verified.
282    VerifyCa,
283    /// As `verify_ca`, plus the certificate must name the host.
284    VerifyFull,
285}
286
287impl RedshiftConnection {
288    /// Build a `Password` connection with sensible defaults (port 5439, TLS on).
289    pub fn new(
290        host: impl Into<String>,
291        database: impl Into<String>,
292        user: impl Into<String>,
293        password: impl Into<String>,
294    ) -> Self {
295        Self {
296            host: host.into(),
297            port: DEFAULT_PORT,
298            database: database.into(),
299            user: user.into(),
300            credentials: RedshiftCredentials::Password {
301                password: password.into(),
302            },
303            tls: true,
304            tls_mode: None,
305            ssl_root_cert: None,
306        }
307    }
308
309    /// Whether statements run through the Redshift Data API rather than the
310    /// PostgreSQL wire protocol.
311    pub fn uses_data_api(&self) -> bool {
312        matches!(
313            self.credentials,
314            RedshiftCredentials::RedshiftDataApi { .. }
315        )
316    }
317
318    /// What the connection addresses, for dataset URIs and logs: the host, or
319    /// `cluster:<id>` / `workgroup:<name>` for the Data API.
320    pub fn endpoint_label(&self) -> String {
321        match &self.credentials {
322            RedshiftCredentials::RedshiftDataApi {
323                cluster_identifier,
324                workgroup_name,
325                ..
326            } => match (cluster_identifier, workgroup_name) {
327                (Some(c), _) => format!("cluster:{c}"),
328                (None, Some(w)) => format!("workgroup:{w}"),
329                (None, None) => String::new(),
330            },
331            _ => self.host.clone(),
332        }
333    }
334}
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339
340    #[test]
341    fn debug_masks_password() {
342        let c = RedshiftCredentials::Password {
343            password: "s3cr3t".into(),
344        };
345        let dbg = format!("{c:?}");
346        assert!(dbg.contains("***"));
347        assert!(!dbg.contains("s3cr3t"));
348    }
349
350    #[test]
351    fn connection_debug_does_not_leak_password() {
352        let conn = RedshiftConnection::new("host", "db", "user", "hunter2");
353        let dbg = format!("{conn:?}");
354        assert!(!dbg.contains("hunter2"));
355        assert!(dbg.contains("host"));
356        assert!(dbg.contains("user"));
357    }
358
359    #[test]
360    fn password_credentials_round_trip() {
361        let c = RedshiftCredentials::Password {
362            password: "pw".into(),
363        };
364        let json = serde_json::to_string(&c).unwrap();
365        assert_eq!(json, r#"{"type":"password","config":{"password":"pw"}}"#);
366        let back: RedshiftCredentials = serde_json::from_str(&json).unwrap();
367        assert!(matches!(back, RedshiftCredentials::Password { .. }));
368    }
369
370    #[test]
371    fn connection_defaults_port_and_tls() {
372        let json = r#"{
373            "host": "h",
374            "database": "db",
375            "user": "u",
376            "credentials": {"type": "password", "config": {"password": "pw"}}
377        }"#;
378        let conn: RedshiftConnection = serde_json::from_str(json).unwrap();
379        assert_eq!(conn.port, DEFAULT_PORT);
380        assert!(conn.tls);
381    }
382
383    #[test]
384    fn iam_variant_deserializes() {
385        let json = r#"{"type":"iam","config":{"region":"us-east-1","db_user":"analyst"}}"#;
386        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
387        match c {
388            RedshiftCredentials::Iam {
389                region, db_user, ..
390            } => {
391                assert_eq!(region.as_deref(), Some("us-east-1"));
392                assert_eq!(db_user.as_deref(), Some("analyst"));
393            }
394            _ => panic!("expected Iam"),
395        }
396    }
397
398    #[test]
399    fn redshift_data_api_variant_deserializes() {
400        let json =
401            r#"{"type":"redshift_data_api","config":{"workgroup_name":"wg","secret_arn":"arn:x"}}"#;
402        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
403        assert!(matches!(c, RedshiftCredentials::RedshiftDataApi { .. }));
404    }
405
406    #[test]
407    fn iam_debug_renders_fields_and_masks_keys() {
408        let c: RedshiftCredentials = serde_json::from_value(serde_json::json!({
409            "type": "iam",
410            "config": {
411                "region": "us-west-2",
412                "cluster_identifier": "prod-cluster",
413                "db_user": "analyst",
414                "aws_credentials": {"type": "access_key", "config": {
415                    "access_key_id": "AKIDX", "secret_access_key": "topsecret"}}
416            }
417        }))
418        .unwrap();
419        let dbg = format!("{c:?}");
420        for want in ["Iam", "us-west-2", "prod-cluster", "analyst", "AKIDX"] {
421            assert!(dbg.contains(want), "{dbg}");
422        }
423        assert!(!dbg.contains("topsecret"), "{dbg}");
424    }
425
426    #[test]
427    fn redshift_data_api_debug_renders_fields() {
428        let c: RedshiftCredentials = serde_json::from_value(serde_json::json!({
429            "type": "redshift_data_api",
430            "config": {"region": "eu-central-1", "workgroup_name": "wg-1",
431                       "secret_arn": "arn:aws:secretsmanager:x", "db_user": "svc"}
432        }))
433        .unwrap();
434        let dbg = format!("{c:?}");
435        for want in ["RedshiftDataApi", "eu-central-1", "wg-1", "svc", "Default"] {
436            assert!(dbg.contains(want), "{dbg}");
437        }
438    }
439
440    #[test]
441    fn aws_credentials_parse_and_debug() {
442        for (json, want) in [
443            (r#"{"type":"default"}"#, "Default"),
444            (r#"{"type":"profile","config":{"name":"p"}}"#, "Profile"),
445            (
446                r#"{"type":"assume_role","config":{"role_arn":"arn:r"}}"#,
447                "arn:r",
448            ),
449            (r#"{"type":"web_identity"}"#, "WebIdentity"),
450        ] {
451            let c: RedshiftAwsCredentials = serde_json::from_str(json).unwrap();
452            assert!(format!("{c:?}").contains(want), "{json}");
453        }
454        assert!(RedshiftAwsCredentials::default().is_default());
455    }
456
457    #[test]
458    fn endpoint_label_names_the_host_or_the_data_api_target() {
459        let mut c = RedshiftConnection::new("h.example", "db", "u", "pw");
460        assert_eq!(c.endpoint_label(), "h.example");
461        assert!(!c.uses_data_api());
462        for (cfg, want) in [
463            (
464                serde_json::json!({"cluster_identifier": "c1"}),
465                "cluster:c1",
466            ),
467            (serde_json::json!({"workgroup_name": "wg"}), "workgroup:wg"),
468            (serde_json::json!({}), ""),
469        ] {
470            c.credentials = serde_json::from_value(
471                serde_json::json!({"type": "redshift_data_api", "config": cfg}),
472            )
473            .unwrap();
474            assert!(c.uses_data_api());
475            assert_eq!(c.endpoint_label(), want);
476        }
477    }
478
479    #[test]
480    fn data_api_connection_needs_no_host_or_user() {
481        let conn: RedshiftConnection = serde_json::from_str(
482            r#"{"database":"dev","credentials":{"type":"redshift_data_api","config":{"workgroup_name":"wg"}}}"#,
483        )
484        .unwrap();
485        assert!(conn.host.is_empty() && conn.user.is_empty());
486    }
487
488    #[test]
489    fn iam_and_data_api_round_trip_full_fields() {
490        let iam = r#"{"type":"iam","config":{"region":"us-east-1","cluster_identifier":"c1","db_user":"u"}}"#;
491        let back: RedshiftCredentials = serde_json::from_str(iam).unwrap();
492        assert_eq!(serde_json::to_string(&back).unwrap(), iam);
493
494        let api = r#"{"type":"redshift_data_api","config":{"region":"us-east-1","cluster_identifier":"c1","workgroup_name":"wg","secret_arn":"arn","db_user":"u"}}"#;
495        let back: RedshiftCredentials = serde_json::from_str(api).unwrap();
496        assert_eq!(serde_json::to_string(&back).unwrap(), api);
497    }
498}
faucet_common_redshift/config.rs

faucet_common_redshift/
config.rs
