krabka_metadata/
acl.rs

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

krabka_metadata/
acl.rs
