cronoq_server/
auth.rs

1//! Authentication (bearer tokens, client certificates) and authorization (roles, pool lists).
2//!
3//! Tokens are kept only as SHA-256 digests and compared in constant time. A request is
4//! authenticated by its `authorization: Bearer <token>` header or, with mTLS, by a client
5//! certificate whose DNS subject alternative name is mapped to a role in `[[auth.certs]]`.
6
7use std::fmt;
8use std::sync::Arc;
9
10use sha2::{Digest, Sha256};
11use subtle::ConstantTimeEq;
12
13use crate::config::{AuthConfig, Role};
14
15/// Minimum accepted token length.
16pub const MIN_TOKEN_LEN: usize = 16;
17/// Token length below which a warning is logged.
18pub const RECOMMENDED_TOKEN_LEN: usize = 32;
19
20/// An authenticated caller.
21#[derive(Clone, Debug, PartialEq, Eq)]
22pub struct Principal {
23    /// Granted role.
24    pub role: Role,
25    /// Pools the caller may use (`None` = all).
26    pub pools: Option<Arc<[Arc<str>]>>,
27    /// Worker labels a worker with these credentials may claim, as `name=value` pairs
28    /// (`None` = any).
29    pub labels: Option<cronoq::Labels>,
30    /// Tenant these credentials submit for (`None` = the caller names it).
31    pub tenant: Option<Arc<str>>,
32    /// The bearer token the caller authenticated with, to forward calls to another shard
33    /// (`None` for client certificates, which cannot be forwarded).
34    pub bearer: Option<Bearer>,
35    /// The call was forwarded by another shard's server: answer it from this shard only.
36    pub forwarded: bool,
37}
38
39/// A bearer token kept to forward a call; never printed.
40#[derive(Clone, PartialEq, Eq)]
41pub struct Bearer(Arc<str>);
42
43impl Bearer {
44    #[cfg(test)]
45    pub(crate) fn for_tests(header: &str) -> Bearer {
46        Bearer(header.into())
47    }
48
49    /// The `authorization` header value.
50    pub fn header(&self) -> &str {
51        &self.0
52    }
53}
54
55impl std::fmt::Debug for Bearer {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        f.write_str("Bearer(<redacted>)")
58    }
59}
60
61/// Header that marks a call forwarded between shards.
62pub const FORWARDED_HEADER: &str = "cronoq-forwarded";
63
64impl Principal {
65    /// Unrestricted administrator (used when authentication is disabled).
66    pub fn anonymous_admin() -> Principal {
67        Principal {
68            role: Role::Admin,
69            pools: None,
70            labels: None,
71            tenant: None,
72            bearer: None,
73            forwarded: false,
74        }
75    }
76
77    /// Whether this principal may act as `required`.
78    pub fn has_role(&self, required: Role) -> bool {
79        self.role == Role::Admin || self.role == required
80    }
81
82    /// Whether a worker with these credentials may claim every one of `labels`.
83    pub fn may_claim_labels(&self, labels: &[(Arc<str>, Arc<str>)]) -> bool {
84        self.labels
85            .as_ref()
86            .is_none_or(|allowed| labels.iter().all(|l| allowed.contains(l)))
87    }
88
89    /// Whether this principal may use `pool`.
90    pub fn may_use_pool(&self, pool: &str) -> bool {
91        self.pools
92            .as_ref()
93            .is_none_or(|ps| ps.iter().any(|p| &**p == pool))
94    }
95}
96
97struct TokenEntry {
98    digest: [u8; 32],
99    principal: Principal,
100}
101
102struct CertEntry {
103    dns_name: String,
104    principal: Principal,
105}
106
107/// Token and certificate tables. With no tokens and no certificate mappings, authentication is
108/// disabled and every request acts as an unrestricted admin.
109#[derive(Clone, Default)]
110pub struct Auth {
111    tokens: Arc<Vec<TokenEntry>>,
112    certs: Arc<Vec<CertEntry>>,
113}
114
115impl fmt::Debug for Auth {
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        f.debug_struct("Auth")
118            .field("tokens", &self.tokens.len())
119            .field("cert_mappings", &self.certs.len())
120            .finish()
121    }
122}
123
124/// Parse `name=value` label pairs from the configuration.
125fn labels(v: &Option<Vec<String>>) -> anyhow::Result<Option<cronoq::Labels>> {
126    let Some(v) = v else { return Ok(None) };
127    v.iter()
128        .map(|l| match l.split_once('=') {
129            Some((k, val)) if !k.is_empty() => Ok((Arc::from(k), Arc::from(val))),
130            _ => Err(anyhow::anyhow!(
131                "auth label '{l}' must have the form name=value"
132            )),
133        })
134        .collect::<anyhow::Result<Vec<_>>>()
135        .map(|v| Some(v.into()))
136}
137
138fn digest(token: &str) -> [u8; 32] {
139    Sha256::digest(token.as_bytes()).into()
140}
141
142fn pools(list: &Option<Vec<String>>) -> Option<Arc<[Arc<str>]>> {
143    list.as_ref()
144        .map(|ps| ps.iter().map(|p| Arc::from(p.as_str())).collect())
145}
146
147impl Auth {
148    /// Build from configuration, reading `token_env` variables.
149    pub fn from_config(cfg: &AuthConfig) -> anyhow::Result<Auth> {
150        let mut tokens: Vec<TokenEntry> = Vec::new();
151        for t in &cfg.tokens {
152            let value = match (&t.token, &t.token_env) {
153                (Some(v), _) => v.clone(),
154                (None, Some(env)) => std::env::var(env).map_err(|_| {
155                    anyhow::anyhow!("environment variable {env} (auth token) is not set")
156                })?,
157                (None, None) => anyhow::bail!("auth token needs `token` or `token_env`"),
158            };
159            let value = value.trim().to_string();
160            anyhow::ensure!(
161                value.len() >= MIN_TOKEN_LEN,
162                "auth tokens must be at least {MIN_TOKEN_LEN} characters ({} role token is shorter)",
163                t.role.name()
164            );
165            if value.len() < RECOMMENDED_TOKEN_LEN {
166                tracing::warn!(
167                    role = t.role.name(),
168                    "auth token is shorter than {RECOMMENDED_TOKEN_LEN} characters; use at least 32 random characters"
169                );
170            }
171            let d = digest(&value);
172            anyhow::ensure!(
173                !tokens.iter().any(|e| bool::from(e.digest.ct_eq(&d))),
174                "duplicate auth token (each token must be unique)"
175            );
176            tokens.push(TokenEntry {
177                digest: d,
178                principal: Principal {
179                    role: t.role,
180                    pools: pools(&t.pools),
181                    labels: labels(&t.labels)?,
182                    tenant: t.tenant.as_deref().map(Arc::from),
183                    bearer: None,
184                    forwarded: false,
185                },
186            });
187        }
188        let certs = cfg
189            .certs
190            .iter()
191            .map(|c| {
192                Ok(CertEntry {
193                    dns_name: c.dns_name.to_ascii_lowercase(),
194                    principal: Principal {
195                        role: c.role,
196                        pools: pools(&c.pools),
197                        labels: labels(&c.labels)?,
198                        tenant: c.tenant.as_deref().map(Arc::from),
199                        bearer: None,
200                        forwarded: false,
201                    },
202                })
203            })
204            .collect::<anyhow::Result<_>>()?;
205        Ok(Auth {
206            tokens: Arc::new(tokens),
207            certs: Arc::new(certs),
208        })
209    }
210
211    /// No authentication.
212    pub fn disabled() -> Auth {
213        Auth::default()
214    }
215
216    /// Whether any credential is configured.
217    pub fn enabled(&self) -> bool {
218        !self.tokens.is_empty() || !self.certs.is_empty()
219    }
220
221    fn token(&self, token: &str) -> Option<&Principal> {
222        let d = digest(token);
223        // Compare against every entry so timing does not depend on which one matches.
224        let mut found = None;
225        for e in self.tokens.iter() {
226            if bool::from(e.digest.ct_eq(&d)) {
227                found = Some(&e.principal);
228            }
229        }
230        found
231    }
232
233    /// Principal for a client certificate (DER), by DNS subject alternative name.
234    pub fn principal_for_cert(&self, der: &[u8]) -> Option<Principal> {
235        if self.certs.is_empty() {
236            return None;
237        }
238        let cert = rustls_pki_types::CertificateDer::from(der);
239        let ee = webpki::EndEntityCert::try_from(&cert).ok()?;
240        let names: Vec<String> = ee
241            .valid_dns_names()
242            .map(|n| n.to_ascii_lowercase())
243            .collect();
244        self.certs
245            .iter()
246            .find(|c| names.contains(&c.dns_name))
247            .map(|c| c.principal.clone())
248    }
249
250    /// Authenticate a request from its `authorization` header and, if any, the principal derived
251    /// from its client certificate, and check the required role.
252    pub fn authorize(
253        &self,
254        header: Option<&str>,
255        cert: Option<&Principal>,
256        required: Role,
257    ) -> Result<Principal, AuthError> {
258        if !self.enabled() {
259            return Ok(Principal::anonymous_admin());
260        }
261        let p = match header {
262            Some(h) => {
263                let token = h
264                    .strip_prefix("Bearer ")
265                    .or_else(|| h.strip_prefix("bearer "))
266                    .ok_or(AuthError::Missing)?
267                    .trim();
268                let mut p = self.token(token).cloned().ok_or(AuthError::Invalid)?;
269                p.bearer = Some(Bearer(format!("Bearer {token}").into()));
270                p
271            }
272            None => cert.cloned().ok_or(AuthError::Missing)?,
273        };
274        if p.has_role(required) {
275            Ok(p)
276        } else {
277            Err(AuthError::Forbidden)
278        }
279    }
280
281    /// Check a header against a required role (no client certificate).
282    pub fn check(&self, header: Option<&str>, required: Role) -> Result<Principal, AuthError> {
283        self.authorize(header, None, required)
284    }
285}
286
287/// Why a request was refused.
288#[derive(Debug, PartialEq, Eq)]
289pub enum AuthError {
290    /// No credentials.
291    Missing,
292    /// Unknown token.
293    Invalid,
294    /// Valid credentials without the required role.
295    Forbidden,
296}
297
298impl AuthError {
299    /// gRPC status for this error.
300    pub fn to_status(&self) -> tonic::Status {
301        match self {
302            AuthError::Missing => tonic::Status::unauthenticated(
303                "missing credentials (bearer token or client certificate)",
304            ),
305            AuthError::Invalid => tonic::Status::unauthenticated("invalid token"),
306            AuthError::Forbidden => {
307                tonic::Status::permission_denied("credentials lack the required role")
308            }
309        }
310    }
311}
312
313/// Authenticate a gRPC request (bearer token or mapped client certificate).
314pub fn check_grpc<T>(
315    auth: &Auth,
316    req: &tonic::Request<T>,
317    role: Role,
318) -> Result<Principal, tonic::Status> {
319    let header = req
320        .metadata()
321        .get("authorization")
322        .and_then(|v| v.to_str().ok());
323    let cert = if auth.certs.is_empty() {
324        None
325    } else {
326        req.peer_certs().and_then(|certs| {
327            certs
328                .first()
329                .and_then(|c| auth.principal_for_cert(c.as_ref()))
330        })
331    };
332    let mut p = auth
333        .authorize(header, cert.as_ref(), role)
334        .map_err(|e| e.to_status())?;
335    p.forwarded = req.metadata().contains_key(FORWARDED_HEADER);
336    Ok(p)
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use crate::config::TokenConfig;
343
344    const C: &str = "client-token-0123456789abcdef0123";
345    const A: &str = "admin-token-0123456789abcdef01234";
346
347    fn tok(t: &str, role: Role, pools: Option<Vec<&str>>) -> TokenConfig {
348        TokenConfig {
349            token: Some(t.into()),
350            token_env: None,
351            role,
352            pools: pools.map(|p| p.into_iter().map(String::from).collect()),
353            labels: None,
354            tenant: None,
355        }
356    }
357
358    #[test]
359    fn roles_and_pools() {
360        let auth = Auth::from_config(&AuthConfig {
361            tokens: vec![
362                tok(C, Role::Client, Some(vec!["a"])),
363                tok(A, Role::Admin, None),
364            ],
365            certs: vec![],
366        })
367        .unwrap();
368        let c = auth
369            .check(Some(&format!("Bearer {C}")), Role::Client)
370            .unwrap();
371        assert!(c.may_use_pool("a") && !c.may_use_pool("b"));
372        assert_eq!(
373            auth.check(Some(&format!("Bearer {C}")), Role::Worker),
374            Err(AuthError::Forbidden)
375        );
376        assert_eq!(
377            auth.check(Some(&format!("Bearer {C}")), Role::Metrics),
378            Err(AuthError::Forbidden)
379        );
380        assert!(
381            auth.check(Some(&format!("Bearer {A}")), Role::Worker)
382                .is_ok()
383        );
384        assert_eq!(
385            auth.check(Some("Bearer xxxxxxxxxxxxxxxxxxxx"), Role::Client),
386            Err(AuthError::Invalid)
387        );
388        assert_eq!(auth.check(None, Role::Client), Err(AuthError::Missing));
389        assert_eq!(
390            Auth::disabled().check(None, Role::Admin),
391            Ok(Principal::anonymous_admin())
392        );
393    }
394
395    #[test]
396    fn rejects_short_and_duplicate_tokens() {
397        let short = Auth::from_config(&AuthConfig {
398            tokens: vec![tok("short", Role::Client, None)],
399            certs: vec![],
400        });
401        assert!(short.is_err());
402        let dup = Auth::from_config(&AuthConfig {
403            tokens: vec![tok(A, Role::Admin, None), tok(A, Role::Client, None)],
404            certs: vec![],
405        });
406        assert!(
407            dup.is_err(),
408            "a duplicate must not silently downgrade the admin token"
409        );
410    }
411}
cronoq_server/auth.rs

cronoq_server/
auth.rs
