krabka_authz/lib.rs
1//! Shared Kafka-ACL authorization evaluator (broker + gateway).
2//!
3//! This crate holds the [`Authorizer`] trait, the ACL evaluator
4//! ([`SimpleAclAuthorizer`] and [`AllowAllAuthorizer`]), and an [`AclSource`]
5//! abstraction. One evaluator therefore serves both the broker and the gateway.
6//! The broker passes a `MetadataImage` snapshot. The gateway passes an
7//! [`AclCache`] over a `Vec<AclEntry>` that it fetched with `DescribeAcls`.
8//!
9//! The decision logic lives here once, so the two callers can never drift. That
10//! logic covers the super-user bypass, deny-wins, and operation implication.
11//!
12//! ## Authorizing a request
13//!
14//! ```rust
15//! use {
16//! krabka_authz::{AllowAllAuthorizer, AuthorizationRequest, AuthorizationResult, Authorizer},
17//! krabka_metadata::{AclOperation, MetadataImage, ResourceType},
18//! krabka_security::{AuthMethod, Principal},
19//! std::net::SocketAddr,
20//! uuid::Uuid,
21//! };
22//!
23//! let image = MetadataImage::new(Uuid::nil());
24//! let principal = Principal {
25//! name: "alice".into(),
26//! auth_method: AuthMethod::SaslPlain,
27//! groups: vec![],
28//! };
29//! let host: SocketAddr = "127.0.0.1:9092".parse().unwrap();
30//! let req = AuthorizationRequest {
31//! principal: &principal,
32//! host: &host,
33//! resource_type: ResourceType::Topic,
34//! resource_name: "orders",
35//! operation: AclOperation::Read,
36//! };
37//!
38//! assert2::assert!(AllowAllAuthorizer.authorize(&image, &req) == AuthorizationResult::Allow);
39//! ```
40#![forbid(unsafe_code)]
41
42mod allow_all;
43pub mod cache;
44pub mod cidr;
45mod host_format;
46#[cfg(test)]
47mod precedence;
48mod simple;
49mod source;
50
51pub use {
52 allow_all::AllowAllAuthorizer,
53 cache::AclCache,
54 host_format::jdk_host_address,
55 simple::SimpleAclAuthorizer,
56 source::{AclEntries, AclSource},
57};
58use {
59 krabka_metadata::{AclOperation, ResourceType},
60 krabka_security::Principal,
61 std::{collections::HashMap, fmt, net::SocketAddr, time::Duration},
62};
63
64/// What the caller asks `authorize`: which principal wants to do which
65/// operation on which resource, and from which host.
66///
67/// The struct borrows its references, so handler-side construction is
68/// allocation-free.
69#[derive(Debug, Clone)]
70pub struct AuthorizationRequest<'a> {
71 pub principal: &'a Principal,
72 pub host: &'a SocketAddr,
73 pub resource_type: ResourceType,
74 pub resource_name: &'a str,
75 pub operation: AclOperation,
76}
77
78/// Binary outcome: Kafka's ACL surface is allow or deny.
79///
80/// The trait boundary does not expose intermediate states, for example "not yet
81/// decided".
82#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83pub enum AuthorizationResult {
84 Allow,
85 Deny,
86}
87
88/// Pluggable per-broker or per-gateway authorization decision point.
89///
90/// Implementations own whatever state they need to make a decision, for example
91/// a super-user set, an HTTP client, or a decision cache. The caller holds a
92/// single `Arc<dyn Authorizer>`.
93///
94/// Implementations MUST be `Send + Sync + Debug`: handler code paths
95/// are async and the broker logs configs at startup.
96///
97/// The decision consults a [`AclSource`]. The broker passes its
98/// `MetadataImage` and the gateway passes an [`AclCache`]. ACL-free
99/// implementations (`AllowAll`, OPA) ignore it.
100pub trait Authorizer: Send + Sync + fmt::Debug {
101 /// Decide whether `req.principal` may do `req.operation` on
102 /// `(req.resource_type, req.resource_name)` from `req.host`.
103 ///
104 /// The authorizer may consult `source`, as the ACL-backed implementations
105 /// do, or ignore it entirely, as `AllowAll` and `Opa` do.
106 fn authorize(
107 &self,
108 source: &dyn AclSource,
109 req: &AuthorizationRequest<'_>,
110 ) -> AuthorizationResult;
111
112 /// The decision of [`Authorizer::authorize`] for a check that Kafka makes
113 /// with `logIfDenied = false`: a probe whose Deny is not a refusal, such
114 /// as the cluster-wide `Create` shortcut of `CreateTopics` that falls back
115 /// to a per-topic check, or the operations bit field a `Describe` reports.
116 ///
117 /// The decision is the same. What differs is what a Deny leaves behind:
118 /// none of the audit records and counters that `authorize` may write. The
119 /// default forwards to `authorize`, which is right for an implementation
120 /// that writes none. A decorator that audits a Deny MUST override this to
121 /// forward to the authorizer it wraps without auditing, and one that only
122 /// forwards MUST forward this method too.
123 fn authorize_quiet(
124 &self,
125 source: &dyn AclSource,
126 req: &AuthorizationRequest<'_>,
127 ) -> AuthorizationResult {
128 self.authorize(source, req)
129 }
130
131 /// Whether this implementation is a real authorization decision point,
132 /// the way Kafka's `authorizer.class.name` names one.
133 ///
134 /// Only [`AllowAllAuthorizer`] answers `false`: it is what a deployment
135 /// gets when it configures no authorizer at all, so a stored ACL would
136 /// never be consulted. The ACL administration RPCs read this to answer
137 /// `SECURITY_DISABLED`, as Kafka's `KafkaApis` does under
138 /// `authorizer.isEmpty`. A decorator must forward it to the authorizer it
139 /// wraps.
140 fn is_configured(&self) -> bool { true }
141
142 /// Upper bound on how long a caller-side cache may reuse a decision from
143 /// this authorizer without asking again, or `None` if a decision is
144 /// exactly as fresh as whatever `source` snapshot it was computed
145 /// against -- true of the ACL-backed implementations, whose grants live
146 /// in the metadata log a caller can already key a cache on.
147 ///
148 /// An authorizer whose decisions can go stale independently of `source`
149 /// -- an HTTP-backed policy engine such as OPA is the motivating case --
150 /// overrides this to its own decision-cache TTL, so a caller-side cache
151 /// keyed on metadata alone cannot outlive it and silently miss a policy
152 /// change the metadata log never saw. A decorator MUST forward the
153 /// wrapped authorizer's answer, as `is_configured` does.
154 fn decision_ttl(&self) -> Option<Duration> { None }
155
156 /// Whether `principal` holds `operation` on at least one resource pattern
157 /// of `resource_type` -- Kafka's `AuthorizerUtils.authorizeByResourceType`,
158 /// used where a request is not itself scoped to one resource name. KIP-599
159 /// motivates the check: `InitProducerId` with no transactional id accepts
160 /// either cluster-wide `IdempotentWrite` or `Write` on any topic, so a
161 /// principal that can produce need not also hold a cluster-wide grant.
162 ///
163 /// The default answers [`AuthorizationResult::Deny`]: an implementation
164 /// that cannot enumerate its ACL space (an HTTP-backed policy engine, for
165 /// example) has no cheap way to answer this honestly, and understating a
166 /// grant is the safe direction. [`AllowAllAuthorizer`] overrides this to
167 /// always answer `Allow`, since it never denies anything.
168 /// [`SimpleAclAuthorizer`] overrides it with a real scan. A decorator MUST
169 /// forward the wrapped authorizer's answer, as it does for `is_configured`
170 /// and `decision_ttl`.
171 fn authorize_by_resource_type(
172 &self,
173 _source: &dyn AclSource,
174 _principal: &Principal,
175 _host: &SocketAddr,
176 _resource_type: ResourceType,
177 _operation: AclOperation,
178 ) -> AuthorizationResult {
179 AuthorizationResult::Deny
180 }
181}
182
183/// Batch-authorize a set of topic names against the same principal, host, and
184/// operation.
185///
186/// The `Produce`, `Fetch`, and `Metadata` per-topic enforcement paths call this
187/// function. The returned map borrows its keys from the input iterator, so
188/// callers can avoid a copy of the topic strings.
189#[must_use]
190pub fn authorize_topics<'a>(
191 authorizer: &dyn Authorizer,
192 source: &dyn AclSource,
193 principal: &Principal,
194 host: &SocketAddr,
195 operation: AclOperation,
196 topic_names: impl IntoIterator<Item = &'a str>,
197) -> HashMap<&'a str, AuthorizationResult> {
198 authorize_topics_logged(
199 authorizer,
200 source,
201 principal,
202 host,
203 operation,
204 topic_names,
205 true,
206 )
207}
208
209/// [`authorize_topics`] with Kafka's `logIfDenied` flag. With `log_denied` off
210/// each name goes through [`Authorizer::authorize_quiet`], and a Deny leaves
211/// no audit record and no counter behind.
212///
213/// `Metadata` is the caller that turns it off: for all topics it filters every
214/// topic by `Describe` and hides the denied ones, so a Deny there is not a
215/// refusal (`logIfDenied = !metadataRequest.isAllTopics`).
216#[must_use]
217#[tracing::instrument(
218 level = "debug", skip_all,
219 fields(principal = %principal.name, operation = ?operation, host = %host.ip())
220)]
221pub fn authorize_topics_logged<'a>(
222 authorizer: &dyn Authorizer,
223 source: &dyn AclSource,
224 principal: &Principal,
225 host: &SocketAddr,
226 operation: AclOperation,
227 topic_names: impl IntoIterator<Item = &'a str>,
228 log_denied: bool,
229) -> HashMap<&'a str, AuthorizationResult> {
230 topic_names
231 .into_iter()
232 .map(|name| {
233 let req = AuthorizationRequest {
234 principal,
235 host,
236 resource_type: ResourceType::Topic,
237 resource_name: name,
238 operation,
239 };
240 let decision = if log_denied {
241 authorizer.authorize(source, &req)
242 } else {
243 authorizer.authorize_quiet(source, &req)
244 };
245 (name, decision)
246 })
247 .collect()
248}
249
250#[cfg(test)]
251mod test_support;