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;