krabka_client_admin/
delegation_tokens.rs

1//! KIP-48 delegation tokens, as Kafka's `Admin.createDelegationToken`,
2//! `renewDelegationToken`, `expireDelegationToken`, and
3//! `describeDelegationToken`.
4//!
5//! Each call takes an options struct that mirrors Kafka's
6//! `CreateDelegationTokenOptions`, `RenewDelegationTokenOptions`,
7//! `ExpireDelegationTokenOptions`, or `DescribeDelegationTokenOptions`, and
8//! sends one request on the admin connection, as `KafkaAdminClient` sends it
9//! to the least-loaded node. A non-zero top-level error code becomes
10//! [`AdminError::Broker`], as Kafka completes the call with
11//! `response.error().exception()`.
12
13use bytes::Bytes;
14use krabka_protocol::owned::{
15    create_delegation_token_request::{CreatableRenewers, CreateDelegationTokenRequest},
16    create_delegation_token_response::CreateDelegationTokenResponse,
17    describe_delegation_token_request::{
18        DescribeDelegationTokenOwner, DescribeDelegationTokenRequest,
19    },
20    describe_delegation_token_response::{
21        DescribeDelegationTokenResponse, DescribedDelegationToken,
22    },
23    expire_delegation_token_request::ExpireDelegationTokenRequest,
24    renew_delegation_token_request::RenewDelegationTokenRequest,
25};
26use krabka_security::KafkaPrincipal;
27use krabka_units::{
28    Time,
29    convert::{TimeExt as _, wire::opt_time_to_millis_i64},
30};
31
32use crate::{AdminClient, AdminError, kafka_error_name};
33
34/// The first `CreateDelegationToken` version with the owner fields
35/// (KIP-373).
36const CREATE_WITH_OWNER_MIN_VERSION: i16 = 3;
37
38/// What [`AdminClientLike::create_delegation_token`](crate::AdminClientLike::create_delegation_token) asks for, as Kafka's
39/// `CreateDelegationTokenOptions`.
40///
41/// The default mints a token that the caller
42/// owns, with no renewers and the broker's maximum lifetime.
43#[derive(Debug, Clone, Default, PartialEq, bon::Builder)]
44pub struct CreateDelegationTokenOptions {
45    /// The owner of the token, of any principal type. `None` makes the
46    /// caller the owner. Another owner needs the `CreateTokens` operation on
47    /// the owner's `User` resource, and `CreateDelegationToken` v3 (KIP-373).
48    pub owner: Option<KafkaPrincipal>,
49    /// The principals that may renew the token.
50    #[builder(default)]
51    pub renewers: Vec<KafkaPrincipal>,
52    /// The maximum lifetime of the token. `None` sends Kafka's `-1`, and the
53    /// broker then uses `delegation.token.max.lifetime.ms`. The broker also
54    /// uses that ceiling for a lifetime of zero or less, and caps a longer
55    /// one at it.
56    #[builder(with = |millis: i64| Time::from_millis(millis))]
57    pub max_lifetime: Option<Time>,
58}
59
60/// What [`AdminClientLike::renew_delegation_token`](crate::AdminClientLike::renew_delegation_token) asks for, as Kafka's
61/// `RenewDelegationTokenOptions`.
62#[derive(Debug, Clone, Copy, Default, PartialEq, bon::Builder)]
63pub struct RenewDelegationTokenOptions {
64    /// How far past now to move the expiry. `None` sends Kafka's `-1`, and
65    /// the broker then uses `delegation.token.expiry.time.ms`. The broker
66    /// also uses that default for a period of zero or less, caps a longer
67    /// one at it, and never moves the expiry past the token's maximum
68    /// timestamp.
69    #[builder(with = |millis: i64| Time::from_millis(millis))]
70    pub renew_time_period: Option<Time>,
71}
72
73/// What [`AdminClientLike::expire_delegation_token`](crate::AdminClientLike::expire_delegation_token) asks for, as Kafka's
74/// `ExpireDelegationTokenOptions`.
75#[derive(Debug, Clone, Copy, Default, PartialEq, bon::Builder)]
76pub struct ExpireDelegationTokenOptions {
77    /// How far past now the token expires. `None` sends Kafka's `-1`, which
78    /// expires the token now, as does any negative period. A period of zero
79    /// or more sets the expiry to that time, capped at the token's maximum
80    /// timestamp.
81    #[builder(with = |millis: i64| Time::from_millis(millis))]
82    pub expiry_time_period: Option<Time>,
83}
84
85/// What [`AdminClientLike::describe_delegation_token`](crate::AdminClientLike::describe_delegation_token) asks for, as Kafka's
86/// `DescribeDelegationTokenOptions`.
87///
88/// The default describes every token that the caller may describe.
89#[derive(Debug, Clone, Default, PartialEq, Eq, bon::Builder)]
90pub struct DescribeDelegationTokenOptions {
91    /// The owners whose tokens to describe. `None` sends a null owner list,
92    /// and the broker returns every token that the caller owns or may
93    /// describe. An empty list sends an empty owner list, and the broker
94    /// returns no tokens.
95    pub owners: Option<Vec<KafkaPrincipal>>,
96}
97
98/// One delegation token, as Kafka's `DelegationToken` with its
99/// `TokenInformation`.
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct DelegationToken {
102    /// The token ID, which a client presents as its SCRAM user name.
103    pub token_id: String,
104    /// The HMAC of the token, which a client presents as its SCRAM password.
105    pub hmac: Vec<u8>,
106    /// The principal that owns the token.
107    pub owner: KafkaPrincipal,
108    /// The principal that asked for the token. It differs from
109    /// [`owner`](Self::owner) when a caller created a token for another
110    /// owner.
111    pub token_requester: KafkaPrincipal,
112    /// The principals that may renew the token.
113    pub renewers: Vec<KafkaPrincipal>,
114    /// When the broker issued the token, in epoch milliseconds.
115    pub issue_timestamp_ms: i64,
116    /// When the token expires unless someone renews it, in epoch
117    /// milliseconds.
118    pub expiry_timestamp_ms: i64,
119    /// The latest expiry that a renewal can set, in epoch milliseconds.
120    pub max_timestamp_ms: i64,
121}
122
123impl AdminClient {
124    /// The call behind [`AdminClientLike::create_delegation_token`](crate::AdminClientLike::create_delegation_token).
125    pub(crate) async fn create_delegation_token_impl(
126        &self,
127        options: &CreateDelegationTokenOptions,
128    ) -> Result<DelegationToken, AdminError> {
129        let request = create_request(options);
130        let response = if options.owner.is_some() {
131            self.conn
132                .send_at_least(request, CREATE_WITH_OWNER_MIN_VERSION)
133                .await?
134        } else {
135            self.conn.send(request).await?
136        };
137        created_token(response, &options.renewers)
138    }
139
140    /// The call behind [`AdminClientLike::renew_delegation_token`](crate::AdminClientLike::renew_delegation_token).
141    pub(crate) async fn renew_delegation_token_impl(
142        &self,
143        hmac: &[u8],
144        options: RenewDelegationTokenOptions,
145    ) -> Result<i64, AdminError> {
146        let response = self.conn.send(renew_request(hmac, options)).await?;
147        check("RenewDelegationToken", response.error_code)?;
148        Ok(response.expiry_timestamp_ms)
149    }
150
151    /// The call behind [`AdminClientLike::expire_delegation_token`](crate::AdminClientLike::expire_delegation_token).
152    pub(crate) async fn expire_delegation_token_impl(
153        &self,
154        hmac: &[u8],
155        options: ExpireDelegationTokenOptions,
156    ) -> Result<i64, AdminError> {
157        let response = self.conn.send(expire_request(hmac, options)).await?;
158        check("ExpireDelegationToken", response.error_code)?;
159        Ok(response.expiry_timestamp_ms)
160    }
161
162    /// The call behind [`AdminClientLike::describe_delegation_token`](crate::AdminClientLike::describe_delegation_token).
163    pub(crate) async fn describe_delegation_token_impl(
164        &self,
165        options: &DescribeDelegationTokenOptions,
166    ) -> Result<Vec<DelegationToken>, AdminError> {
167        let response = self.conn.send(describe_request(options)).await?;
168        described_tokens(response)
169    }
170}
171
172/// The `CreateDelegationToken` request of `options`, as
173/// `KafkaAdminClient.createDelegationToken` builds it. With no owner the
174/// owner fields keep their empty-string defaults, as Kafka's
175/// `CreateDelegationTokenRequestData` does, and the broker makes the caller
176/// the owner.
177fn create_request(options: &CreateDelegationTokenOptions) -> CreateDelegationTokenRequest {
178    let mut request = CreateDelegationTokenRequest {
179        renewers: options
180            .renewers
181            .iter()
182            .map(|renewer| CreatableRenewers {
183                principal_type: renewer.principal_type.clone(),
184                principal_name: renewer.name.clone(),
185                ..Default::default()
186            })
187            .collect(),
188        max_lifetime_ms: opt_time_to_millis_i64(options.max_lifetime),
189        ..Default::default()
190    };
191    if let Some(owner) = &options.owner {
192        request.owner_principal_type = Some(owner.principal_type.clone());
193        request.owner_principal_name = Some(owner.name.clone());
194    }
195    request
196}
197
198fn renew_request(hmac: &[u8], options: RenewDelegationTokenOptions) -> RenewDelegationTokenRequest {
199    RenewDelegationTokenRequest {
200        hmac: Bytes::copy_from_slice(hmac),
201        renew_period_ms: opt_time_to_millis_i64(options.renew_time_period),
202        ..Default::default()
203    }
204}
205
206fn expire_request(
207    hmac: &[u8],
208    options: ExpireDelegationTokenOptions,
209) -> ExpireDelegationTokenRequest {
210    ExpireDelegationTokenRequest {
211        hmac: Bytes::copy_from_slice(hmac),
212        expiry_time_period_ms: opt_time_to_millis_i64(options.expiry_time_period),
213        ..Default::default()
214    }
215}
216
217/// The `DescribeDelegationToken` request of `options`, as Kafka's
218/// `DescribeDelegationTokenRequest.Builder` builds it: a null owner list for
219/// no owners, and the listed owners otherwise.
220fn describe_request(options: &DescribeDelegationTokenOptions) -> DescribeDelegationTokenRequest {
221    DescribeDelegationTokenRequest {
222        owners: options.owners.as_ref().map(|owners| {
223            owners
224                .iter()
225                .map(|owner| DescribeDelegationTokenOwner {
226                    principal_type: owner.principal_type.clone(),
227                    principal_name: owner.name.clone(),
228                    ..Default::default()
229                })
230                .collect()
231        }),
232        ..Default::default()
233    }
234}
235
236/// The token of a `CreateDelegationToken` answer, with the renewers that the
237/// request named.
238fn created_token(
239    response: CreateDelegationTokenResponse,
240    renewers: &[KafkaPrincipal],
241) -> Result<DelegationToken, AdminError> {
242    check("CreateDelegationToken", response.error_code)?;
243    let owner = principal(response.principal_type, response.principal_name);
244    Ok(DelegationToken {
245        token_id: response.token_id,
246        hmac: response.hmac.to_vec(),
247        token_requester: requester(
248            &owner,
249            response.token_requester_principal_type,
250            response.token_requester_principal_name,
251        ),
252        owner,
253        renewers: renewers.to_vec(),
254        issue_timestamp_ms: response.issue_timestamp_ms,
255        expiry_timestamp_ms: response.expiry_timestamp_ms,
256        max_timestamp_ms: response.max_timestamp_ms,
257    })
258}
259
260fn described_tokens(
261    response: DescribeDelegationTokenResponse,
262) -> Result<Vec<DelegationToken>, AdminError> {
263    check("DescribeDelegationToken", response.error_code)?;
264    Ok(response.tokens.into_iter().map(described_token).collect())
265}
266
267fn described_token(token: DescribedDelegationToken) -> DelegationToken {
268    let owner = principal(token.principal_type, token.principal_name);
269    DelegationToken {
270        token_id: token.token_id,
271        hmac: token.hmac.to_vec(),
272        token_requester: requester(
273            &owner,
274            token.token_requester_principal_type,
275            token.token_requester_principal_name,
276        ),
277        owner,
278        renewers: token
279            .renewers
280            .into_iter()
281            .map(|renewer| principal(renewer.principal_type, renewer.principal_name))
282            .collect(),
283        issue_timestamp_ms: token.issue_timestamp,
284        expiry_timestamp_ms: token.expiry_timestamp,
285        max_timestamp_ms: token.max_timestamp,
286    }
287}
288
289/// The requester of a token. The requester fields start at v3; below it they
290/// decode empty, and a token could then only be made by its own owner, so the
291/// owner is the requester.
292fn requester(owner: &KafkaPrincipal, principal_type: String, name: String) -> KafkaPrincipal {
293    if principal_type.is_empty() && name.is_empty() {
294        owner.clone()
295    } else {
296        principal(principal_type, name)
297    }
298}
299
300fn principal(principal_type: String, name: String) -> KafkaPrincipal {
301    KafkaPrincipal {
302        principal_type,
303        name,
304    }
305}
306
307/// `Ok` for a zero error code, and [`AdminError::Broker`] for any other.
308fn check(api: &'static str, code: i16) -> Result<(), AdminError> {
309    if code == 0 {
310        return Ok(());
311    }
312    Err(AdminError::Broker {
313        api,
314        code,
315        name: kafka_error_name(code),
316        message: None,
317    })
318}
319
320#[cfg(test)]
321mod tests {
322    use std::sync::{Arc, Mutex};
323
324    use krabka_client_core::{ClientError, MockBroker};
325    use krabka_protocol::{
326        Decode, Encode,
327        owned::{
328            api_versions_request, create_delegation_token_request,
329            describe_delegation_token_request,
330            describe_delegation_token_response::DescribedDelegationTokenRenewer,
331            expire_delegation_token_request,
332            expire_delegation_token_response::ExpireDelegationTokenResponse,
333            renew_delegation_token_request,
334            renew_delegation_token_response::RenewDelegationTokenResponse,
335        },
336    };
337    use krabka_units::convert::TimeExt;
338
339    use super::*;
340    use crate::{
341        AdminClientLike as _,
342        partition_leaders::test_support::{
343            ApiVersionRange, api_versions, decode_request, encode_response, fast_admin,
344        },
345    };
346
347    fn user(name: &str) -> KafkaPrincipal {
348        principal("User".to_owned(), name.to_owned())
349    }
350
351    fn group(name: &str) -> KafkaPrincipal {
352        principal("Group".to_owned(), name.to_owned())
353    }
354
355    /// Every delegation-token API up to its latest version.
356    const TOKEN_APIS: [ApiVersionRange; 4] = [
357        ApiVersionRange {
358            api_key: create_delegation_token_request::API_KEY,
359            min_version: 1,
360            max_version: 3,
361        },
362        ApiVersionRange {
363            api_key: renew_delegation_token_request::API_KEY,
364            min_version: 1,
365            max_version: 2,
366        },
367        ApiVersionRange {
368            api_key: expire_delegation_token_request::API_KEY,
369            min_version: 1,
370            max_version: 2,
371        },
372        ApiVersionRange {
373            api_key: describe_delegation_token_request::API_KEY,
374            min_version: 1,
375            max_version: 3,
376        },
377    ];
378
379    /// A mock broker that advertises `apis`, answers each `api_key` request
380    /// with `answer` encoded at the negotiated version, and records the
381    /// requests it decodes. Every token API is flexible from v2.
382    async fn token_broker<Req, Resp>(
383        apis: Vec<ApiVersionRange>,
384        api_key: i16,
385        answer: Resp,
386    ) -> (MockBroker, Arc<Mutex<Vec<Req>>>)
387    where
388        Req: for<'a> Decode<'a> + Send + 'static,
389        Resp: Encode + Send + 'static,
390    {
391        let requests = Arc::new(Mutex::new(Vec::new()));
392        let seen = Arc::clone(&requests);
393        let broker = MockBroker::start(move |key, version, _, body| match key {
394            api_versions_request::API_KEY => Some(api_versions().apis(apis.clone()).call()),
395            key if key == api_key => {
396                let flexible = version >= 2;
397                seen.lock()
398                    .expect("requests lock")
399                    .push(decode_request::<Req>(body, version, flexible));
400                Some(encode_response(&answer, version, flexible))
401            }
402            _ => None,
403        })
404        .await
405        .unwrap();
406        (broker, requests)
407    }
408
409    type Outcome<T> = Result<T, (&'static str, i16, &'static str)>;
410
411    fn outcome<T>(result: Result<T, AdminError>) -> Outcome<T> {
412        result.map_err(|error| match error {
413            AdminError::Broker {
414                api,
415                code,
416                name,
417                message: None,
418            } => (api, code, name),
419            other => panic!("unexpected error {other:?}"),
420        })
421    }
422
423    fn created_response(error_code: i16) -> CreateDelegationTokenResponse {
424        CreateDelegationTokenResponse {
425            error_code,
426            principal_type: "Group".to_owned(),
427            principal_name: "ops".to_owned(),
428            token_requester_principal_type: "User".to_owned(),
429            token_requester_principal_name: "admin".to_owned(),
430            issue_timestamp_ms: 1_000,
431            expiry_timestamp_ms: 2_000,
432            max_timestamp_ms: 9_000,
433            token_id: "tok-1".to_owned(),
434            hmac: Bytes::from_static(b"\xde\xad"),
435            ..Default::default()
436        }
437    }
438
439    /// Kafka's `createDelegationToken` sends the renewers and the maximum
440    /// lifetime, sets the owner fields only for an owner, and builds the
441    /// token from the answer and the requested renewers.
442    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
443    async fn create_delegation_token_sends_each_option_as_kafka_does() {
444        let token = |renewers| DelegationToken {
445            token_id: "tok-1".to_owned(),
446            hmac: vec![0xde, 0xad],
447            owner: group("ops"),
448            token_requester: user("admin"),
449            renewers,
450            issue_timestamp_ms: 1_000,
451            expiry_timestamp_ms: 2_000,
452            max_timestamp_ms: 9_000,
453        };
454        let renewers = vec![user("bob"), group("ops")];
455        let wire_renewers = vec![
456            CreatableRenewers {
457                principal_type: "User".to_owned(),
458                principal_name: "bob".to_owned(),
459                ..Default::default()
460            },
461            CreatableRenewers {
462                principal_type: "Group".to_owned(),
463                principal_name: "ops".to_owned(),
464                ..Default::default()
465            },
466        ];
467        for (name, options, request, error_code, expected) in [
468            (
469                "defaults",
470                CreateDelegationTokenOptions::builder().build(),
471                CreateDelegationTokenRequest {
472                    owner_principal_type: Some(String::new()),
473                    owner_principal_name: Some(String::new()),
474                    renewers: Vec::new(),
475                    max_lifetime_ms: -1,
476                    ..Default::default()
477                },
478                0,
479                Ok(token(Vec::new())),
480            ),
481            (
482                "owner of another principal type, renewers, lifetime",
483                CreateDelegationTokenOptions::builder()
484                    .owner(group("ops"))
485                    .renewers(renewers.clone())
486                    .max_lifetime(60_000)
487                    .build(),
488                CreateDelegationTokenRequest {
489                    owner_principal_type: Some("Group".to_owned()),
490                    owner_principal_name: Some("ops".to_owned()),
491                    renewers: wire_renewers.clone(),
492                    max_lifetime_ms: 60_000,
493                    ..Default::default()
494                },
495                0,
496                Ok(token(renewers.clone())),
497            ),
498            (
499                "zero lifetime reaches the broker as zero",
500                CreateDelegationTokenOptions::builder()
501                    .max_lifetime(0)
502                    .build(),
503                CreateDelegationTokenRequest {
504                    max_lifetime_ms: 0,
505                    ..Default::default()
506                },
507                0,
508                Ok(token(Vec::new())),
509            ),
510            (
511                "not authorized to create for the owner",
512                CreateDelegationTokenOptions::builder()
513                    .owner(user("alice"))
514                    .build(),
515                CreateDelegationTokenRequest {
516                    owner_principal_type: Some("User".to_owned()),
517                    owner_principal_name: Some("alice".to_owned()),
518                    max_lifetime_ms: -1,
519                    ..Default::default()
520                },
521                65,
522                Err((
523                    "CreateDelegationToken",
524                    65,
525                    "DELEGATION_TOKEN_AUTHORIZATION_FAILED",
526                )),
527            ),
528            (
529                "renewer of an invalid principal type",
530                CreateDelegationTokenOptions::builder()
531                    .renewers(renewers.clone())
532                    .build(),
533                CreateDelegationTokenRequest {
534                    renewers: wire_renewers.clone(),
535                    max_lifetime_ms: -1,
536                    ..Default::default()
537                },
538                67,
539                Err(("CreateDelegationToken", 67, "INVALID_PRINCIPAL_TYPE")),
540            ),
541        ] {
542            let (broker, requests) = token_broker::<CreateDelegationTokenRequest, _>(
543                TOKEN_APIS.to_vec(),
544                create_delegation_token_request::API_KEY,
545                created_response(error_code),
546            )
547            .await;
548            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
549
550            let result = outcome(admin.create_delegation_token(&options).await);
551
552            broker.stop();
553            let requests = requests.lock().expect("requests lock").clone();
554            assert2::assert!(
555                (result, requests) == (expected, vec![request]),
556                "case {name}"
557            );
558        }
559    }
560
561    /// Each `AdminClientLike` token call reaches the broker and returns the
562    /// result of its `_impl` call.
563    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
564    async fn admin_client_like_forwards_each_token_call() {
565        use crate::AdminClientLike;
566
567        let (create, _) = token_broker::<CreateDelegationTokenRequest, _>(
568            TOKEN_APIS.to_vec(),
569            create_delegation_token_request::API_KEY,
570            created_response(0),
571        )
572        .await;
573        let mut admin = fast_admin(create.addr, krabka_units::secs(5)).await;
574        let created = AdminClientLike::create_delegation_token(
575            &mut admin,
576            &CreateDelegationTokenOptions::builder().build(),
577        )
578        .await
579        .map(|token| token.token_id);
580        create.stop();
581
582        let (renew, _) = token_broker::<RenewDelegationTokenRequest, _>(
583            TOKEN_APIS.to_vec(),
584            renew_delegation_token_request::API_KEY,
585            RenewDelegationTokenResponse {
586                expiry_timestamp_ms: 7_000,
587                ..Default::default()
588            },
589        )
590        .await;
591        let mut admin = fast_admin(renew.addr, krabka_units::secs(5)).await;
592        let renewed = AdminClientLike::renew_delegation_token(
593            &mut admin,
594            b"hmac",
595            RenewDelegationTokenOptions::builder().build(),
596        )
597        .await;
598        renew.stop();
599
600        let (expire, _) = token_broker::<ExpireDelegationTokenRequest, _>(
601            TOKEN_APIS.to_vec(),
602            expire_delegation_token_request::API_KEY,
603            ExpireDelegationTokenResponse {
604                expiry_timestamp_ms: 8_000,
605                ..Default::default()
606            },
607        )
608        .await;
609        let mut admin = fast_admin(expire.addr, krabka_units::secs(5)).await;
610        let expired = AdminClientLike::expire_delegation_token(
611            &mut admin,
612            b"hmac",
613            ExpireDelegationTokenOptions::builder().build(),
614        )
615        .await;
616        expire.stop();
617
618        let (describe, _) = token_broker::<DescribeDelegationTokenRequest, _>(
619            TOKEN_APIS.to_vec(),
620            describe_delegation_token_request::API_KEY,
621            DescribeDelegationTokenResponse::default(),
622        )
623        .await;
624        let mut admin = fast_admin(describe.addr, krabka_units::secs(5)).await;
625        let described = AdminClientLike::describe_delegation_token(
626            &mut admin,
627            &DescribeDelegationTokenOptions::builder().build(),
628        )
629        .await
630        .map(|tokens| tokens.len());
631        describe.stop();
632
633        assert2::assert!(
634            (
635                outcome(created),
636                outcome(renewed),
637                outcome(expired),
638                outcome(described),
639            ) == (Ok("tok-1".to_owned()), Ok(7_000), Ok(8_000), Ok(0))
640        );
641    }
642
643    /// Below v3 the requester fields are absent and decode empty, and a
644    /// token could then only be made by its owner, so the requester of a
645    /// created or described token falls back to the owner. From v3 the answer
646    /// names the requester.
647    #[test]
648    fn a_token_without_requester_fields_names_its_owner_as_requester() {
649        for (name, requester_type, requester_name, expected) in [
650            ("v1 or v2 answer", "", "", user("alice")),
651            ("v3 answer", "User", "admin", user("admin")),
652        ] {
653            let created = created_token(
654                CreateDelegationTokenResponse {
655                    principal_type: "User".to_owned(),
656                    principal_name: "alice".to_owned(),
657                    token_requester_principal_type: requester_type.to_owned(),
658                    token_requester_principal_name: requester_name.to_owned(),
659                    ..Default::default()
660                },
661                &[],
662            )
663            .expect("no error code");
664            let described = described_token(DescribedDelegationToken {
665                principal_type: "User".to_owned(),
666                principal_name: "alice".to_owned(),
667                token_requester_principal_type: requester_type.to_owned(),
668                token_requester_principal_name: requester_name.to_owned(),
669                ..Default::default()
670            });
671            assert2::assert!(
672                (created.token_requester, described.token_requester)
673                    == (expected.clone(), expected),
674                "case {name}"
675            );
676        }
677    }
678
679    /// Kafka's request refuses to write the owner fields below v3, so an
680    /// owner needs v3. Without an owner an older version serves.
681    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
682    async fn create_delegation_token_needs_v3_only_for_an_owner() {
683        for (name, owner, expected) in [
684            ("no owner", None, Ok(1)),
685            (
686                "owner",
687                Some(user("alice")),
688                Err((create_delegation_token_request::API_KEY, 1, 2, 3, 3)),
689            ),
690        ] {
691            let (broker, requests) = token_broker::<CreateDelegationTokenRequest, _>(
692                vec![
693                    ApiVersionRange::builder()
694                        .api_key(create_delegation_token_request::API_KEY)
695                        .min_version(1)
696                        .max_version(2)
697                        .build(),
698                ],
699                create_delegation_token_request::API_KEY,
700                created_response(0),
701            )
702            .await;
703            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
704
705            let result = admin
706                .create_delegation_token(
707                    &CreateDelegationTokenOptions::builder()
708                        .maybe_owner(owner)
709                        .build(),
710                )
711                .await
712                .map(|_| requests.lock().expect("requests lock").len())
713                .map_err(|error| match error {
714                    AdminError::Transport(ClientError::IncompatibleVersion {
715                        api_key,
716                        broker_min,
717                        broker_max,
718                        client_min,
719                        client_max,
720                    }) => (api_key, broker_min, broker_max, client_min, client_max),
721                    other => panic!("case {name}: unexpected error {other:?}"),
722                });
723
724            broker.stop();
725            assert2::assert!(result == expected, "case {name}");
726        }
727    }
728
729    /// Kafka's `renewDelegationToken` sends the renew period, `-1` by
730    /// default, and returns the expiry of the answer.
731    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
732    async fn renew_delegation_token_sends_the_period_and_returns_the_expiry() {
733        for (name, renew_time_period, renew_period_ms, error_code, expected) in [
734            ("broker default", None, -1, 0, Ok(5_000)),
735            (
736                "one hour",
737                Some(krabka_units::secs(3_600)),
738                3_600_000,
739                0,
740                Ok(5_000),
741            ),
742            (
743                "unknown token",
744                None,
745                -1,
746                62,
747                Err(("RenewDelegationToken", 62, "DELEGATION_TOKEN_NOT_FOUND")),
748            ),
749            (
750                "expired token",
751                None,
752                -1,
753                66,
754                Err(("RenewDelegationToken", 66, "DELEGATION_TOKEN_EXPIRED")),
755            ),
756        ] {
757            let (broker, requests) = token_broker::<RenewDelegationTokenRequest, _>(
758                TOKEN_APIS.to_vec(),
759                renew_delegation_token_request::API_KEY,
760                RenewDelegationTokenResponse {
761                    error_code,
762                    expiry_timestamp_ms: 5_000,
763                    ..Default::default()
764                },
765            )
766            .await;
767            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
768
769            let result = outcome(
770                admin
771                    .renew_delegation_token(
772                        b"\x01\x02",
773                        RenewDelegationTokenOptions::builder()
774                            .maybe_renew_time_period(renew_time_period.map(TimeExt::millis_i64))
775                            .build(),
776                    )
777                    .await,
778            );
779
780            broker.stop();
781            let requests = requests.lock().expect("requests lock").clone();
782            assert2::assert!(
783                (result, requests)
784                    == (
785                        expected,
786                        vec![RenewDelegationTokenRequest {
787                            hmac: Bytes::from_static(b"\x01\x02"),
788                            renew_period_ms,
789                            ..Default::default()
790                        }]
791                    ),
792                "case {name}"
793            );
794        }
795    }
796
797    /// Kafka's `expireDelegationToken` sends the expiry period, `-1` (now)
798    /// by default, and returns the expiry of the answer, which
799    /// `kafka-delegation-tokens` prints as the new expiry date.
800    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
801    async fn expire_delegation_token_sends_the_period_and_returns_the_expiry() {
802        for (name, expiry_time_period, expiry_time_period_ms, error_code, expected) in [
803            ("now", None, -1, 0, Ok(7_000)),
804            ("in zero ms", Some(krabka_units::millis(0)), 0, 0, Ok(7_000)),
805            (
806                "in one minute",
807                Some(krabka_units::secs(60)),
808                60_000,
809                0,
810                Ok(7_000),
811            ),
812            (
813                "not a renewer",
814                None,
815                -1,
816                63,
817                Err((
818                    "ExpireDelegationToken",
819                    63,
820                    "DELEGATION_TOKEN_OWNER_MISMATCH",
821                )),
822            ),
823        ] {
824            let (broker, requests) = token_broker::<ExpireDelegationTokenRequest, _>(
825                TOKEN_APIS.to_vec(),
826                expire_delegation_token_request::API_KEY,
827                ExpireDelegationTokenResponse {
828                    error_code,
829                    expiry_timestamp_ms: 7_000,
830                    ..Default::default()
831                },
832            )
833            .await;
834            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
835
836            let result = outcome(
837                admin
838                    .expire_delegation_token(
839                        b"\xaa",
840                        ExpireDelegationTokenOptions::builder()
841                            .maybe_expiry_time_period(expiry_time_period.map(TimeExt::millis_i64))
842                            .build(),
843                    )
844                    .await,
845            );
846
847            broker.stop();
848            let requests = requests.lock().expect("requests lock").clone();
849            assert2::assert!(
850                (result, requests)
851                    == (
852                        expected,
853                        vec![ExpireDelegationTokenRequest {
854                            hmac: Bytes::from_static(b"\xaa"),
855                            expiry_time_period_ms,
856                            ..Default::default()
857                        }]
858                    ),
859                "case {name}"
860            );
861        }
862    }
863
864    /// Kafka's `describeDelegationToken` sends a null owner list for no
865    /// owners, which asks for every token that the caller may describe, and
866    /// an empty list for an empty list, which asks for none. It maps each
867    /// described token with its requester and renewers.
868    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
869    async fn describe_delegation_token_sends_the_owner_filter_as_kafka_does() {
870        let described = DescribedDelegationToken {
871            principal_type: "User".to_owned(),
872            principal_name: "alice".to_owned(),
873            token_requester_principal_type: "User".to_owned(),
874            token_requester_principal_name: "operator".to_owned(),
875            issue_timestamp: 1_000,
876            expiry_timestamp: 2_000,
877            max_timestamp: 9_000,
878            token_id: "tok-1".to_owned(),
879            hmac: Bytes::from_static(b"\xbe\xef"),
880            renewers: vec![DescribedDelegationTokenRenewer {
881                principal_type: "Group".to_owned(),
882                principal_name: "ops".to_owned(),
883                ..Default::default()
884            }],
885            ..Default::default()
886        };
887        let token = DelegationToken {
888            token_id: "tok-1".to_owned(),
889            hmac: vec![0xbe, 0xef],
890            owner: user("alice"),
891            token_requester: user("operator"),
892            renewers: vec![group("ops")],
893            issue_timestamp_ms: 1_000,
894            expiry_timestamp_ms: 2_000,
895            max_timestamp_ms: 9_000,
896        };
897        let owner = |principal_type: &str, principal_name: &str| DescribeDelegationTokenOwner {
898            principal_type: principal_type.to_owned(),
899            principal_name: principal_name.to_owned(),
900            ..Default::default()
901        };
902        for (name, owners, wire_owners, (error_code, tokens), expected) in [
903            (
904                "every token",
905                None,
906                None,
907                (0, vec![described.clone()]),
908                Ok(vec![token.clone()]),
909            ),
910            (
911                "no owner",
912                Some(Vec::new()),
913                Some(Vec::new()),
914                (0, Vec::new()),
915                Ok(Vec::new()),
916            ),
917            (
918                "owners of two principal types",
919                Some(vec![user("alice"), group("ops")]),
920                Some(vec![owner("User", "alice"), owner("Group", "ops")]),
921                (0, vec![described.clone()]),
922                Ok(vec![token.clone()]),
923            ),
924            (
925                "tokens disabled",
926                None,
927                None,
928                (61, Vec::new()),
929                Err((
930                    "DescribeDelegationToken",
931                    61,
932                    "DELEGATION_TOKEN_AUTH_DISABLED",
933                )),
934            ),
935        ] {
936            let (broker, requests) = token_broker::<DescribeDelegationTokenRequest, _>(
937                TOKEN_APIS.to_vec(),
938                describe_delegation_token_request::API_KEY,
939                DescribeDelegationTokenResponse {
940                    error_code,
941                    tokens,
942                    ..Default::default()
943                },
944            )
945            .await;
946            let mut admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
947
948            let result = outcome(
949                admin
950                    .describe_delegation_token(
951                        &DescribeDelegationTokenOptions::builder()
952                            .maybe_owners(owners)
953                            .build(),
954                    )
955                    .await,
956            );
957
958            broker.stop();
959            let requests = requests.lock().expect("requests lock").clone();
960            assert2::assert!(
961                (result, requests)
962                    == (
963                        expected,
964                        vec![DescribeDelegationTokenRequest {
965                            owners: wire_owners,
966                            ..Default::default()
967                        }]
968                    ),
969                "case {name}"
970            );
971        }
972    }
973}
krabka_client_admin/delegation_tokens.rs

krabka_client_admin/
delegation_tokens.rs
