krabka_authz/
cache.rs

1//! Gateway-side ACL snapshot: a flat `Vec<AclEntry>` from `describe_acls`.
2//!
3//! The snapshot implements [`AclSource`] with EXACTLY the broker's matching
4//! semantics. It indexes the entries once, when the snapshot is built, so a
5//! lookup neither allocates nor scans entries of other resources: a LITERAL
6//! name and the `*` wildcard are hash lookups, and the PREFIXED entries that
7//! apply to a name are found by probing each distinct stored prefix length.
8
9use {
10    crate::{AclEntries, AclSource},
11    krabka_metadata::{AclEntry, PatternType, ResourceType},
12    std::{collections::HashMap, iter},
13};
14
15/// The resource name of Kafka's `WILDCARD_RESOURCE`, a LITERAL pattern that
16/// matches every resource of its type.
17const WILDCARD_RESOURCE: &str = "*";
18
19/// Immutable ACL snapshot. Each refresh rebuilds it wholesale.
20#[derive(Debug, Clone, Default)]
21pub struct AclCache {
22    entries: Vec<AclEntry>,
23    by_type: HashMap<ResourceType, TypeIndex>,
24    cidr_hosts_supported: bool,
25}
26
27/// Positions in [`AclCache::entries`] of the entries of one resource type.
28#[derive(Debug, Clone, Default)]
29struct TypeIndex {
30    all: Vec<usize>,
31    literal: HashMap<String, Vec<usize>>,
32    prefixed: HashMap<String, Vec<usize>>,
33    /// The distinct byte lengths of the keys of `prefixed`, ascending.
34    prefix_lens: Vec<usize>,
35}
36
37impl TypeIndex {
38    fn literal(&self, name: &str) -> impl Iterator<Item = usize> + '_ {
39        self.literal.get(name).into_iter().flatten().copied()
40    }
41
42    /// Every PREFIXED entry whose resource name `name` starts with. A stored
43    /// prefix is valid UTF-8, so it ends on a character boundary of every
44    /// name it prefixes; `str::get` skips the lengths that are not one.
45    fn prefixed<'a>(&'a self, name: &'a str) -> impl Iterator<Item = usize> + 'a {
46        self.prefix_lens
47            .iter()
48            .take_while(move |&&len| len <= name.len())
49            .filter_map(move |&len| name.get(..len))
50            .filter_map(|prefix| self.prefixed.get(prefix))
51            .flatten()
52            .copied()
53    }
54}
55
56impl AclCache {
57    /// A snapshot of `entries` that compares every host as text, as Kafka
58    /// 4.3.1 does. Say that the cluster reads a range from a host containing
59    /// `/` with [`Self::with_cidr_hosts_supported`].
60    #[must_use]
61    pub fn new(entries: Vec<AclEntry>) -> Self {
62        let mut by_type: HashMap<ResourceType, TypeIndex> = HashMap::new();
63        for (position, entry) in entries.iter().enumerate() {
64            let index = by_type.entry(entry.resource_type).or_default();
65            index.all.push(position);
66            let names = match entry.pattern_type {
67                PatternType::Literal => &mut index.literal,
68                PatternType::Prefixed => &mut index.prefixed,
69            };
70            names
71                .entry(entry.resource_name.clone())
72                .or_default()
73                .push(position);
74        }
75        for index in by_type.values_mut() {
76            index.prefix_lens = index.prefixed.keys().map(String::len).collect();
77            index.prefix_lens.sort_unstable();
78            index.prefix_lens.dedup();
79        }
80        Self {
81            entries,
82            by_type,
83            cidr_hosts_supported: false,
84        }
85    }
86
87    /// Sets whether a stored host containing `/` is a CIDR range (KIP-1276),
88    /// which it is once the broker the entries came from has a
89    /// `metadata.version` of 4.4-IV1 or higher. A gateway that reads a range
90    /// where that broker compares text would authorize a peer the broker
91    /// denies, and the other way round.
92    #[must_use]
93    pub fn with_cidr_hosts_supported(mut self, supported: bool) -> Self {
94        self.cidr_hosts_supported = supported;
95        self
96    }
97
98    #[must_use]
99    pub fn len(&self) -> usize { self.entries.len() }
100
101    #[must_use]
102    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
103
104    fn positions<'a>(&'a self, positions: impl Iterator<Item = usize> + 'a) -> AclEntries<'a> {
105        Box::new(positions.filter_map(|position| self.entries.get(position)))
106    }
107}
108
109impl AclSource for AclCache {
110    fn matching_acls<'a>(&'a self, rt: ResourceType, name: &'a str) -> AclEntries<'a> {
111        // MUST mirror MetadataImage::matching_acls: same resource_type, and
112        // (LITERAL == name) || (LITERAL == "*") || (PREFIXED && name.starts_with(resource_name)).
113        // The wildcard bucket is skipped when `name` is itself "*", whose
114        // LITERAL lookup already yields it.
115        let Some(index) = self.by_type.get(&rt) else {
116            return Box::new(iter::empty());
117        };
118        let wildcard = (name != WILDCARD_RESOURCE)
119            .then(|| index.literal(WILDCARD_RESOURCE))
120            .into_iter()
121            .flatten();
122        self.positions(
123            index
124                .literal(name)
125                .chain(wildcard)
126                .chain(index.prefixed(name)),
127        )
128    }
129
130    fn acls_of_type(&self, rt: ResourceType) -> AclEntries<'_> {
131        self.positions(
132            self.by_type
133                .get(&rt)
134                .into_iter()
135                .flat_map(|index| index.all.iter().copied()),
136        )
137    }
138
139    fn cidr_hosts_supported(&self) -> bool { self.cidr_hosts_supported }
140}
141
142#[cfg(test)]
143mod tests {
144
145    use {
146        super::*,
147        crate::{AuthorizationResult, test_support::acl as entry},
148        krabka_metadata::{
149            AclOperation, MetadataImage, MetadataRecord, ResourceType,
150            metadata_version::CIDR_ACL_MIN_LEVEL,
151        },
152        proptest::{collection, strategy::Strategy},
153        std::collections::HashSet,
154        uuid::Uuid,
155    };
156
157    /// A stable, comparable identity for an `AclEntry`.
158    ///
159    /// A test can then compare the `matching_acls` output of the two sources as
160    /// sets, in any order. The ACL enums derive `Debug` but not `Ord`, so this
161    /// function keys on the debug rendering of the identifying fields. That
162    /// rendering is a `String`, which is `Ord`.
163    fn key(e: &AclEntry) -> String {
164        format!(
165            "{:?}|{:?}|{}|{:?}",
166            e.resource_type, e.pattern_type, e.resource_name, e.operation
167        )
168    }
169
170    fn sorted_keys(it: AclEntries<'_>) -> Vec<String> {
171        let mut v: Vec<_> = it.map(key).collect();
172        v.sort();
173        v
174    }
175
176    /// Cross-validation guard for the two matching implementations.
177    ///
178    /// The test builds the same `AclEntry` set into BOTH a `MetadataImage`,
179    /// with `apply`, and an `AclCache`. The two must give the SAME matching set
180    /// for every probe. This protects against drift between the broker's image
181    /// matching and the gateway cache's reimplementation.
182    #[test]
183    fn cache_matches_image_for_every_probe() {
184        // A representative ACL set: literal exact, the literal "*" wildcard,
185        // a prefixed entry, an unrelated topic, and a different resource type.
186        let entries = vec![
187            entry().call(),
188            entry().op(AclOperation::Write).name("*").call(),
189            entry().pattern(PatternType::Prefixed).name("team-").call(),
190            entry().name("bar").call(),
191            entry().rt(ResourceType::Group).name("cg-1").call(),
192            entry()
193                .rt(ResourceType::Group)
194                .pattern(PatternType::Prefixed)
195                .name("app-")
196                .call(),
197        ];
198
199        // Build the same set into a MetadataImage (broker side) ...
200        let mut image = MetadataImage::new(Uuid::nil());
201        for e in &entries {
202            image.apply(&MetadataRecord::V1AccessControlEntry(e.clone()));
203        }
204        // ... and into an AclCache (gateway side).
205        let cache = AclCache::new(entries);
206
207        // Probes covering every matching code path.
208        let probes = [
209            (ResourceType::Topic, "foo"),      // literal exact hit (+ "*" wildcard)
210            (ResourceType::Topic, "*"),        // querying the wildcard resource itself
211            (ResourceType::Topic, "team-foo"), // prefixed hit (+ "*" wildcard)
212            (ResourceType::Topic, "team-"),    // prefix boundary (starts_with itself)
213            (ResourceType::Topic, "nomatch"),  // only the "*" wildcard matches
214            (ResourceType::Group, "cg-1"),     // literal hit, no wildcard for Group
215            (ResourceType::Group, "app-svc"),  // prefixed hit on Group
216            (ResourceType::Group, "other"),    // prefixed miss, no wildcard → empty
217            (ResourceType::Cluster, "kafka-cluster"), // wrong type → empty in both
218        ];
219
220        for (rt, name) in probes {
221            let from_image = sorted_keys(AclSource::matching_acls(&image, rt, name));
222            let from_cache = sorted_keys(AclSource::matching_acls(&cache, rt, name));
223            assert2::assert!(from_image == from_cache);
224        }
225    }
226
227    /// A host containing `/` is a CIDR range only where the cluster's
228    /// `metadata.version` has reached 4.4-IV1 (KIP-1276), and a snapshot of its
229    /// ACLs must decide as the cluster does. Each row builds the same range
230    /// ALLOW into an image at a `metadata.version` and into a cache told what
231    /// that image supports, and both must give the answer of the row: text
232    /// comparison below the level, where the range applies to no peer, and a
233    /// range at or above it. A cache that is not told compares as text.
234    #[yare::parameterized(
235        unfinalized_version = { None, Some(false), AuthorizationResult::Deny },
236        below_range_level = {
237            Some(CIDR_ACL_MIN_LEVEL - 1),
238            Some(false),
239            AuthorizationResult::Deny,
240        },
241        minimum_range_level = {
242            Some(CIDR_ACL_MIN_LEVEL),
243            Some(true),
244            AuthorizationResult::Allow,
245        },
246        unconfigured_cache = { None, None, AuthorizationResult::Deny },
247    )]
248    fn a_cache_reads_a_slash_host_as_the_cluster_does(
249        level: Option<i16>,
250        cache_support: Option<bool>,
251        expected: AuthorizationResult,
252    ) {
253        use {
254            crate::{AuthorizationRequest, Authorizer, SimpleAclAuthorizer},
255            krabka_metadata::{FeatureLevelRecord, metadata_version::METADATA_VERSION_FEATURE},
256            krabka_security::{AuthMethod, Principal},
257        };
258
259        let range_allow = AclEntry {
260            host: "10.0.0.0/8".into(),
261            ..entry().call()
262        };
263        let alice = Principal {
264            name: "alice".into(),
265            auth_method: AuthMethod::SaslPlain,
266            groups: vec![],
267        };
268        let peer = "10.1.2.3:5000".parse().unwrap();
269        let request = AuthorizationRequest {
270            principal: &alice,
271            host: &peer,
272            resource_type: ResourceType::Topic,
273            resource_name: "foo",
274            operation: AclOperation::Read,
275        };
276        let auth = SimpleAclAuthorizer::new(HashSet::new());
277        let mut image = MetadataImage::new(Uuid::nil());
278        if let Some(level) = level {
279            image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
280                name: METADATA_VERSION_FEATURE.into(),
281                level,
282            }));
283        }
284        image.apply(&MetadataRecord::V1AccessControlEntry(range_allow.clone()));
285        let cache = match cache_support {
286            Some(supported) => {
287                AclCache::new(vec![range_allow]).with_cidr_hosts_supported(supported)
288            }
289            None => AclCache::new(vec![range_allow]),
290        };
291        assert2::assert!(
292            (
293                auth.authorize(&image, &request),
294                auth.authorize(&cache, &request)
295            ) == (expected, expected)
296        );
297    }
298
299    /// The scan the index replaces: every entry of the snapshot, tested
300    /// one by one against the matching rule.
301    fn scanned(entries: &[AclEntry], rt: ResourceType, name: &str) -> Vec<String> {
302        let mut v: Vec<_> = entries
303            .iter()
304            .filter(|e| {
305                e.resource_type == rt
306                    && match e.pattern_type {
307                        PatternType::Literal => e.resource_name == name || e.resource_name == "*",
308                        PatternType::Prefixed => name.starts_with(e.resource_name.as_str()),
309                    }
310            })
311            .map(key)
312            .collect();
313        v.sort();
314        v
315    }
316
317    fn arb_entry() -> impl Strategy<Value = AclEntry> {
318        use proptest::prelude::*;
319        (
320            prop_oneof![Just(ResourceType::Topic), Just(ResourceType::Group)],
321            prop_oneof![Just(PatternType::Literal), Just(PatternType::Prefixed)],
322            prop_oneof![Just(AclOperation::Read), Just(AclOperation::Write)],
323            "(\\*|[aé-]{0,4})",
324        )
325            .prop_map(|(rt, pattern, op, name)| {
326                entry().rt(rt).pattern(pattern).op(op).name(&name).call()
327            })
328    }
329
330    proptest::proptest! {
331        /// The indexed lookup yields exactly the entries, duplicates included,
332        /// that a scan of the whole snapshot does, for names that share
333        /// prefixes, multi-byte characters a prefix length can split, the
334        /// empty name and the wildcard name.
335        #[test]
336        fn index_matches_a_full_scan(
337            entries in collection::vec(arb_entry(), 0..24),
338            probes in collection::vec("(\\*|[aé-]{0,6})", 1..8),
339        ) {
340            let cache = AclCache::new(entries.clone());
341            for name in &probes {
342                for rt in [ResourceType::Topic, ResourceType::Group, ResourceType::Cluster] {
343                    let indexed = sorted_keys(AclSource::matching_acls(&cache, rt, name));
344                    assert2::assert!(indexed == scanned(&entries, rt, name));
345                    let mut of_type: Vec<_> = cache.acls_of_type(rt).map(key).collect();
346                    of_type.sort();
347                    let mut expected: Vec<_> =
348                        entries.iter().filter(|e| e.resource_type == rt).map(key).collect();
349                    expected.sort();
350                    assert2::assert!(of_type == expected);
351                }
352            }
353        }
354    }
355
356    /// Several prefixes of different lengths all apply to a name that extends
357    /// each of them, and a prefix longer than the name never does.
358    #[test]
359    fn every_applicable_prefix_length_is_found() {
360        let cache = AclCache::new(vec![
361            entry().pattern(PatternType::Prefixed).name("t").call(),
362            entry().pattern(PatternType::Prefixed).name("team-").call(),
363            entry().pattern(PatternType::Prefixed).name("team-a").call(),
364            entry()
365                .pattern(PatternType::Prefixed)
366                .name("team-a-long")
367                .call(),
368            entry().pattern(PatternType::Prefixed).name("other").call(),
369        ]);
370        let names: Vec<_> = AclSource::matching_acls(&cache, ResourceType::Topic, "team-a1")
371            .map(|e| e.resource_name.as_str())
372            .collect();
373        assert2::assert!(names == ["t", "team-", "team-a"]);
374    }
375
376    #[test]
377    fn len_and_is_empty_track_entries() {
378        let empty = AclCache::default();
379        assert2::assert!(empty.is_empty());
380        assert2::assert!(empty.len() == 0);
381
382        let cache = AclCache::new(vec![entry().call()]);
383        assert2::assert!(!cache.is_empty());
384        assert2::assert!(cache.len() == 1);
385    }
386}
krabka_authz/cache.rs

krabka_authz/
cache.rs
