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`; `false` maps to
132    /// `sslmode=prefer` (opportunistic TLS with plaintext fallback) — it never
133    /// forbids encryption outright.
134    #[serde(default = "default_tls")]
135    pub tls: bool,
136}
137
138impl RedshiftConnection {
139    /// Build a `Password` connection with sensible defaults (port 5439, TLS on).
140    pub fn new(
141        host: impl Into<String>,
142        database: impl Into<String>,
143        user: impl Into<String>,
144        password: impl Into<String>,
145    ) -> Self {
146        Self {
147            host: host.into(),
148            port: DEFAULT_PORT,
149            database: database.into(),
150            user: user.into(),
151            credentials: RedshiftCredentials::Password {
152                password: password.into(),
153            },
154            tls: true,
155        }
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162
163    #[test]
164    fn debug_masks_password() {
165        let c = RedshiftCredentials::Password {
166            password: "s3cr3t".into(),
167        };
168        let dbg = format!("{c:?}");
169        assert!(dbg.contains("***"));
170        assert!(!dbg.contains("s3cr3t"));
171    }
172
173    #[test]
174    fn connection_debug_does_not_leak_password() {
175        let conn = RedshiftConnection::new("host", "db", "user", "hunter2");
176        let dbg = format!("{conn:?}");
177        assert!(!dbg.contains("hunter2"));
178        assert!(dbg.contains("host"));
179        assert!(dbg.contains("user"));
180    }
181
182    #[test]
183    fn password_credentials_round_trip() {
184        let c = RedshiftCredentials::Password {
185            password: "pw".into(),
186        };
187        let json = serde_json::to_string(&c).unwrap();
188        assert_eq!(json, r#"{"type":"password","config":{"password":"pw"}}"#);
189        let back: RedshiftCredentials = serde_json::from_str(&json).unwrap();
190        assert!(matches!(back, RedshiftCredentials::Password { .. }));
191    }
192
193    #[test]
194    fn connection_defaults_port_and_tls() {
195        let json = r#"{
196            "host": "h",
197            "database": "db",
198            "user": "u",
199            "credentials": {"type": "password", "config": {"password": "pw"}}
200        }"#;
201        let conn: RedshiftConnection = serde_json::from_str(json).unwrap();
202        assert_eq!(conn.port, DEFAULT_PORT);
203        assert!(conn.tls);
204    }
205
206    #[test]
207    fn iam_variant_deserializes() {
208        let json = r#"{"type":"iam","config":{"region":"us-east-1","db_user":"analyst"}}"#;
209        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
210        match c {
211            RedshiftCredentials::Iam {
212                region, db_user, ..
213            } => {
214                assert_eq!(region.as_deref(), Some("us-east-1"));
215                assert_eq!(db_user.as_deref(), Some("analyst"));
216            }
217            _ => panic!("expected Iam"),
218        }
219    }
220
221    #[test]
222    fn redshift_data_api_variant_deserializes() {
223        let json =
224            r#"{"type":"redshift_data_api","config":{"workgroup_name":"wg","secret_arn":"arn:x"}}"#;
225        let c: RedshiftCredentials = serde_json::from_str(json).unwrap();
226        assert!(matches!(c, RedshiftCredentials::RedshiftDataApi { .. }));
227    }
228
229    #[test]
230    fn iam_debug_renders_fields() {
231        let c = RedshiftCredentials::Iam {
232            region: Some("us-west-2".into()),
233            cluster_identifier: Some("prod-cluster".into()),
234            db_user: Some("analyst".into()),
235        };
236        let dbg = format!("{c:?}");
237        assert!(dbg.contains("Iam"));
238        assert!(dbg.contains("us-west-2"));
239        assert!(dbg.contains("prod-cluster"));
240        assert!(dbg.contains("analyst"));
241    }
242
243    #[test]
244    fn redshift_data_api_debug_renders_fields() {
245        let c = RedshiftCredentials::RedshiftDataApi {
246            region: Some("eu-central-1".into()),
247            cluster_identifier: None,
248            workgroup_name: Some("wg-1".into()),
249            secret_arn: Some("arn:aws:secretsmanager:x".into()),
250            db_user: Some("svc".into()),
251        };
252        let dbg = format!("{c:?}");
253        assert!(dbg.contains("RedshiftDataApi"));
254        assert!(dbg.contains("eu-central-1"));
255        assert!(dbg.contains("wg-1"));
256        assert!(dbg.contains("svc"));
257    }
258
259    #[test]
260    fn iam_and_data_api_round_trip_full_fields() {
261        let iam = r#"{"type":"iam","config":{"region":"us-east-1","cluster_identifier":"c1","db_user":"u"}}"#;
262        let back: RedshiftCredentials = serde_json::from_str(iam).unwrap();
263        assert_eq!(serde_json::to_string(&back).unwrap(), iam);
264
265        let api = r#"{"type":"redshift_data_api","config":{"region":"us-east-1","cluster_identifier":"c1","workgroup_name":"wg","secret_arn":"arn","db_user":"u"}}"#;
266        let back: RedshiftCredentials = serde_json::from_str(api).unwrap();
267        assert_eq!(serde_json::to_string(&back).unwrap(), api);
268    }
269}
faucet_common_redshift/config.rs

faucet_common_redshift/
config.rs
