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    pub host: String,
121    /// Port. Defaults to [`DEFAULT_PORT`] (5439).
122    #[serde(default = "default_port")]
123    pub port: u16,
124    /// Database name.
125    pub database: String,
126    /// Database user.
127    pub user: String,
128    /// Authentication credentials.
129    pub credentials: RedshiftCredentials,
130    /// Whether to require TLS. Defaults to `true` (Redshift clusters require SSL
131    /// by default). `true` maps to `sslmode=require` (encrypted, but the server
132    /// certificate is **not** verified); `false` maps to `sslmode=prefer`
133    /// (opportunistic TLS with plaintext fallback). Ignored when `tls_mode` is
134    /// set.
135    #[serde(default = "default_tls")]
136    pub tls: bool,
137    /// Explicit TLS mode; overrides `tls`. `verify_ca` / `verify_full` check
138    /// the server certificate against `ssl_root_cert` (or the system roots),
139    /// the latter also its host name.
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    pub tls_mode: Option<RedshiftTlsMode>,
142    /// PEM file of the CA that signed the server certificate (the Redshift CA
143    /// bundle), for `tls_mode: verify_ca | verify_full`.
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub ssl_root_cert: Option<String>,
146}
147
148/// How a Redshift connection uses TLS.
149#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
150#[serde(rename_all = "snake_case")]
151pub enum RedshiftTlsMode {
152    /// Plaintext only.
153    Disable,
154    /// TLS when the server offers it, else plaintext.
155    Prefer,
156    /// TLS required; the certificate is not verified.
157    Require,
158    /// TLS required and the certificate chain verified.
159    VerifyCa,
160    /// As `verify_ca`, plus the certificate must name the host.
161    VerifyFull,
162}
163
164impl RedshiftConnection {
165    /// Build a `Password` connection with sensible defaults (port 5439, TLS on).
166    pub fn new(
167        host: impl Into<String>,
168        database: impl Into<String>,
169        user: impl Into<String>,
170        password: impl Into<String>,
171    ) -> Self {
172        Self {
173            host: host.into(),
174            port: DEFAULT_PORT,
175            database: database.into(),
176            user: user.into(),
177            credentials: RedshiftCredentials::Password {
178                password: password.into(),
179            },
180            tls: true,
181            tls_mode: None,
182            ssl_root_cert: None,
183        }
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    #[test]
192    fn debug_masks_password() {
193        let c = RedshiftCredentials::Password {
194            password: "s3cr3t".into(),
195        };
196        let dbg = format!("{c:?}");
197        assert!(dbg.contains("***"));
198        assert!(!dbg.contains("s3cr3t"));
199    }
200
201    #[test]
202    fn connection_debug_does_not_leak_password() {
203        let conn = RedshiftConnection::new("host", "db", "user", "hunter2");
204        let dbg = format!("{conn:?}");
205        assert!(!dbg.contains("hunter2"));
206        assert!(dbg.contains("host"));
207        assert!(dbg.contains("user"));
208    }
209
210    #[test]
211    fn password_credentials_round_trip() {
212        let c = RedshiftCredentials::Password {
213            password: "pw".into(),
214        };
215        let json = serde_json::to_string(&c).unwrap();
216        assert_eq!(json, r#"{"type":"password","config":{"password":"pw"}}"#);
217        let back: RedshiftCredentials = serde_json::from_str(&json).unwrap();
218        assert!(matches!(back, RedshiftCredentials::Password { .. }));
219    }
220
221    #[test]
222    fn connection_defaults_port_and_tls() {
223        let json = r#"{
224            "host": "h",
225            "database": "db",
226            "user": "u",
227            "credentials": {"type": "password", "config": {"password": "pw"}}
228        }"#;
229        let conn: RedshiftConnection = serde_json::from_str(json).unwrap();
230        assert_eq!(conn.port, DEFAULT_PORT);
231        assert!(conn.tls);
232    }
233
234    #[test]
235    fn iam_variant_deserializes() {
236        let json = r#"{"type":"iam","config":{"region":"us-east-1","db_user":"analyst"}}"#;
237        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
238        match c {
239            RedshiftCredentials::Iam {
240                region, db_user, ..
241            } => {
242                assert_eq!(region.as_deref(), Some("us-east-1"));
243                assert_eq!(db_user.as_deref(), Some("analyst"));
244            }
245            _ => panic!("expected Iam"),
246        }
247    }
248
249    #[test]
250    fn redshift_data_api_variant_deserializes() {
251        let json =
252            r#"{"type":"redshift_data_api","config":{"workgroup_name":"wg","secret_arn":"arn:x"}}"#;
253        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
254        assert!(matches!(c, RedshiftCredentials::RedshiftDataApi { .. }));
255    }
256
257    #[test]
258    fn iam_debug_renders_fields() {
259        let c = RedshiftCredentials::Iam {
260            region: Some("us-west-2".into()),
261            cluster_identifier: Some("prod-cluster".into()),
262            db_user: Some("analyst".into()),
263        };
264        let dbg = format!("{c:?}");
265        assert!(dbg.contains("Iam"));
266        assert!(dbg.contains("us-west-2"));
267        assert!(dbg.contains("prod-cluster"));
268        assert!(dbg.contains("analyst"));
269    }
270
271    #[test]
272    fn redshift_data_api_debug_renders_fields() {
273        let c = RedshiftCredentials::RedshiftDataApi {
274            region: Some("eu-central-1".into()),
275            cluster_identifier: None,
276            workgroup_name: Some("wg-1".into()),
277            secret_arn: Some("arn:aws:secretsmanager:x".into()),
278            db_user: Some("svc".into()),
279        };
280        let dbg = format!("{c:?}");
281        assert!(dbg.contains("RedshiftDataApi"));
282        assert!(dbg.contains("eu-central-1"));
283        assert!(dbg.contains("wg-1"));
284        assert!(dbg.contains("svc"));
285    }
286
287    #[test]
288    fn iam_and_data_api_round_trip_full_fields() {
289        let iam = r#"{"type":"iam","config":{"region":"us-east-1","cluster_identifier":"c1","db_user":"u"}}"#;
290        let back: RedshiftCredentials = serde_json::from_str(iam).unwrap();
291        assert_eq!(serde_json::to_string(&back).unwrap(), iam);
292
293        let api = r#"{"type":"redshift_data_api","config":{"region":"us-east-1","cluster_identifier":"c1","workgroup_name":"wg","secret_arn":"arn","db_user":"u"}}"#;
294        let back: RedshiftCredentials = serde_json::from_str(api).unwrap();
295        assert_eq!(serde_json::to_string(&back).unwrap(), api);
296    }
297}
faucet_common_redshift/config.rs

faucet_common_redshift/
config.rs
