krabka_client_admin/
users.rs

1//! `KafkaUser` admin RPCs.
2//!
3//! Five admin operations the `KafkaUser` reconciler drives:
4//! `AlterUserScramCredentials` (upsert + delete in a single call),
5//! `CreateAcls`, `DeleteAcls`, `DescribeAcls`.
6//!
7//! Wire `i8` discriminants stay private to this module. Callers use the typed
8//! Rust enums below. `crates/client-admin` depends on `krabka-metadata` for
9//! shared image types such as `DelegationToken`, but it stays free of
10//! `krabka-broker`, so out-of-process clients can use the crate. Unit tests
11//! cover the wire round-trip of the local enum copies.
12
13use bytes::Bytes;
14#[cfg(test)]
15use krabka_protocol::owned::alter_user_scram_credentials_response;
16#[cfg(test)]
17use krabka_protocol::owned::describe_user_scram_credentials_response;
18use krabka_protocol::owned::{
19    alter_user_scram_credentials_request::{
20        AlterUserScramCredentialsRequest, ScramCredentialDeletion, ScramCredentialUpsertion,
21    },
22    create_acls_request::{AclCreation, CreateAclsRequest},
23    delete_acls_request::{DeleteAclsFilter, DeleteAclsRequest},
24    describe_acls_request::DescribeAclsRequest,
25    describe_user_scram_credentials_request::{DescribeUserScramCredentialsRequest, UserName},
26};
27use krabka_security::SaslMechanism;
28use ring::rand::{SecureRandom, SystemRandom};
29
30use crate::{AdminClient, AdminError, KafkaError, kafka_error_if, kafka_error_name};
31
32/// KIP-554 wire byte for SCRAM-SHA-512. SHA-256 is byte `1`.
33const SCRAM_SHA_512_WIRE: i8 = 2;
34/// SCRAM mechanism byte for SCRAM-SHA-256 (1, KIP-554).
35/// Paired with the `*_sha256` builders/helpers below.
36const SCRAM_SHA_256_WIRE: i8 = 1;
37
38pub use krabka_security::scram::{
39    DEFAULT_SCRAM_ITERATIONS, MAX_SCRAM_ITERATIONS, MIN_SCRAM_ITERATIONS, ScramIterations,
40};
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
43pub enum ResourceType {
44    Topic,
45    Group,
46    Cluster,
47    TransactionalId,
48    /// A delegation token, named by its token id (KIP-48).
49    DelegationToken,
50    /// A user principal, for `CreateTokens` and `DescribeTokens` (KIP-373).
51    User,
52}
53
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
55pub enum PatternType {
56    /// Filter only: matches every stored pattern that applies to the filter's
57    /// resource name, the literal name, `*`, and every matching prefix, as
58    /// Kafka's `PatternType.MATCH` does. A stored ACL is never `Match`.
59    Match,
60    Literal,
61    Prefixed,
62}
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
65pub enum PermissionType {
66    Allow,
67    Deny,
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
71pub enum AclOperation {
72    All,
73    Read,
74    Write,
75    Create,
76    Delete,
77    Alter,
78    Describe,
79    ClusterAction,
80    DescribeConfigs,
81    AlterConfigs,
82    IdempotentWrite,
83    /// KIP-373: permission to create delegation tokens for other users.
84    CreateTokens,
85    /// KIP-373: permission to describe the delegation tokens of other users.
86    DescribeTokens,
87    /// KIP-939: 2PC participation permission on a `TransactionalId`.
88    TwoPhaseCommit,
89}
90
91/// A concrete, non-filter, ACL entry with every field populated. It matches
92/// the shape the broker stores in its metadata image.
93#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
94pub struct AclEntry {
95    pub resource_type: ResourceType,
96    pub resource_name: String,
97    pub pattern_type: PatternType,
98    pub principal: String,
99    pub host: String,
100    pub operation: AclOperation,
101    pub permission_type: PermissionType,
102}
103
104/// Filter for `DescribeAcls` / `DeleteAcls`. Each `None` axis matches
105/// anything.
106#[derive(Debug, Clone, Default, PartialEq, Eq, bon::Builder)]
107pub struct AclEntryFilter {
108    pub resource_type: Option<ResourceType>,
109    #[builder(into)]
110    pub resource_name: Option<String>,
111    pub pattern_type: Option<PatternType>,
112    #[builder(into)]
113    pub principal: Option<String>,
114    #[builder(into)]
115    pub host: Option<String>,
116    pub operation: Option<AclOperation>,
117    pub permission_type: Option<PermissionType>,
118}
119
120#[derive(Debug, Clone)]
121pub struct ScramUpsertion {
122    pub username: String,
123    pub password: String,
124    pub iterations: i32,
125}
126
127#[derive(Debug, Clone)]
128pub struct ScramDeletion {
129    pub username: String,
130}
131
132#[derive(Debug, Clone)]
133pub struct ScramUserOutcome {
134    pub username: String,
135    pub error: Option<KafkaError>,
136}
137
138#[derive(Debug, Clone, PartialEq, Eq)]
139pub struct UserScramCredential {
140    pub mechanism: String,
141    pub iterations: i32,
142}
143
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct UserScramCredentials {
146    pub username: String,
147    pub credentials: Vec<UserScramCredential>,
148    pub error: Option<KafkaError>,
149}
150
151#[derive(Debug, Clone)]
152pub struct CreateAclOutcome {
153    pub error: Option<KafkaError>,
154}
155
156#[derive(Debug, Clone)]
157pub struct DeleteAclFilterOutcome {
158    pub error: Option<KafkaError>,
159    pub matched: Vec<AclEntry>,
160}
161
162impl AdminClient {
163    /// The call behind [`AdminClientLike::alter_user_scram_credentials_sha512`](crate::AdminClientLike::alter_user_scram_credentials_sha512).
164    pub(crate) async fn alter_user_scram_credentials_sha512_impl(
165        &self,
166        upsertions: &[ScramUpsertion],
167        deletions: &[ScramDeletion],
168    ) -> Result<Vec<ScramUserOutcome>, AdminError> {
169        let rng = SystemRandom::new();
170        let req = build_alter_scram_request_sha512(upsertions, deletions, &rng)?;
171        let resp = self.conn.send(req).await?;
172        Ok(parse_alter_scram_results(resp))
173    }
174
175    /// The call behind [`AdminClientLike::alter_user_scram_credentials_sha256`](crate::AdminClientLike::alter_user_scram_credentials_sha256).
176    pub(crate) async fn alter_user_scram_credentials_sha256_impl(
177        &self,
178        upsertions: &[ScramUpsertion],
179        deletions: &[ScramDeletion],
180    ) -> Result<Vec<ScramUserOutcome>, AdminError> {
181        let rng = SystemRandom::new();
182        let req = build_alter_scram_request_sha256(upsertions, deletions, &rng)?;
183        let resp = self.conn.send(req).await?;
184        Ok(parse_alter_scram_results(resp))
185    }
186
187    /// The call behind [`AdminClientLike::describe_acls`](crate::AdminClientLike::describe_acls).
188    pub(crate) async fn describe_acls_impl(
189        &self,
190        filter: &AclEntryFilter,
191    ) -> Result<Vec<AclEntry>, AdminError> {
192        let resp = self.conn.send(filter_to_describe_request(filter)).await?;
193        parse_describe_acls(resp)
194    }
195
196    /// # Errors
197    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
198    pub async fn describe_user_scram_credentials(
199        &mut self,
200        users: Option<&[String]>,
201    ) -> Result<Vec<UserScramCredentials>, AdminError> {
202        let req = DescribeUserScramCredentialsRequest {
203            users: users.map(|users| {
204                users
205                    .iter()
206                    .map(|name| UserName {
207                        name: name.clone(),
208                        ..Default::default()
209                    })
210                    .collect()
211            }),
212            ..Default::default()
213        };
214        let resp = self.conn.send(req).await?;
215
216        parse_describe_user_scram_credentials_response(resp)
217    }
218
219    /// The call behind [`AdminClientLike::create_acls`](crate::AdminClientLike::create_acls).
220    pub(crate) async fn create_acls_impl(
221        &self,
222        creations: &[AclEntry],
223    ) -> Result<Vec<CreateAclOutcome>, AdminError> {
224        check_concrete_patterns(creations)?;
225        let req = CreateAclsRequest {
226            creations: creations.iter().map(acl_to_creation).collect(),
227            ..Default::default()
228        };
229        let resp = self.conn.send(req).await?;
230        Ok(resp
231            .results
232            .into_iter()
233            .map(|r| CreateAclOutcome {
234                error: kafka_error_if(r.error_code, r.error_message),
235            })
236            .collect())
237    }
238
239    /// The call behind [`AdminClientLike::delete_acls`](crate::AdminClientLike::delete_acls).
240    pub(crate) async fn delete_acls_impl(
241        &self,
242        filters: &[AclEntryFilter],
243    ) -> Result<Vec<DeleteAclFilterOutcome>, AdminError> {
244        let req = DeleteAclsRequest {
245            filters: filters.iter().map(acl_filter_to_wire).collect(),
246            ..Default::default()
247        };
248        let resp = self.conn.send(req).await?;
249        let mut out = Vec::with_capacity(resp.filter_results.len());
250        for fr in resp.filter_results {
251            if let Some(err) = kafka_error_if(fr.error_code, fr.error_message) {
252                out.push(DeleteAclFilterOutcome {
253                    error: Some(err),
254                    matched: Vec::new(),
255                });
256                continue;
257            }
258            let mut matched = Vec::with_capacity(fr.matching_acls.len());
259            for m in fr.matching_acls {
260                if m.error_code != 0 {
261                    // Per-row deletion error — bubble it up as the
262                    // filter-level error so the reconciler retries.
263                    return Err(AdminError::Broker {
264                        api: "DeleteAcls",
265                        code: m.error_code,
266                        name: kafka_error_name(m.error_code),
267                        message: m.error_message,
268                    });
269                }
270                matched.push(AclEntry {
271                    resource_type: wire_to_resource_type(m.resource_type)?,
272                    resource_name: m.resource_name,
273                    pattern_type: wire_to_stored_pattern_type(m.pattern_type)?,
274                    principal: m.principal,
275                    host: m.host,
276                    operation: wire_to_operation(m.operation)?,
277                    permission_type: wire_to_permission(m.permission_type)?,
278                });
279            }
280            out.push(DeleteAclFilterOutcome {
281                error: None,
282                matched,
283            });
284        }
285        Ok(out)
286    }
287}
288
289fn build_alter_scram_request_sha512(
290    upsertions: &[ScramUpsertion],
291    deletions: &[ScramDeletion],
292    rng: &SystemRandom,
293) -> Result<AlterUserScramCredentialsRequest, AdminError> {
294    build_alter_scram_request(
295        upsertions,
296        deletions,
297        rng,
298        SaslMechanism::ScramSha512,
299        SCRAM_SHA_512_WIRE,
300    )
301}
302
303/// SCRAM-SHA-256 sibling of [`build_alter_scram_request_sha512`]. It shares
304/// [`build_alter_scram_request`], so the two helpers cannot drift.
305fn build_alter_scram_request_sha256(
306    upsertions: &[ScramUpsertion],
307    deletions: &[ScramDeletion],
308    rng: &SystemRandom,
309) -> Result<AlterUserScramCredentialsRequest, AdminError> {
310    build_alter_scram_request(
311        upsertions,
312        deletions,
313        rng,
314        SaslMechanism::ScramSha256,
315        SCRAM_SHA_256_WIRE,
316    )
317}
318
319fn build_alter_scram_request(
320    upsertions: &[ScramUpsertion],
321    deletions: &[ScramDeletion],
322    rng: &SystemRandom,
323    mechanism: SaslMechanism,
324    wire_mechanism: i8,
325) -> Result<AlterUserScramCredentialsRequest, AdminError> {
326    let mut wire_upserts = Vec::with_capacity(upsertions.len());
327    for u in upsertions {
328        let mut salt = vec![0u8; 16];
329        rng.fill(&mut salt)
330            .map_err(|_| AdminError::Protocol("system RNG failure".into()))?;
331        let salted = krabka_security::pbkdf2_salted(
332            u.password.as_bytes(),
333            mechanism,
334            u32::try_from(u.iterations.max(0)).unwrap_or(0),
335            &salt,
336        );
337        wire_upserts.push(ScramCredentialUpsertion {
338            name: u.username.clone(),
339            mechanism: wire_mechanism,
340            iterations: u.iterations,
341            salt: Bytes::from(salt),
342            salted_password: Bytes::from(salted),
343            ..Default::default()
344        });
345    }
346    let wire_deletes = deletions
347        .iter()
348        .map(|d| ScramCredentialDeletion {
349            name: d.username.clone(),
350            mechanism: wire_mechanism,
351            ..Default::default()
352        })
353        .collect();
354    Ok(AlterUserScramCredentialsRequest {
355        upsertions: wire_upserts,
356        deletions: wire_deletes,
357        ..Default::default()
358    })
359}
360
361fn parse_alter_scram_results(
362    resp: <AlterUserScramCredentialsRequest as krabka_protocol::ProtocolRequest>::Response,
363) -> Vec<ScramUserOutcome> {
364    resp.results
365        .into_iter()
366        .map(|r| ScramUserOutcome {
367            username: r.user,
368            error: kafka_error_if(r.error_code, r.error_message),
369        })
370        .collect()
371}
372
373fn parse_describe_user_scram_credentials_response(
374    resp: <DescribeUserScramCredentialsRequest as krabka_protocol::ProtocolRequest>::Response,
375) -> Result<Vec<UserScramCredentials>, AdminError> {
376    if resp.error_code != 0 {
377        return Err(AdminError::Broker {
378            api: "DescribeUserScramCredentials",
379            code: resp.error_code,
380            name: kafka_error_name(resp.error_code),
381            message: resp.error_message,
382        });
383    }
384
385    Ok(resp
386        .results
387        .into_iter()
388        .map(|result| UserScramCredentials {
389            username: result.user,
390            credentials: result
391                .credential_infos
392                .into_iter()
393                .map(|credential| UserScramCredential {
394                    mechanism: scram_mechanism_name(credential.mechanism).to_owned(),
395                    iterations: credential.iterations,
396                })
397                .collect(),
398            error: kafka_error_if(result.error_code, result.error_message),
399        })
400        .collect())
401}
402
403fn scram_mechanism_name(mechanism: i8) -> &'static str {
404    match mechanism {
405        SCRAM_SHA_256_WIRE => "SCRAM-SHA-256",
406        SCRAM_SHA_512_WIRE => "SCRAM-SHA-512",
407        _ => "UNKNOWN",
408    }
409}
410
411fn filter_to_describe_request(f: &AclEntryFilter) -> DescribeAclsRequest {
412    let wire = acl_filter_wire_fields(f);
413    DescribeAclsRequest {
414        resource_type_filter: wire.resource_type_filter,
415        resource_name_filter: wire.resource_name_filter,
416        pattern_type_filter: wire.pattern_type_filter,
417        principal_filter: wire.principal_filter,
418        host_filter: wire.host_filter,
419        operation: wire.operation,
420        permission_type: wire.permission_type,
421        ..Default::default()
422    }
423}
424
425fn parse_describe_acls(
426    resp: <DescribeAclsRequest as krabka_protocol::ProtocolRequest>::Response,
427) -> Result<Vec<AclEntry>, AdminError> {
428    if resp.error_code != 0 {
429        return Err(AdminError::Broker {
430            api: "DescribeAcls",
431            code: resp.error_code,
432            name: kafka_error_name(resp.error_code),
433            message: resp.error_message,
434        });
435    }
436    let mut out = Vec::new();
437    for resource in resp.resources {
438        let rt = wire_to_resource_type(resource.resource_type)?;
439        let pt = wire_to_stored_pattern_type(resource.pattern_type)?;
440        for desc in resource.acls {
441            out.push(AclEntry {
442                resource_type: rt,
443                resource_name: resource.resource_name.clone(),
444                pattern_type: pt,
445                principal: desc.principal,
446                host: desc.host,
447                operation: wire_to_operation(desc.operation)?,
448                permission_type: wire_to_permission(desc.permission_type)?,
449            });
450        }
451    }
452    Ok(out)
453}
454
455/// Refuses a stored ACL with the filter-only [`PatternType::Match`], with the
456/// message of Kafka's `ResourcePattern` constructor.
457fn check_concrete_patterns(entries: &[AclEntry]) -> Result<(), AdminError> {
458    if entries
459        .iter()
460        .any(|entry| entry.pattern_type == PatternType::Match)
461    {
462        return Err(AdminError::InvalidArgument(
463            "patternType must not be MATCH".into(),
464        ));
465    }
466    Ok(())
467}
468
469/// Pure function. It serializes an `AclEntry` to the wire representation that
470/// `CreateAcls` expects.
471pub(crate) fn acl_to_creation(e: &AclEntry) -> AclCreation {
472    AclCreation {
473        resource_type: resource_type_to_wire(e.resource_type),
474        resource_name: e.resource_name.clone(),
475        resource_pattern_type: pattern_type_to_wire(e.pattern_type),
476        principal: e.principal.clone(),
477        host: e.host.clone(),
478        operation: operation_to_wire(e.operation),
479        permission_type: permission_to_wire(e.permission_type),
480        ..Default::default()
481    }
482}
483
484/// Pure function. It serializes an `AclEntryFilter` to the wire `DeleteAcls`
485/// filter. A `None` axis uses the wire ANY discriminant.
486pub(crate) fn acl_filter_to_wire(f: &AclEntryFilter) -> DeleteAclsFilter {
487    let wire = acl_filter_wire_fields(f);
488    DeleteAclsFilter {
489        resource_type_filter: wire.resource_type_filter,
490        resource_name_filter: wire.resource_name_filter,
491        pattern_type_filter: wire.pattern_type_filter,
492        principal_filter: wire.principal_filter,
493        host_filter: wire.host_filter,
494        operation: wire.operation,
495        permission_type: wire.permission_type,
496        ..Default::default()
497    }
498}
499
500struct AclFilterWireFields {
501    resource_type_filter: i8,
502    resource_name_filter: Option<String>,
503    pattern_type_filter: i8,
504    principal_filter: Option<String>,
505    host_filter: Option<String>,
506    operation: i8,
507    permission_type: i8,
508}
509
510fn acl_filter_wire_fields(f: &AclEntryFilter) -> AclFilterWireFields {
511    AclFilterWireFields {
512        resource_type_filter: f.resource_type.map_or(WIRE_ANY, resource_type_to_wire),
513        resource_name_filter: f.resource_name.clone(),
514        pattern_type_filter: f.pattern_type.map_or(WIRE_ANY, pattern_type_to_wire),
515        principal_filter: f.principal.clone(),
516        host_filter: f.host.clone(),
517        operation: f.operation.map_or(WIRE_ANY, operation_to_wire),
518        permission_type: f.permission_type.map_or(WIRE_ANY, permission_to_wire),
519    }
520}
521
522// --- wire constants & enum encoding/decoding -----------------------------
523//
524// Kept private to this module; the broker has its own copy in
525// `krabka_broker::handlers::acl_wire`. Round-trip tests below lock the
526// encoding against the values Kafka's protocol-spec docs publish.
527
528/// Kafka `AclBindingFilter.ANY` discriminant. Filter requests use it as the
529/// "match anything" placeholder.
530const WIRE_ANY: i8 = 1;
531
532macro_rules! acl_wire_enum {
533    ($to_wire:ident, $from_wire:ident, $ty:ty, $unknown:literal, {$($variant:path => $wire:literal),+ $(,)?}) => {
534        pub(crate) fn $to_wire(value: $ty) -> i8 {
535            match value {
536                $($variant => $wire,)+
537            }
538        }
539
540        pub(crate) fn $from_wire(value: i8) -> Result<$ty, AdminError> {
541            match value {
542                $($wire => Ok($variant),)+
543                _ => Err(AdminError::Protocol(format!(concat!($unknown, ": {}"), value))),
544            }
545        }
546    };
547}
548
549acl_wire_enum!(
550    resource_type_to_wire,
551    wire_to_resource_type,
552    ResourceType,
553    "unknown ACL resource_type discriminant",
554    {
555        ResourceType::Topic => 2,
556        ResourceType::Group => 3,
557        ResourceType::Cluster => 4,
558        ResourceType::TransactionalId => 5,
559        ResourceType::DelegationToken => 6,
560        ResourceType::User => 7,
561    }
562);
563
564acl_wire_enum!(
565    pattern_type_to_wire,
566    wire_to_pattern_type,
567    PatternType,
568    "unknown ACL pattern_type discriminant",
569    {
570        PatternType::Match => 2,
571        PatternType::Literal => 3,
572        PatternType::Prefixed => 4,
573    }
574);
575
576/// Decodes the pattern type of a stored ACL in a `DescribeAcls` or
577/// `DeleteAcls` response. `MATCH` (2) is a filter value that no stored ACL
578/// carries, so it is refused like ANY.
579fn wire_to_stored_pattern_type(value: i8) -> Result<PatternType, AdminError> {
580    match wire_to_pattern_type(value)? {
581        PatternType::Match => Err(AdminError::Protocol(format!(
582            "unknown ACL pattern_type discriminant: {value}"
583        ))),
584        stored => Ok(stored),
585    }
586}
587
588acl_wire_enum!(
589    permission_to_wire,
590    wire_to_permission,
591    PermissionType,
592    "unknown ACL permission discriminant",
593    {
594        PermissionType::Deny => 2,
595        PermissionType::Allow => 3,
596    }
597);
598
599acl_wire_enum!(
600    operation_to_wire,
601    wire_to_operation,
602    AclOperation,
603    "unknown ACL operation discriminant",
604    {
605        AclOperation::All => 2,
606        AclOperation::Read => 3,
607        AclOperation::Write => 4,
608        AclOperation::Create => 5,
609        AclOperation::Delete => 6,
610        AclOperation::Alter => 7,
611        AclOperation::Describe => 8,
612        AclOperation::ClusterAction => 9,
613        AclOperation::DescribeConfigs => 10,
614        AclOperation::AlterConfigs => 11,
615        AclOperation::IdempotentWrite => 12,
616        AclOperation::CreateTokens => 13,
617        AclOperation::DescribeTokens => 14,
618        AclOperation::TwoPhaseCommit => 15,
619    }
620);
621
622#[cfg(test)]
623mod tests {
624    use std::sync::{Arc, Mutex};
625
626    use assert2::check;
627    use bytes::{Buf, BytesMut};
628    use krabka_client_core::MockBroker;
629    use krabka_protocol::{
630        Decode, Encode, UnknownTaggedFields,
631        owned::{
632            alter_user_scram_credentials_request,
633            alter_user_scram_credentials_response::{
634                AlterUserScramCredentialsResponse, AlterUserScramCredentialsResult,
635            },
636            api_versions_request,
637            api_versions_response::{ApiVersion, ApiVersionsResponse},
638            create_acls_request,
639            create_acls_response::{AclCreationResult, CreateAclsResponse},
640            delete_acls_request,
641            delete_acls_response::{
642                DeleteAclsFilterResult, DeleteAclsMatchingAcl, DeleteAclsResponse,
643            },
644        },
645    };
646
647    use super::*;
648    use crate::AdminClientLike as _;
649
650    #[test]
651    fn scram_iterations_match_broker_bounds() {
652        assert2::assert!(ScramIterations::default().into_value() == DEFAULT_SCRAM_ITERATIONS);
653        assert2::assert!(ScramIterations::new(4_096).is_ok());
654        assert2::assert!(ScramIterations::new(16_384).is_ok());
655        for invalid in [0, 4_095, 16_385, i32::MAX] {
656            assert2::assert!(ScramIterations::new(invalid).is_err());
657        }
658        assert2::assert!("8192".parse::<ScramIterations>().is_ok());
659        assert2::assert!("not-a-number".parse::<ScramIterations>().is_err());
660    }
661
662    fn encode_v0(resp: &impl Encode) -> Vec<u8> {
663        encode_at(resp, 0)
664    }
665
666    fn encode_at(resp: &impl Encode, version: i16) -> Vec<u8> {
667        let mut buf = BytesMut::new();
668        resp.encode(&mut buf, version).unwrap();
669        buf.to_vec()
670    }
671
672    fn api_versions_response(api_key: i16, version: i16) -> Vec<u8> {
673        encode_v0(&ApiVersionsResponse {
674            api_keys: vec![
675                ApiVersion {
676                    api_key: api_versions_request::API_KEY,
677                    min_version: 0,
678                    max_version: 0,
679                    ..Default::default()
680                },
681                ApiVersion {
682                    api_key,
683                    min_version: version,
684                    max_version: version,
685                    ..Default::default()
686                },
687            ],
688            ..Default::default()
689        })
690    }
691
692    fn request_body_after_header(mut body: &[u8], flexible_header: bool) -> &[u8] {
693        let client_id_len = body.get_i16();
694        assert2::assert!(client_id_len >= 0);
695        body.advance(usize::try_from(client_id_len).expect("client id length is non-negative"));
696        if flexible_header {
697            assert2::assert!(body.get_u8() == 0);
698        }
699        body
700    }
701
702    fn sample_entry() -> AclEntry {
703        AclEntry {
704            resource_type: ResourceType::Topic,
705            resource_name: "orders".into(),
706            pattern_type: PatternType::Literal,
707            principal: "User:alice".into(),
708            host: "*".into(),
709            operation: AclOperation::Read,
710            permission_type: PermissionType::Allow,
711        }
712    }
713
714    #[test]
715    fn resource_type_round_trips() {
716        for (_name, rt) in [
717            ("topic", ResourceType::Topic),
718            ("group", ResourceType::Group),
719            ("cluster", ResourceType::Cluster),
720            ("transactional id", ResourceType::TransactionalId),
721            ("delegation token", ResourceType::DelegationToken),
722            ("user", ResourceType::User),
723        ] {
724            assert2::assert!(wire_to_resource_type(resource_type_to_wire(rt)).unwrap() == rt);
725        }
726    }
727
728    #[test]
729    fn pattern_type_round_trips() {
730        for (_name, pt) in [
731            ("match", PatternType::Match),
732            ("literal", PatternType::Literal),
733            ("prefixed", PatternType::Prefixed),
734        ] {
735            assert2::assert!(wire_to_pattern_type(pattern_type_to_wire(pt)).unwrap() == pt);
736        }
737    }
738
739    #[test]
740    fn permission_round_trips() {
741        for (_name, permission) in [
742            ("allow", PermissionType::Allow),
743            ("deny", PermissionType::Deny),
744        ] {
745            assert2::assert!(
746                wire_to_permission(permission_to_wire(permission)).unwrap() == permission
747            );
748        }
749    }
750
751    #[test]
752    fn operation_round_trips() {
753        for (_name, operation) in [
754            ("all", AclOperation::All),
755            ("read", AclOperation::Read),
756            ("write", AclOperation::Write),
757            ("create", AclOperation::Create),
758            ("delete", AclOperation::Delete),
759            ("alter", AclOperation::Alter),
760            ("describe", AclOperation::Describe),
761            ("cluster action", AclOperation::ClusterAction),
762            ("describe configs", AclOperation::DescribeConfigs),
763            ("alter configs", AclOperation::AlterConfigs),
764            ("idempotent write", AclOperation::IdempotentWrite),
765            ("create tokens", AclOperation::CreateTokens),
766            ("describe tokens", AclOperation::DescribeTokens),
767            // KIP-939: TWO_PHASE_COMMIT (wire byte 15).
768            ("two phase commit", AclOperation::TwoPhaseCommit),
769        ] {
770            assert2::assert!(wire_to_operation(operation_to_wire(operation)).unwrap() == operation);
771        }
772    }
773
774    #[test]
775    fn a_stored_acl_never_decodes_as_match() {
776        for (wire, expected) in [
777            (2, None),
778            (3, Some(PatternType::Literal)),
779            (4, Some(PatternType::Prefixed)),
780        ] {
781            check!(wire_to_stored_pattern_type(wire).ok() == expected);
782        }
783    }
784
785    #[test]
786    fn wire_to_unknown_resource_type_errors() {
787        assert2::assert!(matches!(
788            wire_to_resource_type(99),
789            Err(AdminError::Protocol(_))
790        ));
791        // ANY (1) is intentionally rejected on the concrete decoder so
792        // `DescribeAcls`/`DeleteAcls` responses can never silently
793        // claim an "any-type" match — Kafka never returns this in real
794        // responses.
795        assert2::assert!(matches!(
796            wire_to_resource_type(1),
797            Err(AdminError::Protocol(_))
798        ));
799    }
800
801    #[test]
802    fn a_match_pattern_cannot_be_stored() {
803        let cases = [
804            (PatternType::Literal, Ok(())),
805            (PatternType::Prefixed, Ok(())),
806            (
807                PatternType::Match,
808                Err("patternType must not be MATCH".to_owned()),
809            ),
810        ];
811        for (pattern_type, expected) in cases {
812            let entry = AclEntry {
813                pattern_type,
814                ..sample_entry()
815            };
816            let actual =
817                check_concrete_patterns(&[sample_entry(), entry]).map_err(|error| match error {
818                    AdminError::InvalidArgument(message) => message,
819                    other => format!("{other:?}"),
820                });
821            check!(actual == expected);
822        }
823    }
824
825    #[test]
826    fn acl_to_creation_matches_discriminants() {
827        let e = AclEntry {
828            pattern_type: PatternType::Prefixed,
829            ..sample_entry()
830        };
831        let c = acl_to_creation(&e);
832        assert2::assert!(
833            c == AclCreation {
834                resource_type: 2,
835                resource_name: "orders".to_owned(),
836                resource_pattern_type: 4,
837                principal: "User:alice".to_owned(),
838                host: "*".to_owned(),
839                operation: 3,
840                permission_type: 3,
841                unknown_tagged_fields: UnknownTaggedFields(vec![]),
842            }
843        );
844    }
845
846    /// The SCRAM-SHA-512 and SCRAM-SHA-256 calls send their own mechanism
847    /// (2 and 1, KIP-554) for each upsertion and deletion, and map the result
848    /// of each user.
849    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
850    async fn alter_user_scram_credentials_sends_the_mechanism_of_the_call() {
851        for (name, sha512, wire_mechanism) in [("SHA-512", true, 2_i8), ("SHA-256", false, 1)] {
852            let seen = Arc::new(Mutex::new(Vec::new()));
853            let captured = Arc::clone(&seen);
854            let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
855                if api_key == api_versions_request::API_KEY {
856                    return Some(api_versions_response(
857                        alter_user_scram_credentials_request::API_KEY,
858                        0,
859                    ));
860                }
861                if api_key != alter_user_scram_credentials_request::API_KEY {
862                    return None;
863                }
864                let mut body = request_body_after_header(body, true);
865                let request = AlterUserScramCredentialsRequest::decode(&mut body, version)
866                    .expect("alter SCRAM request decodes");
867                captured.lock().expect("request capture lock").extend(
868                    request
869                        .upsertions
870                        .iter()
871                        .map(|upsertion| (upsertion.name.clone(), upsertion.mechanism))
872                        .chain(
873                            request
874                                .deletions
875                                .iter()
876                                .map(|deletion| (deletion.name.clone(), deletion.mechanism)),
877                        ),
878                );
879                let mut response = vec![0];
880                response.extend(encode_at(
881                    &AlterUserScramCredentialsResponse {
882                        results: vec![
883                            AlterUserScramCredentialsResult {
884                                user: "alice".into(),
885                                ..Default::default()
886                            },
887                            AlterUserScramCredentialsResult {
888                                user: "bob".into(),
889                                error_code: 69,
890                                error_message: Some("no credential".into()),
891                                ..Default::default()
892                            },
893                        ],
894                        ..Default::default()
895                    },
896                    version,
897                ));
898                Some(response)
899            })
900            .await
901            .unwrap();
902            let mut admin = AdminClient::connect(&[mock.addr.to_string()])
903                .await
904                .expect("admin connects to mock broker");
905            let upsertions = [ScramUpsertion {
906                username: "alice".into(),
907                password: "secret".into(),
908                iterations: DEFAULT_SCRAM_ITERATIONS,
909            }];
910            let deletions = [ScramDeletion {
911                username: "bob".into(),
912            }];
913
914            let outcomes = if sha512 {
915                admin
916                    .alter_user_scram_credentials_sha512(&upsertions, &deletions)
917                    .await
918            } else {
919                admin
920                    .alter_user_scram_credentials_sha256(&upsertions, &deletions)
921                    .await
922            }
923            .expect("alter SCRAM response maps");
924
925            mock.stop();
926            let outcomes = outcomes
927                .into_iter()
928                .map(|outcome| (outcome.username, outcome.error.map(|error| error.code)))
929                .collect::<Vec<_>>();
930            let sent = seen.lock().expect("request capture lock").clone();
931            check!(
932                (outcomes, sent)
933                    == (
934                        vec![("alice".to_owned(), None), ("bob".to_owned(), Some(69))],
935                        vec![
936                            ("alice".to_owned(), wire_mechanism),
937                            ("bob".to_owned(), wire_mechanism),
938                        ],
939                    ),
940                "{name}"
941            );
942        }
943    }
944
945    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
946    async fn create_acls_maps_non_empty_broker_results() {
947        let seen_request = Arc::new(Mutex::new(None));
948        let captured_request = Arc::clone(&seen_request);
949        let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
950            if api_key == api_versions_request::API_KEY {
951                return Some(api_versions_response(create_acls_request::API_KEY, 1));
952            }
953            if api_key == create_acls_request::API_KEY {
954                let mut body =
955                    request_body_after_header(body, version >= create_acls_request::FLEXIBLE_MIN);
956                let request = CreateAclsRequest::decode(&mut body, version)
957                    .expect("create ACLs request decodes");
958                assert2::assert!(body.is_empty());
959                *captured_request.lock().expect("request capture lock") = Some(request);
960                return Some(encode_at(
961                    &CreateAclsResponse {
962                        results: vec![AclCreationResult {
963                            error_code: 36,
964                            error_message: Some("acl exists".into()),
965                            ..Default::default()
966                        }],
967                        ..Default::default()
968                    },
969                    1,
970                ));
971            }
972            None
973        })
974        .await
975        .unwrap();
976        let mut admin = AdminClient::connect(&[mock.addr.to_string()])
977            .await
978            .expect("admin connects to mock broker");
979        let creation = AclEntry {
980            pattern_type: PatternType::Prefixed,
981            ..sample_entry()
982        };
983
984        let outcomes = admin
985            .create_acls(&[creation])
986            .await
987            .expect("create ACL response maps");
988
989        let error = outcomes[0]
990            .error
991            .as_ref()
992            .expect("broker error is surfaced");
993        assert2::assert!(
994            (outcomes.len(), error.code, error.message.as_deref()) == (1, 36, Some("acl exists"))
995        );
996        let request = seen_request
997            .lock()
998            .expect("request capture lock")
999            .take()
1000            .expect("create ACLs request was captured");
1001        assert2::assert!(
1002            request
1003                == CreateAclsRequest {
1004                    creations: vec![AclCreation {
1005                        resource_type: resource_type_to_wire(ResourceType::Topic),
1006                        resource_name: "orders".into(),
1007                        resource_pattern_type: pattern_type_to_wire(PatternType::Prefixed),
1008                        principal: "User:alice".into(),
1009                        host: "*".into(),
1010                        operation: operation_to_wire(AclOperation::Read),
1011                        permission_type: permission_to_wire(PermissionType::Allow),
1012                        ..Default::default()
1013                    }],
1014                    ..Default::default()
1015                }
1016        );
1017        mock.stop();
1018    }
1019
1020    #[test]
1021    fn acl_filter_to_wire_uses_any_for_none_axes() {
1022        let f = AclEntryFilter::builder().build();
1023        let w = acl_filter_to_wire(&f);
1024        // 1 == Kafka's `AclBindingFilter.ANY` discriminant (`WIRE_ANY`).
1025        assert2::assert!(
1026            w == DeleteAclsFilter {
1027                resource_type_filter: 1,
1028                resource_name_filter: None,
1029                pattern_type_filter: 1,
1030                principal_filter: None,
1031                host_filter: None,
1032                operation: 1,
1033                permission_type: 1,
1034                unknown_tagged_fields: UnknownTaggedFields(vec![]),
1035            }
1036        );
1037    }
1038
1039    #[test]
1040    fn acl_filter_to_wire_passes_concrete_axes_through() {
1041        let f = AclEntryFilter::builder()
1042            .resource_type(ResourceType::Topic)
1043            .resource_name("orders")
1044            .pattern_type(PatternType::Literal)
1045            .principal("User:alice")
1046            .host("10.0.0.0")
1047            .operation(AclOperation::Read)
1048            .permission_type(PermissionType::Allow)
1049            .build();
1050        let w = acl_filter_to_wire(&f);
1051        assert2::assert!(
1052            w == DeleteAclsFilter {
1053                resource_type_filter: 2,
1054                resource_name_filter: Some("orders".to_owned()),
1055                pattern_type_filter: 3,
1056                principal_filter: Some("User:alice".to_owned()),
1057                host_filter: Some("10.0.0.0".to_owned()),
1058                operation: 3,
1059                permission_type: 3,
1060                unknown_tagged_fields: UnknownTaggedFields(vec![]),
1061            }
1062        );
1063    }
1064
1065    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1066    async fn delete_acls_maps_non_empty_matched_acls() {
1067        let seen_request = Arc::new(Mutex::new(None));
1068        let captured_request = Arc::clone(&seen_request);
1069        let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
1070            if api_key == api_versions_request::API_KEY {
1071                return Some(api_versions_response(delete_acls_request::API_KEY, 1));
1072            }
1073            if api_key == delete_acls_request::API_KEY {
1074                let mut body =
1075                    request_body_after_header(body, version >= delete_acls_request::FLEXIBLE_MIN);
1076                let request = DeleteAclsRequest::decode(&mut body, version)
1077                    .expect("delete ACLs request decodes");
1078                assert2::assert!(body.is_empty());
1079                *captured_request.lock().expect("request capture lock") = Some(request);
1080                return Some(encode_at(
1081                    &DeleteAclsResponse {
1082                        filter_results: vec![DeleteAclsFilterResult {
1083                            matching_acls: vec![DeleteAclsMatchingAcl {
1084                                resource_type: resource_type_to_wire(ResourceType::Topic),
1085                                resource_name: "orders".into(),
1086                                pattern_type: pattern_type_to_wire(PatternType::Literal),
1087                                principal: "User:alice".into(),
1088                                host: "*".into(),
1089                                operation: operation_to_wire(AclOperation::Read),
1090                                permission_type: permission_to_wire(PermissionType::Allow),
1091                                ..Default::default()
1092                            }],
1093                            ..Default::default()
1094                        }],
1095                        ..Default::default()
1096                    },
1097                    1,
1098                ));
1099            }
1100            None
1101        })
1102        .await
1103        .unwrap();
1104        let mut admin = AdminClient::connect(&[mock.addr.to_string()])
1105            .await
1106            .expect("admin connects to mock broker");
1107        let filter = AclEntryFilter::builder()
1108            .resource_type(ResourceType::Topic)
1109            .resource_name("orders")
1110            .build();
1111
1112        let outcomes = admin
1113            .delete_acls(&[filter])
1114            .await
1115            .expect("delete ACL response maps");
1116
1117        assert2::assert!(
1118            (
1119                outcomes.len(),
1120                outcomes[0].error.as_ref(),
1121                outcomes[0].matched.as_slice()
1122            ) == (1, None, [sample_entry()].as_slice())
1123        );
1124        let request = seen_request
1125            .lock()
1126            .expect("request capture lock")
1127            .take()
1128            .expect("delete ACLs request was captured");
1129        assert2::assert!(
1130            request
1131                == DeleteAclsRequest {
1132                    filters: vec![DeleteAclsFilter {
1133                        resource_type_filter: resource_type_to_wire(ResourceType::Topic),
1134                        resource_name_filter: Some("orders".into()),
1135                        pattern_type_filter: WIRE_ANY,
1136                        principal_filter: None,
1137                        host_filter: None,
1138                        operation: WIRE_ANY,
1139                        permission_type: WIRE_ANY,
1140                        ..Default::default()
1141                    }],
1142                    ..Default::default()
1143                }
1144        );
1145        mock.stop();
1146    }
1147
1148    #[test]
1149    fn parse_alter_scram_results_preserves_usernames_and_errors() {
1150        let outcomes = parse_alter_scram_results(AlterUserScramCredentialsResponse {
1151            results: vec![AlterUserScramCredentialsResult {
1152                user: "alice".into(),
1153                error_code: 42,
1154                error_message: Some("bad credentials".into()),
1155                ..Default::default()
1156            }],
1157            ..Default::default()
1158        });
1159
1160        let error = outcomes[0].error.as_ref().expect("error is preserved");
1161        assert2::assert!(
1162            (
1163                outcomes.len(),
1164                outcomes[0].username.as_str(),
1165                error.code,
1166                error.message.as_deref()
1167            ) == (1, "alice", 42, Some("bad credentials"))
1168        );
1169    }
1170
1171    #[test]
1172    fn scram_request_carries_pbkdf2_intermediate_not_password() {
1173        let rng = SystemRandom::new();
1174        let upserts = [ScramUpsertion {
1175            username: "alice".into(),
1176            password: "hunter2".into(),
1177            iterations: 4096,
1178        }];
1179        let req = build_alter_scram_request_sha512(&upserts, &[], &rng).unwrap();
1180        let u = &req.upsertions[0];
1181        check!(
1182            (
1183                req.upsertions.len(),
1184                u.name.as_str(),
1185                u.mechanism,
1186                u.iterations,
1187                u.salt.len(),
1188                u.salted_password.len(),
1189                u.salted_password.as_ref() != b"hunter2",
1190            ) == (1, "alice", 2, 4096, 16, 64, true)
1191        );
1192    }
1193
1194    #[test]
1195    fn scram_request_deletions_use_sha512_mechanism() {
1196        let rng = SystemRandom::new();
1197        let dels = [ScramDeletion {
1198            username: "alice".into(),
1199        }];
1200        let req = build_alter_scram_request_sha512(&[], &dels, &rng).unwrap();
1201        assert2::assert!(
1202            req == AlterUserScramCredentialsRequest {
1203                deletions: vec![ScramCredentialDeletion {
1204                    name: "alice".to_owned(),
1205                    mechanism: 2, // SCRAM_SHA_512_WIRE, KIP-554
1206                    unknown_tagged_fields: UnknownTaggedFields(vec![]),
1207                }],
1208                upsertions: vec![],
1209                unknown_tagged_fields: UnknownTaggedFields(vec![]),
1210            }
1211        );
1212    }
1213
1214    #[test]
1215    fn scram_request_two_upserts_get_distinct_salts() {
1216        let rng = SystemRandom::new();
1217        let upserts = [
1218            ScramUpsertion {
1219                username: "alice".into(),
1220                password: "p".into(),
1221                iterations: 4096,
1222            },
1223            ScramUpsertion {
1224                username: "bob".into(),
1225                password: "p".into(),
1226                iterations: 4096,
1227            },
1228        ];
1229        let req = build_alter_scram_request_sha512(&upserts, &[], &rng).unwrap();
1230        assert2::assert!(req.upsertions[0].salt != req.upsertions[1].salt);
1231    }
1232
1233    #[test]
1234    fn describe_request_uses_any_for_unspecified_axes() {
1235        let f = AclEntryFilter::builder().principal("User:alice").build();
1236        let r = filter_to_describe_request(&f);
1237        // 1 == Kafka's `AclBindingFilter.ANY` discriminant (`WIRE_ANY`).
1238        assert2::assert!(
1239            r == DescribeAclsRequest {
1240                resource_type_filter: 1,
1241                resource_name_filter: None,
1242                pattern_type_filter: 1,
1243                principal_filter: Some("User:alice".to_owned()),
1244                host_filter: None,
1245                operation: 1,
1246                permission_type: 1,
1247                unknown_tagged_fields: UnknownTaggedFields(vec![]),
1248            }
1249        );
1250    }
1251
1252    #[test]
1253    fn users_describe_scram_top_level_error_returns_broker_error() {
1254        let resp = describe_user_scram_credentials_response::DescribeUserScramCredentialsResponse {
1255            error_code: 31,
1256            error_message: Some("cluster auth denied".to_owned()),
1257            ..Default::default()
1258        };
1259
1260        let err = parse_describe_user_scram_credentials_response(resp)
1261            .expect_err("top-level SCRAM describe errors must fail the request");
1262
1263        match err {
1264            AdminError::Broker {
1265                api,
1266                code,
1267                name,
1268                message,
1269            } => {
1270                check!(
1271                    (api, code, name, message.as_deref())
1272                        == (
1273                            "DescribeUserScramCredentials",
1274                            31,
1275                            "CLUSTER_AUTHORIZATION_FAILED",
1276                            Some("cluster auth denied"),
1277                        )
1278                );
1279            }
1280            other => panic!("expected broker error, got {other:?}"),
1281        }
1282    }
1283
1284    #[test]
1285    fn users_describe_scram_preserves_credential_iterations() {
1286        let resp = describe_user_scram_credentials_response::DescribeUserScramCredentialsResponse {
1287            results: vec![
1288                describe_user_scram_credentials_response::DescribeUserScramCredentialsResult {
1289                    user: "alice".to_owned(),
1290                    credential_infos: vec![
1291                        describe_user_scram_credentials_response::CredentialInfo {
1292                            mechanism: 1,
1293                            iterations: 4096,
1294                            ..Default::default()
1295                        },
1296                        describe_user_scram_credentials_response::CredentialInfo {
1297                            mechanism: 2,
1298                            iterations: 8192,
1299                            ..Default::default()
1300                        },
1301                    ],
1302                    ..Default::default()
1303                },
1304            ],
1305            ..Default::default()
1306        };
1307
1308        let users = parse_describe_user_scram_credentials_response(resp)
1309            .expect("valid SCRAM describe response should parse");
1310
1311        check!(
1312            (
1313                users.len(),
1314                users[0].username.as_str(),
1315                users[0].credentials.as_slice(),
1316            ) == (
1317                1,
1318                "alice",
1319                [
1320                    UserScramCredential {
1321                        mechanism: "SCRAM-SHA-256".to_owned(),
1322                        iterations: 4096,
1323                    },
1324                    UserScramCredential {
1325                        mechanism: "SCRAM-SHA-512".to_owned(),
1326                        iterations: 8192,
1327                    },
1328                ]
1329                .as_slice(),
1330            )
1331        );
1332    }
1333
1334    #[test]
1335    fn users_describe_scram_unknown_user_preserves_resource_not_found_name() {
1336        let resp = describe_user_scram_credentials_response::DescribeUserScramCredentialsResponse {
1337            results: vec![
1338                describe_user_scram_credentials_response::DescribeUserScramCredentialsResult {
1339                    user: "ghost".to_owned(),
1340                    error_code: 91,
1341                    error_message: Some("no such SCRAM user".to_owned()),
1342                    ..Default::default()
1343                },
1344            ],
1345            ..Default::default()
1346        };
1347
1348        let users = parse_describe_user_scram_credentials_response(resp)
1349            .expect("per-user SCRAM describe errors should parse into rows");
1350
1351        check!(
1352            (
1353                users.len(),
1354                users[0].username.as_str(),
1355                users[0].error.as_ref()
1356            ) == (
1357                1,
1358                "ghost",
1359                Some(&KafkaError {
1360                    code: 91,
1361                    name: "RESOURCE_NOT_FOUND",
1362                    message: Some("no such SCRAM user".to_owned()),
1363                })
1364            )
1365        );
1366    }
1367
1368    #[test]
1369    fn users_describe_scram_duplicate_user_preserves_duplicate_resource_name() {
1370        let resp = describe_user_scram_credentials_response::DescribeUserScramCredentialsResponse {
1371            results: vec![
1372                describe_user_scram_credentials_response::DescribeUserScramCredentialsResult {
1373                    user: "alice".to_owned(),
1374                    error_code: 92,
1375                    error_message: Some("duplicate user".to_owned()),
1376                    ..Default::default()
1377                },
1378            ],
1379            ..Default::default()
1380        };
1381
1382        let users = parse_describe_user_scram_credentials_response(resp)
1383            .expect("per-user SCRAM describe errors should parse into rows");
1384
1385        check!(
1386            (
1387                users.len(),
1388                users[0].username.as_str(),
1389                users[0].error.as_ref()
1390            ) == (
1391                1,
1392                "alice",
1393                Some(&KafkaError {
1394                    code: 92,
1395                    name: "DUPLICATE_RESOURCE",
1396                    message: Some("duplicate user".to_owned()),
1397                })
1398            )
1399        );
1400    }
1401
1402    #[test]
1403    fn users_alter_scram_preserves_scram_error_names() {
1404        let resp = alter_user_scram_credentials_response::AlterUserScramCredentialsResponse {
1405            results: vec![
1406                alter_user_scram_credentials_response::AlterUserScramCredentialsResult {
1407                    user: "alice".to_owned(),
1408                    error_code: 33,
1409                    error_message: Some("unknown mechanism".to_owned()),
1410                    ..Default::default()
1411                },
1412                alter_user_scram_credentials_response::AlterUserScramCredentialsResult {
1413                    user: "bob".to_owned(),
1414                    error_code: 93,
1415                    error_message: Some("too many iterations".to_owned()),
1416                    ..Default::default()
1417                },
1418            ],
1419            ..Default::default()
1420        };
1421
1422        let users = parse_alter_scram_results(resp);
1423
1424        check!(
1425            (
1426                users.len(),
1427                users[0].error.as_ref(),
1428                users[1].error.as_ref(),
1429            ) == (
1430                2,
1431                Some(&KafkaError {
1432                    code: 33,
1433                    name: "UNSUPPORTED_SASL_MECHANISM",
1434                    message: Some("unknown mechanism".to_owned()),
1435                }),
1436                Some(&KafkaError {
1437                    code: 93,
1438                    name: "UNACCEPTABLE_CREDENTIAL",
1439                    message: Some("too many iterations".to_owned()),
1440                }),
1441            )
1442        );
1443    }
1444
1445    #[test]
1446    fn describe_request_passes_resource_name_and_host_filters_through() {
1447        let f = AclEntryFilter::builder()
1448            .resource_type(ResourceType::Topic)
1449            .resource_name("orders")
1450            .pattern_type(PatternType::Literal)
1451            .principal("User:alice")
1452            .host("10.0.0.0")
1453            .operation(AclOperation::Read)
1454            .permission_type(PermissionType::Allow)
1455            .build();
1456
1457        let r = filter_to_describe_request(&f);
1458
1459        assert2::assert!(
1460            r == DescribeAclsRequest {
1461                resource_type_filter: 2,
1462                resource_name_filter: Some("orders".to_owned()),
1463                pattern_type_filter: 3,
1464                principal_filter: Some("User:alice".to_owned()),
1465                host_filter: Some("10.0.0.0".to_owned()),
1466                operation: 3,
1467                permission_type: 3,
1468                unknown_tagged_fields: UnknownTaggedFields(vec![]),
1469            }
1470        );
1471    }
1472}
krabka_client_admin/users.rs

krabka_client_admin/
users.rs
