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/// v1 implements only [`RedshiftCredentials::Password`]. The [`Iam`] and
29/// [`RedshiftDataApi`] variants are accepted by the config parser (so a future
30/// version can add them without a breaking change) but currently return a typed
31/// [`FaucetError::Config`](faucet_core::FaucetError::Config) at client-build
32/// time.
33///
34/// [`Iam`]: RedshiftCredentials::Iam
35/// [`RedshiftDataApi`]: RedshiftCredentials::RedshiftDataApi
36#[derive(Clone, Serialize, Deserialize, JsonSchema)]
37#[serde(tag = "type", content = "config", rename_all = "snake_case")]
38pub enum RedshiftCredentials {
39    /// Username/password authentication (the user comes from
40    /// [`RedshiftConnection::user`]). The only mechanism supported in v1.
41    Password {
42        /// The password (use `${env:…}` / `${vault:…}` to inject it, never a
43        /// literal).
44        password: String,
45    },
46    /// IAM authentication via temporary cluster credentials
47    /// (`GetClusterCredentials`). **Not yet supported** — reserved for a future
48    /// version; building a client with this variant returns a typed error.
49    Iam {
50        /// AWS region of the cluster.
51        #[serde(default, skip_serializing_if = "Option::is_none")]
52        region: Option<String>,
53        /// Provisioned cluster identifier used to request temporary credentials.
54        #[serde(default, skip_serializing_if = "Option::is_none")]
55        cluster_identifier: Option<String>,
56        /// Database user to authenticate as.
57        #[serde(default, skip_serializing_if = "Option::is_none")]
58        db_user: Option<String>,
59    },
60    /// Authentication through the Redshift Data API (HTTP, not the PG wire).
61    /// **Not yet supported** — reserved for a future version; building a client
62    /// with this variant returns a typed error.
63    RedshiftDataApi {
64        /// AWS region.
65        #[serde(default, skip_serializing_if = "Option::is_none")]
66        region: Option<String>,
67        /// Provisioned cluster identifier.
68        #[serde(default, skip_serializing_if = "Option::is_none")]
69        cluster_identifier: Option<String>,
70        /// Serverless workgroup name (alternative to `cluster_identifier`).
71        #[serde(default, skip_serializing_if = "Option::is_none")]
72        workgroup_name: Option<String>,
73        /// Secrets Manager ARN holding the database credentials.
74        #[serde(default, skip_serializing_if = "Option::is_none")]
75        secret_arn: Option<String>,
76        /// Database user to authenticate as.
77        #[serde(default, skip_serializing_if = "Option::is_none")]
78        db_user: Option<String>,
79    },
80}
81
82impl std::fmt::Debug for RedshiftCredentials {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        match self {
85            Self::Password { .. } => write!(f, "Password(***)"),
86            Self::Iam {
87                region,
88                cluster_identifier,
89                db_user,
90            } => f
91                .debug_struct("Iam")
92                .field("region", region)
93                .field("cluster_identifier", cluster_identifier)
94                .field("db_user", db_user)
95                .finish(),
96            Self::RedshiftDataApi {
97                region,
98                cluster_identifier,
99                workgroup_name,
100                secret_arn,
101                db_user,
102            } => f
103                .debug_struct("RedshiftDataApi")
104                .field("region", region)
105                .field("cluster_identifier", cluster_identifier)
106                .field("workgroup_name", workgroup_name)
107                .field("secret_arn", secret_arn)
108                .field("db_user", db_user)
109                .finish(),
110        }
111    }
112}
113
114/// A Redshift connection: endpoint, database, user, credentials, and a TLS
115/// toggle. Flattened (`#[serde(flatten)]`) into both the source and sink
116/// configs so the connection fields appear at the config top level.
117#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
118pub struct RedshiftConnection {
119    /// Cluster / endpoint host (e.g. `my-cluster.abc123.us-east-1.redshift.amazonaws.com`).
120    #[schemars(extend("x-faucet-resource" = "url"))]
121    pub host: String,
122    /// Port. Defaults to [`DEFAULT_PORT`] (5439).
123    #[serde(default = "default_port")]
124    pub port: u16,
125    /// Database name.
126    pub database: String,
127    /// Database user.
128    pub user: String,
129    /// Authentication credentials.
130    pub credentials: RedshiftCredentials,
131    /// Whether to require TLS. Defaults to `true` (Redshift clusters require SSL
132    /// by default). `true` maps to `sslmode=require` (encrypted, but the server
133    /// certificate is **not** verified); `false` maps to `sslmode=prefer`
134    /// (opportunistic TLS with plaintext fallback). Ignored when `tls_mode` is
135    /// set.
136    #[serde(default = "default_tls")]
137    pub tls: bool,
138    /// Explicit TLS mode; overrides `tls`. `verify_ca` / `verify_full` check
139    /// the server certificate against `ssl_root_cert` (or, without one, the
140    /// Mozilla root store bundled into the binary — not the operating
141    /// system's), the latter also its host name.
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub tls_mode: Option<RedshiftTlsMode>,
144    /// PEM file of the CA that signed the server certificate (the Redshift CA
145    /// bundle), for `tls_mode: verify_ca | verify_full`.
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    #[schemars(extend("x-faucet-resource" = "path"))]
148    pub ssl_root_cert: Option<String>,
149}
150
151/// How a Redshift connection uses TLS.
152#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
153#[serde(rename_all = "snake_case")]
154pub enum RedshiftTlsMode {
155    /// Plaintext only.
156    Disable,
157    /// TLS when the server offers it, else plaintext.
158    Prefer,
159    /// TLS required; the certificate is not verified.
160    Require,
161    /// TLS required and the certificate chain verified.
162    VerifyCa,
163    /// As `verify_ca`, plus the certificate must name the host.
164    VerifyFull,
165}
166
167impl RedshiftConnection {
168    /// Build a `Password` connection with sensible defaults (port 5439, TLS on).
169    pub fn new(
170        host: impl Into<String>,
171        database: impl Into<String>,
172        user: impl Into<String>,
173        password: impl Into<String>,
174    ) -> Self {
175        Self {
176            host: host.into(),
177            port: DEFAULT_PORT,
178            database: database.into(),
179            user: user.into(),
180            credentials: RedshiftCredentials::Password {
181                password: password.into(),
182            },
183            tls: true,
184            tls_mode: None,
185            ssl_root_cert: None,
186        }
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193
194    #[test]
195    fn debug_masks_password() {
196        let c = RedshiftCredentials::Password {
197            password: "s3cr3t".into(),
198        };
199        let dbg = format!("{c:?}");
200        assert!(dbg.contains("***"));
201        assert!(!dbg.contains("s3cr3t"));
202    }
203
204    #[test]
205    fn connection_debug_does_not_leak_password() {
206        let conn = RedshiftConnection::new("host", "db", "user", "hunter2");
207        let dbg = format!("{conn:?}");
208        assert!(!dbg.contains("hunter2"));
209        assert!(dbg.contains("host"));
210        assert!(dbg.contains("user"));
211    }
212
213    #[test]
214    fn password_credentials_round_trip() {
215        let c = RedshiftCredentials::Password {
216            password: "pw".into(),
217        };
218        let json = serde_json::to_string(&c).unwrap();
219        assert_eq!(json, r#"{"type":"password","config":{"password":"pw"}}"#);
220        let back: RedshiftCredentials = serde_json::from_str(&json).unwrap();
221        assert!(matches!(back, RedshiftCredentials::Password { .. }));
222    }
223
224    #[test]
225    fn connection_defaults_port_and_tls() {
226        let json = r#"{
227            "host": "h",
228            "database": "db",
229            "user": "u",
230            "credentials": {"type": "password", "config": {"password": "pw"}}
231        }"#;
232        let conn: RedshiftConnection = serde_json::from_str(json).unwrap();
233        assert_eq!(conn.port, DEFAULT_PORT);
234        assert!(conn.tls);
235    }
236
237    #[test]
238    fn iam_variant_deserializes() {
239        let json = r#"{"type":"iam","config":{"region":"us-east-1","db_user":"analyst"}}"#;
240        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
241        match c {
242            RedshiftCredentials::Iam {
243                region, db_user, ..
244            } => {
245                assert_eq!(region.as_deref(), Some("us-east-1"));
246                assert_eq!(db_user.as_deref(), Some("analyst"));
247            }
248            _ => panic!("expected Iam"),
249        }
250    }
251
252    #[test]
253    fn redshift_data_api_variant_deserializes() {
254        let json =
255            r#"{"type":"redshift_data_api","config":{"workgroup_name":"wg","secret_arn":"arn:x"}}"#;
256        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
257        assert!(matches!(c, RedshiftCredentials::RedshiftDataApi { .. }));
258    }
259
260    #[test]
261    fn iam_debug_renders_fields() {
262        let c = RedshiftCredentials::Iam {
263            region: Some("us-west-2".into()),
264            cluster_identifier: Some("prod-cluster".into()),
265            db_user: Some("analyst".into()),
266        };
267        let dbg = format!("{c:?}");
268        assert!(dbg.contains("Iam"));
269        assert!(dbg.contains("us-west-2"));
270        assert!(dbg.contains("prod-cluster"));
271        assert!(dbg.contains("analyst"));
272    }
273
274    #[test]
275    fn redshift_data_api_debug_renders_fields() {
276        let c = RedshiftCredentials::RedshiftDataApi {
277            region: Some("eu-central-1".into()),
278            cluster_identifier: None,
279            workgroup_name: Some("wg-1".into()),
280            secret_arn: Some("arn:aws:secretsmanager:x".into()),
281            db_user: Some("svc".into()),
282        };
283        let dbg = format!("{c:?}");
284        assert!(dbg.contains("RedshiftDataApi"));
285        assert!(dbg.contains("eu-central-1"));
286        assert!(dbg.contains("wg-1"));
287        assert!(dbg.contains("svc"));
288    }
289
290    #[test]
291    fn iam_and_data_api_round_trip_full_fields() {
292        let iam = r#"{"type":"iam","config":{"region":"us-east-1","cluster_identifier":"c1","db_user":"u"}}"#;
293        let back: RedshiftCredentials = serde_json::from_str(iam).unwrap();
294        assert_eq!(serde_json::to_string(&back).unwrap(), iam);
295
296        let api = r#"{"type":"redshift_data_api","config":{"region":"us-east-1","cluster_identifier":"c1","workgroup_name":"wg","secret_arn":"arn","db_user":"u"}}"#;
297        let back: RedshiftCredentials = serde_json::from_str(api).unwrap();
298        assert_eq!(serde_json::to_string(&back).unwrap(), api);
299    }
300}
faucet_common_redshift/config.rs

faucet_common_redshift/
config.rs
