nisshi_auth/
lib.rs

1// Copyright ⓒ 2024-2026 Peter Morgan <peter.james.morgan@gmail.com>
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use nisshi_sans_io::ScramMechanism;
16use nisshi_storage::Storage;
17use rsasl::{
18    callback::{Context, Request, SessionCallback, SessionData},
19    config::SASLConfig,
20    mechanisms::scram::properties::ScramStoredPassword,
21    prelude::{SASLError, SASLServer, Session, SessionError, Validation},
22    property::{AuthId, AuthzId, Password},
23    validate::{Validate, ValidationError},
24};
25use std::{
26    fmt::{self, Debug, Formatter},
27    str::FromStr,
28    sync::{Arc, Mutex, PoisonError},
29};
30use thiserror::Error;
31use tokio::task::JoinError;
32use tracing::{debug, instrument};
33
34mod authenticate;
35mod handshake;
36
37pub use authenticate::SaslAuthenticateService;
38pub use handshake::SaslHandshakeService;
39
40#[derive(Clone, Debug, Error)]
41pub enum Error {
42    Join(Arc<JoinError>),
43    Poison,
44    SansIo(#[from] nisshi_sans_io::Error),
45    Sasl(Arc<SASLError>),
46    SaslSession(Arc<SessionError>),
47}
48
49impl fmt::Display for Error {
50    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
51        write!(f, "{self:?}")
52    }
53}
54
55impl From<JoinError> for Error {
56    fn from(value: JoinError) -> Self {
57        Self::Join(Arc::new(value))
58    }
59}
60
61impl<T> From<PoisonError<T>> for Error {
62    fn from(_value: PoisonError<T>) -> Self {
63        Self::Poison
64    }
65}
66
67impl From<SASLError> for Error {
68    fn from(value: SASLError) -> Self {
69        Self::Sasl(Arc::new(value))
70    }
71}
72
73impl From<SessionError> for Error {
74    fn from(value: SessionError) -> Self {
75        Self::SaslSession(Arc::new(value))
76    }
77}
78
79#[derive(Clone)]
80pub struct Authentication {
81    config: Arc<SASLConfig>,
82    stage: Arc<Mutex<Option<Stage>>>,
83}
84
85pub enum Stage {
86    Server(SASLServer<Justification>),
87    Session(Session<Justification>),
88    Finished(Option<Success>),
89}
90
91impl Debug for Stage {
92    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
93        f.debug_struct(stringify!(Stage)).finish()
94    }
95}
96
97impl Authentication {
98    pub fn server(config: Arc<SASLConfig>) -> Self {
99        let server = SASLServer::<Justification>::new(config.clone());
100        Self {
101            config,
102            stage: Arc::new(Mutex::new(Some(Stage::Server(server)))),
103        }
104    }
105
106    pub fn is_authenticated(&self) -> bool {
107        self.stage
108            .lock()
109            .map(|guard| matches!(guard.as_ref(), Some(Stage::Finished(_))))
110            .ok()
111            .unwrap_or_default()
112    }
113
114    /// Build a fresh `Stage::Server` from the stored config. Used by
115    /// the SASL handshake handler to support re-authentication
116    /// (KIP-368): a client periodically issues a new SaslHandshake on
117    /// an existing connection, and the broker must be willing to
118    /// initiate a new SASL exchange even though the previous one
119    /// already succeeded.
120    pub fn fresh_server(&self) -> Stage {
121        Stage::Server(SASLServer::<Justification>::new(self.config.clone()))
122    }
123}
124
125impl Debug for Authentication {
126    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
127        f.debug_struct(stringify!(Authentication)).finish()
128    }
129}
130
131#[derive(Debug, Error)]
132pub enum AuthError {
133    Bad,
134    Io(nisshi_sans_io::Error),
135    MissingProperty { mechanism: String, property: String },
136    NoSuchUser,
137    UnknownMechanism(String),
138}
139
140impl fmt::Display for AuthError {
141    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
142        write!(f, "{self:?}")
143    }
144}
145
146#[derive(Clone, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
147pub struct Success {
148    auth_id: String,
149}
150
151#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
152pub struct Justification;
153
154impl Validation for Justification {
155    type Value = Result<Success, AuthError>;
156}
157
158#[derive(Clone, Debug)]
159pub struct Callback<S> {
160    storage: S,
161}
162
163impl<S> Callback<S>
164where
165    S: Storage,
166{
167    pub fn new(storage: S) -> Self
168    where
169        S: Storage,
170    {
171        Self { storage }
172    }
173
174    #[instrument(skip_all)]
175    fn check(
176        &self,
177        session_data: &SessionData,
178        context: &Context<'_>,
179    ) -> Result<Result<Success, AuthError>, Error> {
180        debug!(mechanism = %session_data.mechanism().mechanism);
181
182        if session_data.mechanism().mechanism == "PLAIN" {
183            Ok(context
184                .get_ref::<Password>()
185                .ok_or(AuthError::MissingProperty {
186                    mechanism: session_data.mechanism().mechanism.to_string(),
187                    property: "Password".into(),
188                })
189                .and(
190                    context
191                        .get_ref::<AuthId>()
192                        .inspect(|auth_id| {
193                            debug!(mechanism = %session_data.mechanism().mechanism, auth_id)
194                        })
195                        .ok_or(AuthError::MissingProperty {
196                            mechanism: session_data.mechanism().mechanism.to_string(),
197                            property: "AuthId".into(),
198                        }).map(ToString::to_string).map(|auth_id| {
199                            Success { auth_id }
200                        })
201                ))
202        } else if session_data.mechanism().mechanism.starts_with("SCRAM-") {
203            Ok(context
204                .get_ref::<AuthId>()
205                .inspect(|auth_id| debug!(mechanism = %session_data.mechanism().mechanism, auth_id))
206                .ok_or(AuthError::MissingProperty {
207                    mechanism: session_data.mechanism().mechanism.to_string(),
208                    property: "AuthId".into(),
209                })
210                .and_then(|auth_id| {
211                    context
212                        .get_ref::<AuthzId>()
213                        .inspect(|authz_id| {
214                            debug!(mechanism = %session_data.mechanism().mechanism, authz_id)
215                        })
216                        .map_or(Ok(Success{
217                            auth_id:auth_id.to_string()
218                        }), |authz_id| {
219                            if authz_id == auth_id {
220                                Ok(Success{
221                                    auth_id:auth_id.to_string()
222                                })
223                            } else {
224                                Err(AuthError::Bad)
225                            }
226                        })
227                }))
228        } else {
229            Ok(Err(AuthError::UnknownMechanism(
230                session_data.mechanism().mechanism.to_string(),
231            )))
232        }
233    }
234}
235
236impl<S> SessionCallback for Callback<S>
237where
238    S: Storage,
239{
240    #[instrument(skip_all)]
241    fn callback(
242        &self,
243        session_data: &SessionData,
244        context: &Context<'_>,
245        request: &mut Request<'_>,
246    ) -> Result<(), SessionError> {
247        debug!(?session_data);
248
249        if session_data.mechanism().mechanism.starts_with("SCRAM-") {
250            let mechanism = ScramMechanism::from_str(session_data.mechanism().mechanism)
251                .map_err(|error| SessionError::Boxed(Box::new(error)))?;
252
253            let auth_id = context
254                .get_ref::<AuthId>()
255                .ok_or(SessionError::ValidationError(
256                    ValidationError::MissingRequiredProperty,
257                ))?;
258
259            debug!(?auth_id, ?mechanism);
260
261            let rt = tokio::runtime::Builder::new_current_thread()
262                .enable_all()
263                .build()?;
264
265            if let Ok(Some(credential)) = rt
266                .block_on(
267                    async move { self.storage.user_scram_credential(auth_id, mechanism).await },
268                )
269                .inspect_err(|err| debug!(auth_id, ?mechanism, ?err))
270            {
271                _ = request
272                    .satisfy::<ScramStoredPassword<'_>>(&ScramStoredPassword::new(
273                        credential.iterations as u32,
274                        &credential.salt[..],
275                        &credential.stored_key[..],
276                        &credential.server_key[..],
277                    ))
278                    .inspect_err(|err| debug!(auth_id, ?mechanism, ?err))?;
279            }
280        }
281
282        Ok(())
283    }
284
285    #[instrument(skip_all)]
286    fn validate(
287        &self,
288        session_data: &SessionData,
289        context: &Context<'_>,
290        validate: &mut Validate<'_>,
291    ) -> Result<(), ValidationError> {
292        debug!(?session_data);
293
294        _ = validate.with::<Justification, _>(|| {
295            self.check(session_data, context)
296                .map_err(|e| ValidationError::Boxed(Box::new(e)))
297        })?;
298
299        Ok(())
300    }
301}
302
303pub fn configuration<S>(storage: S) -> Result<Arc<SASLConfig>, Error>
304where
305    S: Storage,
306{
307    SASLConfig::builder()
308        .with_defaults()
309        .with_callback(Callback::new(storage))
310        .map_err(Into::into)
311}
312
313#[cfg(test)]
314mod tests {
315    use super::*;
316
317    fn is_send<T: Send>() {}
318    fn is_sync<T: Sync>() {}
319
320    #[test]
321    fn authentication() {
322        is_send::<Authentication>();
323        is_sync::<Authentication>();
324    }
325}
nisshi_auth/lib.rs

nisshi_auth/
lib.rs
