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

faucet_common_redshift/
config.rs
