krabka_admin_ui/
auth.rs

1//! Broker-backed login for the admin UI.
2
3use std::{fmt, future::Future, pin::Pin, time::Duration};
4
5use krabka_client_admin::{AdminClient, AdminError};
6use krabka_client_core::security::{ClientSecurity, SaslCredentials};
7use krabka_security::SaslMechanism;
8use krabka_units::prelude::*;
9use serde::{Deserialize, Serialize};
10
11use crate::{
12    admin::AdminFacade,
13    config::AdminUiConfig,
14    error::UiError,
15    permissions::{Capabilities, derive_capabilities},
16    session::{SessionCredentials, SessionStore, SessionUser},
17};
18
19/// How long the login path waits for one bootstrap address to accept a TCP
20/// connection while it decides why an authenticated connect failed.
21const BROKER_PROBE_TIMEOUT: Time = secs(2);
22
23#[derive(Clone, PartialEq, Eq, Deserialize)]
24pub struct LoginRequest {
25    pub username: String,
26    pub password: String,
27}
28
29#[derive(Clone, PartialEq, Eq, Serialize)]
30pub struct LoginSuccess {
31    pub username: String,
32    pub principal: String,
33    pub session_id: String,
34}
35
36impl fmt::Debug for LoginSuccess {
37    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
38        formatter
39            .debug_struct("LoginSuccess")
40            .field("username", &self.username)
41            .field("principal", &self.principal)
42            .field("session_id", &"<redacted>")
43            .finish()
44    }
45}
46
47#[must_use]
48pub fn build_scram_sha512_security(
49    cfg: &AdminUiConfig,
50    username: &str,
51    password: &str,
52) -> ClientSecurity {
53    ClientSecurity {
54        protocol: cfg.security.listener_protocol(),
55        tls: cfg.security.tls(),
56        sasl: Some(SaslCredentials::Scram {
57            mechanism: SaslMechanism::ScramSha512,
58            username: username.to_string(),
59            password: password.to_string(),
60            delegation_token: false,
61        }),
62        sasl_host: None,
63    }
64}
65
66#[derive(Debug, Clone, Copy, Default)]
67pub struct AdminClientLoginBroker;
68
69pub trait LoginBroker {
70    /// Signs the operator in and reports what the operator's ACLs permit.
71    ///
72    /// Returns [`UiError::NotAuthenticated`] when the broker rejects the
73    /// credentials, and [`UiError::BrokerConnection`] when no broker answers.
74    /// The caller reports those two as different HTTP statuses.
75    fn authenticate<'a>(
76        &'a self,
77        cfg: &'a AdminUiConfig,
78        username: &'a str,
79        password: &'a str,
80    ) -> Pin<Box<dyn Future<Output = Result<Capabilities, UiError>> + Send + 'a>>;
81}
82
83impl LoginBroker for AdminClientLoginBroker {
84    fn authenticate<'a>(
85        &'a self,
86        cfg: &'a AdminUiConfig,
87        username: &'a str,
88        password: &'a str,
89    ) -> Pin<Box<dyn Future<Output = Result<Capabilities, UiError>> + Send + 'a>> {
90        Box::pin(async move {
91            let security = build_scram_sha512_security(cfg, username, password);
92            let client =
93                match AdminClient::connect_secured(&cfg.bootstrap_addrs, Some(security)).await {
94                    Ok(client) => client,
95                    Err(error) => return Err(classify_connect_failure(cfg, error).await),
96                };
97
98            Ok(capabilities_from_broker(client, username).await)
99        })
100    }
101}
102
103impl<T: LoginBroker + ?Sized> LoginBroker for &T {
104    fn authenticate<'a>(
105        &'a self,
106        cfg: &'a AdminUiConfig,
107        username: &'a str,
108        password: &'a str,
109    ) -> Pin<Box<dyn Future<Output = Result<Capabilities, UiError>> + Send + 'a>> {
110        (*self).authenticate(cfg, username, password)
111    }
112}
113
114/// Reads the operator's ACLs and derives what the UI shows.
115///
116/// A broker that refuses the ACL read, and an operator that matches no ACL at
117/// all, both give every capability: a Kafka super user bypasses ACLs and so
118/// matches no entry, and a cluster that runs without an authorizer refuses the
119/// read. The UI cannot tell either case from a truly ungranted operator, so it
120/// shows every page and lets the broker refuse what the operator may not do.
121async fn capabilities_from_broker(client: AdminClient, username: &str) -> Capabilities {
122    let mut facade = AdminFacade::new(client);
123    let principal = format!("User:{username}");
124
125    let Ok(entries) = facade.acl_entries().await else {
126        return Capabilities::all();
127    };
128    let derived = derive_capabilities(&principal, &entries);
129
130    if derived == Capabilities::none() {
131        return Capabilities::all();
132    }
133
134    derived
135}
136
137/// Separates a rejected password from an unreachable broker.
138///
139/// `AdminClient::connect_secured` reports every broker-bootstrap failure as
140/// `AdminError::Connect`, whatever the cause, so the error alone cannot say
141/// which happened. A TCP probe of the same addresses decides it: a broker that
142/// accepts a TCP connection runs, so the authenticated connect failed in the
143/// handshake, and for this UI that means the credentials.
144async fn classify_connect_failure(cfg: &AdminUiConfig, error: AdminError) -> UiError {
145    if !matches!(error, AdminError::Connect { .. }) {
146        return UiError::from(error);
147    }
148
149    if any_address_accepts_tcp(&cfg.bootstrap_addrs, BROKER_PROBE_TIMEOUT).await {
150        return UiError::NotAuthenticated;
151    }
152
153    UiError::from(error)
154}
155
156/// Reports whether any of the addresses accepts a TCP connection inside the
157/// timeout.
158pub async fn any_address_accepts_tcp(addrs: &[String], timeout: Time) -> bool {
159    let deadline = Duration::try_from_secs_f64(timeout.secs_f64()).unwrap_or(Duration::ZERO);
160
161    for addr in addrs {
162        if let Ok(Ok(stream)) =
163            tokio::time::timeout(deadline, tokio::net::TcpStream::connect(addr.as_str())).await
164        {
165            drop(stream);
166            return true;
167        }
168    }
169
170    false
171}
172
173pub struct AuthService<'a, B = AdminClientLoginBroker> {
174    cfg: &'a AdminUiConfig,
175    sessions: &'a SessionStore,
176    broker: B,
177}
178
179impl<'a> AuthService<'a, AdminClientLoginBroker> {
180    #[must_use]
181    pub const fn new(cfg: &'a AdminUiConfig, sessions: &'a SessionStore) -> Self {
182        Self {
183            cfg,
184            sessions,
185            broker: AdminClientLoginBroker,
186        }
187    }
188}
189
190impl<'a, B: LoginBroker> AuthService<'a, B> {
191    #[must_use]
192    pub const fn new_with_broker(
193        cfg: &'a AdminUiConfig,
194        sessions: &'a SessionStore,
195        broker: B,
196    ) -> Self {
197        Self {
198            cfg,
199            sessions,
200            broker,
201        }
202    }
203
204    /// # Errors
205    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
206    pub async fn login(&self, request: LoginRequest) -> Result<LoginSuccess, UiError> {
207        let capabilities = self
208            .broker
209            .authenticate(self.cfg, &request.username, &request.password)
210            .await?;
211
212        let principal = format!("User:{}", request.username);
213        let session_id = self.sessions.create_authenticated(
214            SessionUser {
215                username: request.username.clone(),
216                principal: principal.clone(),
217            },
218            SessionCredentials::scram_sha512(request.password),
219            capabilities,
220        );
221
222        Ok(LoginSuccess {
223            username: request.username,
224            principal,
225            session_id: session_id.expose_for_cookie().to_string(),
226        })
227    }
228}
krabka_admin_ui/auth.rs

krabka_admin_ui/
auth.rs
