|
| 1 | +//! Minimal `SaslMechanismPlugin` reference: a static-token `OAUTHBEARER` plugin. |
| 2 | +//! |
| 3 | +//! This example shows the smallest useful implementation of the |
| 4 | +//! [`SaslMechanismPlugin`] extension trait: hand a static JWT (or any opaque |
| 5 | +//! bearer token) to the `KafkaClient`, and the handshake completes against |
| 6 | +//! any broker configured to accept that token. |
| 7 | +//! |
| 8 | +//! Real-world plugins typically wrap a token-refresh cache — an OIDC client, |
| 9 | +//! an AWS MSK IAM presigner, a Keycloak adapter, etc. The canonical |
| 10 | +//! production implementations live in `kafka-backup-enterprise-core` and |
| 11 | +//! cover MSK IAM (`aws-sigv4`), Confluent Cloud, and OIDC. |
| 12 | +//! |
| 13 | +//! Run it with: |
| 14 | +//! |
| 15 | +//! ```bash |
| 16 | +//! cargo build --example custom_sasl_plugin -p kafka-backup-core |
| 17 | +//! ``` |
| 18 | +//! |
| 19 | +//! The example does not actually connect to a broker — it just proves the |
| 20 | +//! wiring compiles against the public trait surface. |
| 21 | +
|
| 22 | +use std::sync::Arc; |
| 23 | + |
| 24 | +use async_trait::async_trait; |
| 25 | +use kafka_backup_core::config::{KafkaConfig, SecurityConfig, SecurityProtocol, TopicSelection}; |
| 26 | +use kafka_backup_core::kafka::{ |
| 27 | + KafkaClient, SaslMechanismPlugin, SaslMechanismPluginHandle, SaslPluginError, |
| 28 | +}; |
| 29 | + |
| 30 | +/// A static-token OAUTHBEARER plugin. |
| 31 | +/// |
| 32 | +/// Constructs an RFC 7628 client-initial-response payload from a fixed |
| 33 | +/// principal and bearer token. Suitable for local testing against brokers |
| 34 | +/// running the Apache Kafka `OAuthBearerUnsecuredValidatorCallbackHandler` |
| 35 | +/// (unsecured-JWS mode) or any broker that accepts static tokens. |
| 36 | +#[derive(Debug)] |
| 37 | +struct StaticTokenOauthBearer { |
| 38 | + principal: String, |
| 39 | + token: String, |
| 40 | +} |
| 41 | + |
| 42 | +impl StaticTokenOauthBearer { |
| 43 | + fn new(principal: impl Into<String>, token: impl Into<String>) -> Self { |
| 44 | + Self { |
| 45 | + principal: principal.into(), |
| 46 | + token: token.into(), |
| 47 | + } |
| 48 | + } |
| 49 | + |
| 50 | + fn into_handle(self) -> SaslMechanismPluginHandle { |
| 51 | + Arc::new(self) |
| 52 | + } |
| 53 | +} |
| 54 | + |
| 55 | +#[async_trait] |
| 56 | +impl SaslMechanismPlugin for StaticTokenOauthBearer { |
| 57 | + fn mechanism_name(&self) -> &str { |
| 58 | + "OAUTHBEARER" |
| 59 | + } |
| 60 | + |
| 61 | + async fn initial_payload(&self) -> Result<Vec<u8>, SaslPluginError> { |
| 62 | + Ok(build_rfc7628_cir(&self.principal, &self.token)) |
| 63 | + } |
| 64 | +} |
| 65 | + |
| 66 | +/// Build an RFC 7628 §3.1 client initial response: |
| 67 | +/// `n,a=<authzid>,<0x01>auth=Bearer <token><0x01><0x01>`. |
| 68 | +fn build_rfc7628_cir(principal: &str, token: &str) -> Vec<u8> { |
| 69 | + let mut buf = Vec::with_capacity(principal.len() + token.len() + 16); |
| 70 | + buf.extend_from_slice(b"n,a="); |
| 71 | + buf.extend_from_slice(principal.as_bytes()); |
| 72 | + buf.push(b','); |
| 73 | + buf.push(0x01); |
| 74 | + buf.extend_from_slice(b"auth=Bearer "); |
| 75 | + buf.extend_from_slice(token.as_bytes()); |
| 76 | + buf.push(0x01); |
| 77 | + buf.push(0x01); |
| 78 | + buf |
| 79 | +} |
| 80 | + |
| 81 | +fn main() { |
| 82 | + // An unsecured JWT (`alg: none`) for local testing. Production tokens |
| 83 | + // come from an IdP — Okta, Keycloak, AWS STS, etc. |
| 84 | + let unsecured_jwt = concat!( |
| 85 | + "eyJhbGciOiJub25lIn0", |
| 86 | + ".", |
| 87 | + "eyJzdWIiOiJ0ZXN0LXVzZXIiLCJleHAiOjk5OTk5OTk5OTksImlhdCI6MTAwMH0", |
| 88 | + ".", |
| 89 | + ); |
| 90 | + |
| 91 | + let plugin = StaticTokenOauthBearer::new("test-user", unsecured_jwt).into_handle(); |
| 92 | + |
| 93 | + let config = KafkaConfig { |
| 94 | + bootstrap_servers: vec!["localhost:9097".to_string()], |
| 95 | + security: SecurityConfig { |
| 96 | + security_protocol: SecurityProtocol::SaslPlaintext, |
| 97 | + sasl_mechanism_plugin: Some(plugin), |
| 98 | + ..Default::default() |
| 99 | + }, |
| 100 | + topics: TopicSelection::default(), |
| 101 | + connection: Default::default(), |
| 102 | + }; |
| 103 | + |
| 104 | + let _client = KafkaClient::new(config); |
| 105 | + println!( |
| 106 | + "OAUTHBEARER plugin wired into KafkaClient. \ |
| 107 | + Connect would go to localhost:9097 — skipped (no broker in this example)." |
| 108 | + ); |
| 109 | +} |
0 commit comments