krabka_authz/
simple.rs

1//! ACL-based authorizer behind the [`Authorizer`] trait.
2//!
3//! The authorizer applies the super-user bypass, deny-wins-over-allow, LITERAL
4//! and PREFIXED matching, and principal, host, and operation wildcards.
5//!
6//! [`SimpleAclAuthorizer`] with an empty source and empty super-users denies
7//! everything by default. This default-deny behavior matches Kafka's
8//! `StandardAuthorizer` once an operator explicitly configures an authorizer.
9//! Kafka's `allow.everyone.if.no.acl.found` (default `false`) flips that
10//! default: when enabled and NO ACL at all applies to the resource --
11//! regardless of principal, host, operation, or permission type -- the
12//! request is allowed. If at least one ACL applies to the resource and none
13//! of them matches, the request is still denied. Set it with
14//! [`SimpleAclAuthorizer::with_allow_everyone_if_no_acl_found`].
15
16use {
17    krabka_metadata::{AclOperation, PatternType, PermissionType, ResourceType},
18    krabka_security::Principal,
19    krabka_verified::{
20        AclDecision, acl_decision,
21        authz::{AclDefault, AclFacts},
22    },
23    std::{collections::HashSet, net::SocketAddr},
24    tracing::field::Empty,
25};
26
27mod coverage;
28mod matching;
29#[cfg(test)]
30mod test_support;
31#[cfg(test)]
32mod tests;
33
34use {
35    self::{
36        coverage::DeniedNames,
37        matching::{matches_host, matches_operation, matches_principal, matches_resource},
38    },
39    crate::{AclSource, AuthorizationRequest, AuthorizationResult, Authorizer, jdk_host_address},
40};
41
42/// The resource name Kafka's default `authorizeByResourceType` probes first,
43/// a name that no real resource is expected to carry.
44const HARDCODED_RESOURCE_NAME: &str = "hardcode";
45
46/// Authorizer that consults the cluster's persisted ACLs.
47///
48/// The caller supplies the [`AclSource`] per call: a `MetadataImage` for the
49/// broker, an `AclCache` for the gateway.
50///
51/// This type holds the configured super-user set. Principals in this set bypass
52/// ACL evaluation and always get `Allow`.
53#[derive(Debug)]
54pub struct SimpleAclAuthorizer {
55    super_users: HashSet<String>,
56    allow_everyone_if_no_acl_found: bool,
57}
58
59impl SimpleAclAuthorizer {
60    #[must_use]
61    pub fn new(super_users: HashSet<String>) -> Self {
62        Self {
63            super_users,
64            allow_everyone_if_no_acl_found: false,
65        }
66    }
67
68    /// Sets Kafka's `allow.everyone.if.no.acl.found` (default `false`). See
69    /// the module doc for the exact semantics.
70    #[must_use]
71    pub fn with_allow_everyone_if_no_acl_found(mut self, allow: bool) -> Self {
72        self.allow_everyone_if_no_acl_found = allow;
73        self
74    }
75}
76
77impl Authorizer for SimpleAclAuthorizer {
78    // Per-request ACL decision: skip_all keeps the borrowed principal/host
79    // structs (which may carry the raw name) out of span fields; only
80    // non-sensitive routing context is recorded. No `err` — this returns a
81    // plain Allow/Deny, not a Result. `decision` is filled in before each
82    // return path.
83    #[tracing::instrument(
84        level = "debug",
85        skip_all,
86        fields(
87            principal = %req.principal.name,
88            resource_type = ?req.resource_type,
89            resource = %req.resource_name,
90            operation = ?req.operation,
91            host = %jdk_host_address(req.host.ip()),
92            decision = Empty,
93        )
94    )]
95    fn authorize(
96        &self,
97        source: &dyn AclSource,
98        req: &AuthorizationRequest<'_>,
99    ) -> AuthorizationResult {
100        let span = tracing::Span::current();
101        let super_user = self.super_users.contains(&req.principal.name);
102        let mut saw_allow = false;
103        let mut saw_deny = false;
104        let mut has_resource_acls = false;
105        if !super_user {
106            let cidr_hosts_supported = source.cidr_hosts_supported();
107            for entry in source.matching_acls(req.resource_type, req.resource_name) {
108                if !matches_resource(entry, req.resource_type, req.resource_name) {
109                    continue;
110                }
111                has_resource_acls = true;
112                if !matches_principal(entry, &req.principal.name)
113                    || !matches_host(entry, req.host.ip(), cidr_hosts_supported)
114                    || !matches_operation(entry.operation, req.operation, entry.permission_type)
115                {
116                    continue;
117                }
118                match entry.permission_type {
119                    PermissionType::Allow => saw_allow = true,
120                    PermissionType::Deny => {
121                        saw_deny = true;
122                        break;
123                    }
124                }
125            }
126        }
127        let default_decision = if self.allow_everyone_if_no_acl_found && !has_resource_acls {
128            AclDefault::Allow
129        } else {
130            AclDefault::Deny
131        };
132        let decision = acl_decision(AclFacts {
133            super_user,
134            saw_allow,
135            saw_deny,
136            default_decision,
137        });
138
139        let (label, result) = match decision {
140            AclDecision::AllowSuperuser => ("allow-superuser", AuthorizationResult::Allow),
141            AclDecision::AllowAcl => ("allow-acl", AuthorizationResult::Allow),
142            AclDecision::AllowNoAcl => ("allow-no-acl", AuthorizationResult::Allow),
143            AclDecision::DenyExplicit => ("deny-explicit", AuthorizationResult::Deny),
144            AclDecision::DenyDefault => ("deny-default", AuthorizationResult::Deny),
145        };
146        span.record("decision", label);
147        result
148    }
149
150    /// Allows first when the ordinary [`Self::authorize`] decision allows the
151    /// hard-coded name `hardcode`, as Kafka's default implementation does.
152    /// Otherwise it scans every stored entry of `resource_type` for an ALLOW ACL, matching
153    /// `principal`, `host`, and `operation`, that no DENY covers.
154    ///
155    /// A literal ALLOW needs its exact name to survive the matching DENYs.
156    /// Kafka checks a prefix ALLOW only against dominating DENY prefixes and
157    /// the DENY wildcard; literal exclusions cannot dominate a prefix grant.
158    /// A wildcard ALLOW survives unless a wildcard DENY exists. These pattern
159    /// checks need no synthetic name that another ACL could happen to deny.
160    #[tracing::instrument(
161        level = "debug",
162        skip_all,
163        fields(
164            principal = %principal.name,
165            resource_type = ?resource_type,
166            operation = ?operation,
167            host = %jdk_host_address(host.ip()),
168            decision = Empty,
169        )
170    )]
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        let span = tracing::Span::current();
180        if self.super_users.contains(&principal.name) {
181            span.record("decision", "allow-superuser");
182            return AuthorizationResult::Allow;
183        }
184        // Kafka's `Authorizer.authorizeByResourceType` first authorizes a
185        // hard-coded resource name, and StandardAuthorizer inherits it. The
186        // probe is allowed when a wildcard ALLOW covers the type or, under
187        // `allow.everyone.if.no.acl.found`, when no ACL of the type covers
188        // the name, so a cluster that runs the flag admits a principal that
189        // has no topic ACL at all.
190        let hardcode = AuthorizationRequest {
191            principal,
192            host,
193            resource_type,
194            resource_name: HARDCODED_RESOURCE_NAME,
195            operation,
196        };
197        if self.authorize(source, &hardcode) == AuthorizationResult::Allow {
198            span.record("decision", "allow-hardcode");
199            return AuthorizationResult::Allow;
200        }
201        let cidr_hosts_supported = source.cidr_hosts_supported();
202        let applicable: Vec<_> = source
203            .acls_of_type(resource_type)
204            .filter(|entry| {
205                matches_principal(entry, &principal.name)
206                    && matches_host(entry, host.ip(), cidr_hosts_supported)
207                    && matches_operation(entry.operation, operation, entry.permission_type)
208            })
209            .collect();
210        let denied = DeniedNames::new(
211            applicable
212                .iter()
213                .copied()
214                .filter(|entry| entry.permission_type == PermissionType::Deny),
215        );
216        let uncovered = applicable.iter().any(|entry| {
217            entry.permission_type == PermissionType::Allow
218                && match entry.pattern_type {
219                    PatternType::Literal if entry.resource_name == "*" => denied.allows_wildcard(),
220                    PatternType::Literal => denied.allows_literal(&entry.resource_name),
221                    PatternType::Prefixed => denied.allows_prefix(&entry.resource_name),
222                }
223        });
224        if uncovered {
225            span.record("decision", "allow-acl");
226            return AuthorizationResult::Allow;
227        }
228        span.record("decision", "deny-default");
229        AuthorizationResult::Deny
230    }
231}
krabka_authz/simple.rs

krabka_authz/
simple.rs
