faucet_common_kinesis/
auth.rs

1//! Credential configuration + client construction shared by the Kinesis
2//! source and sink.
3
4use faucet_core::FaucetError;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// How to authenticate with AWS Kinesis.
9///
10/// Serializes as `{ type: <method>, config: { … } }` (adjacent tagging,
11/// snake_case discriminators) — the consistent auth wire shape shared by
12/// every faucet connector.
13#[derive(Clone, Default, Serialize, Deserialize, JsonSchema)]
14#[serde(tag = "type", content = "config", rename_all = "snake_case")]
15pub enum KinesisCredentials {
16    /// The AWS SDK default provider chain: environment variables, shared
17    /// config/credentials files, ECS/EKS container credentials, web-identity
18    /// tokens (`AWS_WEB_IDENTITY_TOKEN_FILE` / EKS IRSA), and EC2 instance
19    /// profiles — with automatic refresh/rotation.
20    #[default]
21    Default,
22    /// A named profile from the shared AWS config/credentials files.
23    Profile {
24        /// Profile name (as in `~/.aws/credentials`).
25        name: String,
26    },
27    /// Static access keys. Prefer `${env:…}` / secrets-manager interpolation
28    /// over literals in config files.
29    AccessKey {
30        /// AWS access key id.
31        access_key_id: String,
32        /// AWS secret access key.
33        secret_access_key: String,
34        /// Optional session token (for temporary credentials).
35        #[serde(default, skip_serializing_if = "Option::is_none")]
36        session_token: Option<String>,
37    },
38    /// Assume an IAM role via STS on top of the default provider chain.
39    AssumeRole {
40        /// ARN of the role to assume.
41        role_arn: String,
42        /// Session name recorded in CloudTrail. Defaults to `faucet-stream`.
43        #[serde(default, skip_serializing_if = "Option::is_none")]
44        session_name: Option<String>,
45        /// Optional external id for cross-account trust policies.
46        #[serde(default, skip_serializing_if = "Option::is_none")]
47        external_id: Option<String>,
48    },
49    /// Web-identity federation (e.g. EKS IRSA). Equivalent to the `default`
50    /// chain, which honors `AWS_WEB_IDENTITY_TOKEN_FILE` + `AWS_ROLE_ARN` —
51    /// kept as an explicit variant so intent is visible in configs.
52    WebIdentity,
53}
54
55impl std::fmt::Debug for KinesisCredentials {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        match self {
58            Self::Default => write!(f, "Default"),
59            Self::Profile { name } => f.debug_struct("Profile").field("name", name).finish(),
60            // Never print key material.
61            Self::AccessKey { access_key_id, .. } => f
62                .debug_struct("AccessKey")
63                .field("access_key_id", access_key_id)
64                .field("secret_access_key", &"***")
65                .finish(),
66            Self::AssumeRole { role_arn, .. } => f
67                .debug_struct("AssumeRole")
68                .field("role_arn", role_arn)
69                .finish(),
70            Self::WebIdentity => write!(f, "WebIdentity"),
71        }
72    }
73}
74
75/// Build an `aws_sdk_kinesis::Client` from region / endpoint / credentials.
76///
77/// `region: None` defers to the SDK default chain (env, profile, IMDS);
78/// `endpoint_url` overrides the endpoint for LocalStack / VPC endpoints.
79/// Credential resolution is delegated to `aws-config`, so rotating
80/// credentials (web identity, instance profiles, assumed roles) refresh
81/// automatically — the client never caches static credentials itself.
82pub async fn build_client(
83    region: Option<&str>,
84    endpoint_url: Option<&str>,
85    credentials: &KinesisCredentials,
86) -> Result<aws_sdk_kinesis::Client, FaucetError> {
87    let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
88    if let Some(region) = region {
89        loader = loader.region(aws_config::Region::new(region.to_owned()));
90    }
91    match credentials {
92        KinesisCredentials::Default | KinesisCredentials::WebIdentity => {}
93        KinesisCredentials::Profile { name } => {
94            loader = loader.profile_name(name);
95        }
96        KinesisCredentials::AccessKey {
97            access_key_id,
98            secret_access_key,
99            session_token,
100        } => {
101            let creds = aws_sdk_kinesis::config::Credentials::new(
102                access_key_id.clone(),
103                secret_access_key.clone(),
104                session_token.clone(),
105                None,
106                "faucet-config",
107            );
108            loader = loader.credentials_provider(creds);
109        }
110        KinesisCredentials::AssumeRole {
111            role_arn,
112            session_name,
113            external_id,
114        } => {
115            let mut builder = aws_config::sts::AssumeRoleProvider::builder(role_arn)
116                .session_name(session_name.as_deref().unwrap_or("faucet-stream"));
117            if let Some(region) = region {
118                builder = builder.region(aws_config::Region::new(region.to_owned()));
119            }
120            if let Some(external_id) = external_id {
121                builder = builder.external_id(external_id);
122            }
123            loader = loader.credentials_provider(builder.build().await);
124        }
125    }
126    let sdk_config = loader.load().await;
127    let mut kinesis_config = aws_sdk_kinesis::config::Builder::from(&sdk_config);
128    if let Some(endpoint) = endpoint_url {
129        kinesis_config = kinesis_config.endpoint_url(endpoint);
130    }
131    Ok(aws_sdk_kinesis::Client::from_conf(kinesis_config.build()))
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    #[test]
139    fn credentials_parse_the_consistent_wire_shape() {
140        let yaml = "type: default\n";
141        let c: KinesisCredentials = serde_yaml::from_str(yaml).unwrap();
142        assert!(matches!(c, KinesisCredentials::Default));
143
144        let yaml = "type: profile\nconfig: { name: prod }\n";
145        let c: KinesisCredentials = serde_yaml::from_str(yaml).unwrap();
146        assert!(matches!(c, KinesisCredentials::Profile { name } if name == "prod"));
147
148        let yaml =
149            "type: access_key\nconfig:\n  access_key_id: AKIA\n  secret_access_key: s3cr3t\n";
150        let c: KinesisCredentials = serde_yaml::from_str(yaml).unwrap();
151        match &c {
152            KinesisCredentials::AccessKey {
153                access_key_id,
154                session_token,
155                ..
156            } => {
157                assert_eq!(access_key_id, "AKIA");
158                assert!(session_token.is_none());
159            }
160            other => panic!("unexpected: {other:?}"),
161        }
162        // Debug never leaks the secret.
163        let dbg = format!("{c:?}");
164        assert!(!dbg.contains("s3cr3t"), "{dbg}");
165
166        let yaml = "type: assume_role\nconfig: { role_arn: 'arn:aws:iam::1:role/x' }\n";
167        let c: KinesisCredentials = serde_yaml::from_str(yaml).unwrap();
168        assert!(matches!(c, KinesisCredentials::AssumeRole { .. }));
169
170        let yaml = "type: web_identity\n";
171        let c: KinesisCredentials = serde_yaml::from_str(yaml).unwrap();
172        assert!(matches!(c, KinesisCredentials::WebIdentity));
173    }
174
175    #[test]
176    fn default_is_default() {
177        assert!(matches!(
178            KinesisCredentials::default(),
179            KinesisCredentials::Default
180        ));
181    }
182
183    #[tokio::test]
184    async fn build_client_honors_endpoint_and_static_keys() {
185        // Offline: constructing the client performs no network I/O.
186        let creds = KinesisCredentials::AccessKey {
187            access_key_id: "test".into(),
188            secret_access_key: "test".into(),
189            session_token: None,
190        };
191        let client = build_client(Some("us-east-1"), Some("http://127.0.0.1:4566"), &creds)
192            .await
193            .expect("client builds");
194        assert_eq!(
195            client.config().region().map(|r| r.as_ref()),
196            Some("us-east-1")
197        );
198
199        // Every other variant also constructs offline (no network I/O).
200        for creds in [
201            KinesisCredentials::Default,
202            KinesisCredentials::WebIdentity,
203            KinesisCredentials::Profile {
204                name: "no-such-profile".into(),
205            },
206            KinesisCredentials::AssumeRole {
207                role_arn: "arn:aws:iam::123456789012:role/x".into(),
208                session_name: Some("t".into()),
209                external_id: Some("e".into()),
210            },
211        ] {
212            build_client(Some("us-east-1"), None, &creds)
213                .await
214                .expect("client builds offline");
215        }
216    }
217}
faucet_common_kinesis/auth.rs

faucet_common_kinesis/
auth.rs
