krabka_metadata/
acl.rs

1//! Wire-stable ACL types replicated through the raft quorum
2//! with `MetadataRecord::V1AccessControlEntry` and
3//! `V1DeleteAccessControlEntry`.
4//!
5//! They mirror the shape that Kafka exposes on the `CreateAcls`, `DeleteAcls`,
6//! and `DescribeAcls` wire messages, but they stay pure data. The authorizer in
7//! `krabka-broker` evaluates them.
8
9use serde::{Deserialize, Serialize};
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
12pub enum ResourceType {
13    Topic,
14    Group,
15    Cluster,
16    TransactionalId,
17    /// KIP-48 delegation tokens. The resource name is the `KafkaPrincipal`
18    /// string form of the owner, for example `"User:alice"`. The pattern types
19    /// `LITERAL` and `PREFIXED` apply through the existing matcher. Only the
20    /// `Describe` operation is externally grantable, because `Create`,
21    /// `Renew`, and `Expire` are implicit on ownership.
22    DelegationToken,
23    /// KIP-373 users. The resource name is a user principal name, without the
24    /// `User:` prefix. `CreateTokens` and `DescribeTokens` on a `User`
25    /// resource let one principal create or describe delegation tokens owned
26    /// by another. Kafka wire discriminant 7.
27    User,
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
31pub enum PatternType {
32    Literal,
33    Prefixed,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
37pub enum PermissionType {
38    Allow,
39    Deny,
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
43pub enum AclOperation {
44    All,
45    Read,
46    Write,
47    Create,
48    Delete,
49    Alter,
50    Describe,
51    ClusterAction,
52    DescribeConfigs,
53    AlterConfigs,
54    IdempotentWrite,
55    /// KIP-939: permission to take part in two-phase commit (2PC) on a
56    /// `TransactionalId`. An `InitProducerId` that carries `enable2Pc=true`
57    /// needs it in addition to `Write`. Kafka wire discriminant 15.
58    TwoPhaseCommit,
59    /// KIP-373: permission to create a delegation token on behalf of the
60    /// `User` resource. Kafka wire discriminant 13.
61    CreateTokens,
62    /// KIP-373: permission to describe the delegation tokens the `User`
63    /// resource owns. Kafka wire discriminant 14.
64    DescribeTokens,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
68pub struct AclEntry {
69    pub resource_type: ResourceType,
70    pub resource_name: String,
71    pub pattern_type: PatternType,
72    pub principal: String,
73    pub host: String,
74    pub operation: AclOperation,
75    pub permission_type: PermissionType,
76}
77
78#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
79pub struct AclEntryFilter {
80    pub resource_type: Option<ResourceType>,
81    pub resource_name: Option<String>,
82    pub pattern_type: Option<PatternType>,
83    pub principal: Option<String>,
84    pub host: Option<String>,
85    pub operation: Option<AclOperation>,
86    pub permission_type: Option<PermissionType>,
87}
88
89impl AclEntryFilter {
90    /// Returns true if every populated axis matches `entry`. `None` axes
91    /// match anything.
92    #[must_use]
93    pub fn matches(&self, entry: &AclEntry) -> bool {
94        self.resource_type
95            .is_none_or(|rt| rt == entry.resource_type)
96            && self
97                .resource_name
98                .as_ref()
99                .is_none_or(|rn| rn == &entry.resource_name)
100            && self.pattern_type.is_none_or(|pt| pt == entry.pattern_type)
101            && self
102                .principal
103                .as_ref()
104                .is_none_or(|p| p == &entry.principal)
105            && self.host.as_ref().is_none_or(|h| h == &entry.host)
106            && self.operation.is_none_or(|op| op == entry.operation)
107            && self
108                .permission_type
109                .is_none_or(|pt| pt == entry.permission_type)
110    }
111}
112
113#[cfg(test)]
114mod tests {
115
116    use std::fmt::Debug;
117
118    use serde_wincode::SerdeCompat;
119    use wincode::{Deserialize as _, Serialize as _};
120
121    use super::*;
122
123    fn rt<T>(value: &T) -> T
124    where
125        T: Serialize + for<'de> Deserialize<'de> + PartialEq + Debug,
126    {
127        let bytes = <SerdeCompat<T>>::serialize(value).unwrap();
128        <SerdeCompat<T>>::deserialize(&bytes).unwrap()
129    }
130
131    #[test]
132    fn acl_entry_round_trip() {
133        let entry = AclEntry {
134            resource_type: ResourceType::Topic,
135            resource_name: "foo".into(),
136            pattern_type: PatternType::Literal,
137            principal: "User:alice".into(),
138            host: "*".into(),
139            operation: AclOperation::Read,
140            permission_type: PermissionType::Allow,
141        };
142        assert2::assert!(rt(&entry) == entry);
143    }
144
145    /// The metadata log stores ACL enums by variant index, so a variant
146    /// inserted anywhere but the end would make every stored entry after it
147    /// decode as its neighbour. Pin the encoding of every variant.
148    #[test]
149    fn acl_enum_log_encoding_is_stable() {
150        let index = |i: u32| i.to_le_bytes().to_vec();
151        let resource_types = [
152            (ResourceType::Topic, 0),
153            (ResourceType::Group, 1),
154            (ResourceType::Cluster, 2),
155            (ResourceType::TransactionalId, 3),
156            (ResourceType::DelegationToken, 4),
157            (ResourceType::User, 5),
158        ];
159        for (variant, want) in resource_types {
160            let bytes = <SerdeCompat<ResourceType>>::serialize(&variant).unwrap();
161            assert2::assert!((variant, bytes) == (variant, index(want)));
162        }
163        let operations = [
164            (AclOperation::All, 0),
165            (AclOperation::Read, 1),
166            (AclOperation::Write, 2),
167            (AclOperation::Create, 3),
168            (AclOperation::Delete, 4),
169            (AclOperation::Alter, 5),
170            (AclOperation::Describe, 6),
171            (AclOperation::ClusterAction, 7),
172            (AclOperation::DescribeConfigs, 8),
173            (AclOperation::AlterConfigs, 9),
174            (AclOperation::IdempotentWrite, 10),
175            (AclOperation::TwoPhaseCommit, 11),
176            (AclOperation::CreateTokens, 12),
177            (AclOperation::DescribeTokens, 13),
178        ];
179        for (variant, want) in operations {
180            let bytes = <SerdeCompat<AclOperation>>::serialize(&variant).unwrap();
181            assert2::assert!((variant, bytes) == (variant, index(want)));
182        }
183        for (variant, want) in [(PatternType::Literal, 0), (PatternType::Prefixed, 1)] {
184            let bytes = <SerdeCompat<PatternType>>::serialize(&variant).unwrap();
185            assert2::assert!((variant, bytes) == (variant, index(want)));
186        }
187        for (variant, want) in [(PermissionType::Allow, 0), (PermissionType::Deny, 1)] {
188            let bytes = <SerdeCompat<PermissionType>>::serialize(&variant).unwrap();
189            assert2::assert!((variant, bytes) == (variant, index(want)));
190        }
191    }
192
193    /// KIP-373's `User` resource and token operations survive the metadata
194    /// log's serde encoding like every other variant.
195    #[test]
196    fn kip_373_entries_round_trip() {
197        for operation in [AclOperation::CreateTokens, AclOperation::DescribeTokens] {
198            let entry = AclEntry {
199                resource_type: ResourceType::User,
200                resource_name: "alice".into(),
201                pattern_type: PatternType::Literal,
202                principal: "User:bob".into(),
203                host: "*".into(),
204                operation,
205                permission_type: PermissionType::Allow,
206            };
207            assert2::assert!(rt(&entry) == entry);
208        }
209    }
210
211    #[test]
212    fn acl_entry_filter_round_trip() {
213        let filter = AclEntryFilter {
214            resource_type: Some(ResourceType::Group),
215            resource_name: None,
216            pattern_type: Some(PatternType::Prefixed),
217            principal: Some("User:bob".into()),
218            host: None,
219            operation: Some(AclOperation::All),
220            permission_type: None,
221        };
222        assert2::assert!(rt(&filter) == filter);
223    }
224
225    #[test]
226    fn filter_with_all_none_matches_anything() {
227        let f = AclEntryFilter::default();
228        let entry = AclEntry {
229            resource_type: ResourceType::Cluster,
230            resource_name: "kafka-cluster".into(),
231            pattern_type: PatternType::Literal,
232            principal: "User:admin".into(),
233            host: "*".into(),
234            operation: AclOperation::All,
235            permission_type: PermissionType::Allow,
236        };
237        assert2::assert!(f.matches(&entry));
238    }
239
240    #[test]
241    fn filter_with_specific_axis_filters_correctly() {
242        let f = AclEntryFilter {
243            resource_type: Some(ResourceType::Topic),
244            ..AclEntryFilter::default()
245        };
246        let topic_entry = AclEntry {
247            resource_type: ResourceType::Topic,
248            resource_name: "foo".into(),
249            pattern_type: PatternType::Literal,
250            principal: "User:alice".into(),
251            host: "*".into(),
252            operation: AclOperation::Read,
253            permission_type: PermissionType::Allow,
254        };
255        let group_entry = AclEntry {
256            resource_type: ResourceType::Group,
257            ..topic_entry.clone()
258        };
259        assert2::assert!((f.matches(&topic_entry), f.matches(&group_entry)) == (true, false));
260    }
261}
krabka_metadata/acl.rs

krabka_metadata/
acl.rs
