krabka_security/
principal.rs

1use std::{fmt, str::FromStr};
2
3use serde::{Deserialize, Serialize};
4use thiserror::Error;
5
6use crate::SaslMechanism;
7
8/// How a [`Principal`] was authenticated.
9///
10/// This is a strict superset of [`SaslMechanism`]. It also covers mTLS
11/// client-cert authentication and the implicit ANONYMOUS path on PLAINTEXT and
12/// SSL-no-mTLS listeners.
13///
14/// This enum stays distinct from `SaslMechanism`, because `SaslMechanism` has a
15/// `from_wire`/`wire_name` contract and the broker stores it verbatim in
16/// `V1ScramCredential` metadata records. Neither applies to mTLS.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
18pub enum AuthMethod {
19    /// Anonymous, with no SASL and no mTLS. This covers PLAINTEXT listeners and
20    /// SSL listeners where the client did not present a cert.
21    Anonymous,
22    /// SASL/PLAIN.
23    SaslPlain,
24    /// SASL/SCRAM-SHA-256.
25    SaslScramSha256,
26    /// SASL/SCRAM-SHA-512.
27    SaslScramSha512,
28    /// SASL/OAUTHBEARER.
29    SaslOAuthBearer,
30    /// SASL/GSSAPI (Kerberos, RFC 4752).
31    SaslGssapi,
32    /// mTLS client-cert verified against the listener's
33    /// `client_ca_path`.
34    MTls,
35}
36
37impl AuthMethod {
38    /// Map a SASL `SaslMechanism` onto its `AuthMethod` equivalent.
39    #[must_use]
40    pub fn from_sasl(m: SaslMechanism) -> Self {
41        match m {
42            SaslMechanism::Plain => Self::SaslPlain,
43            SaslMechanism::ScramSha256 => Self::SaslScramSha256,
44            SaslMechanism::ScramSha512 => Self::SaslScramSha512,
45            SaslMechanism::OAuthBearer => Self::SaslOAuthBearer,
46            SaslMechanism::Gssapi => Self::SaslGssapi,
47        }
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct Principal {
53    pub name: String,
54    pub auth_method: AuthMethod,
55    /// OAuth-derived group memberships from the listener's `groupsClaim`.
56    ///
57    /// The vec is empty for non-OAuth principals, which are PLAIN, SCRAM, mTLS,
58    /// and anonymous. It is also empty for OAuth principals whose listener has
59    /// no `groupsClaim` configured. No broker-side authorizer reads this field
60    /// yet. The broker fills it as scaffolding and for observability.
61    pub groups: Vec<String>,
62}
63
64impl Principal {
65    /// Project a runtime session [`Principal`] onto the Kafka wire-level
66    /// [`KafkaPrincipal`].
67    ///
68    /// The wire-level form is `principalType:name`, and ACLs and
69    /// delegation-token records use it. All authenticated callers ride under
70    /// `principal_type = "User"`, which matches Kafka's
71    /// `DefaultKafkaPrincipalBuilder`.
72    #[must_use]
73    pub fn to_kafka(&self) -> KafkaPrincipal {
74        KafkaPrincipal {
75            principal_type: "User".to_owned(),
76            name: self.name.clone(),
77        }
78    }
79}
80
81/// KIP-48: Kafka wire-level principal.
82///
83/// This is the `(principalType, name)` pair that delegation-token records, ACL
84/// entries, and `KafkaPrincipal`-shaped fields carry across the Kafka protocol.
85/// It is distinct from [`Principal`], which models the *runtime session*
86/// identity, that is the auth method and the OAuth groups. The format is
87/// stable: `Display` and `FromStr` round-trip the canonical `Type:Name` form.
88#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
89pub struct KafkaPrincipal {
90    pub principal_type: String,
91    pub name: String,
92}
93
94impl fmt::Display for KafkaPrincipal {
95    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
96        write!(f, "{}:{}", self.principal_type, self.name)
97    }
98}
99
100impl FromStr for KafkaPrincipal {
101    type Err = String;
102    fn from_str(s: &str) -> Result<Self, String> {
103        let (pt, n) = s
104            .split_once(':')
105            .ok_or_else(|| format!("invalid principal {s:?}"))?;
106        Ok(Self {
107            principal_type: pt.into(),
108            name: n.into(),
109        })
110    }
111}
112
113#[derive(Debug, Error, Clone, PartialEq, Eq)]
114pub enum AuthError {
115    #[error("unknown user")]
116    UnknownUser,
117    #[error("bad password")]
118    BadPassword,
119    #[error("bad proof")]
120    BadProof,
121    #[error("malformed message")]
122    MalformedMessage,
123    #[error("unsupported mechanism")]
124    UnsupportedMechanism,
125    /// SCRAM client-first named a GS2 authorization id other than the
126    /// authenticating user, which Kafka's `ScramSaslServer` refuses.
127    #[error("authorization id differs from the authenticating user")]
128    AuthorizationIdMismatch,
129    /// OAUTHBEARER token failed validation. The token was expired, carried bad
130    /// claims, was a signed token that the unsecured validator rejected, had no
131    /// principal, and so on. This maps to the RFC 7628 `invalid_token` server
132    /// error status.
133    #[error("invalid token")]
134    InvalidToken,
135    /// OAUTHBEARER introspection HTTP round-trip failed at the transport layer.
136    ///
137    /// This is distinct from `InvalidToken`, so the SASL handler can surface
138    /// "`IdP` unreachable" separately from "client supplied a bad token". It
139    /// maps to the RFC 7628 `invalid_token` server error status.
140    #[error("oauthbearer introspection transport: {0}")]
141    IntrospectionTransport(String),
142    /// OAUTHBEARER token passed every other check but lacked an item of the
143    /// scope the validator requires. This maps to the RFC 7628
144    /// `insufficient_scope` server error status, and `required_scope` is the
145    /// whole configured list, which Kafka's challenge carries in its `scope`
146    /// field. `oauth_error_json` renders that challenge.
147    #[error("insufficient scope: requires {required_scope:?}")]
148    InsufficientScope { required_scope: Vec<String> },
149}
150
151#[cfg(test)]
152mod tests {
153
154    use super::*;
155
156    #[test]
157    fn from_sasl_mapping() {
158        for (mechanism, want) in [
159            (SaslMechanism::Plain, AuthMethod::SaslPlain),
160            (SaslMechanism::ScramSha256, AuthMethod::SaslScramSha256),
161            (SaslMechanism::ScramSha512, AuthMethod::SaslScramSha512),
162            (SaslMechanism::OAuthBearer, AuthMethod::SaslOAuthBearer),
163            (SaslMechanism::Gssapi, AuthMethod::SaslGssapi),
164        ] {
165            assert2::assert!(AuthMethod::from_sasl(mechanism) == want);
166        }
167    }
168
169    #[test]
170    fn principal_display_is_type_colon_name() {
171        let p = KafkaPrincipal {
172            principal_type: "User".into(),
173            name: "alice".into(),
174        };
175        assert2::assert!(p.to_string() == "User:alice");
176        // FromStr is the inverse.
177        assert2::assert!("User:alice".parse::<KafkaPrincipal>().unwrap() == p);
178    }
179}
krabka_security/principal.rs

krabka_security/
principal.rs
