use std::sync::Arc;
use async_trait::async_trait;
use axum::http::request::Parts;
use crate::sql::sqlx;
use crate::sql::Pool;
use super::auth::parse_basic_auth;
use super::password;
use super::password::HashLane;
use crate::login_throttle::{ClientIp, LoginAttempt, LoginScope};
#[derive(Debug, Clone)]
pub struct AuthUser {
pub id: i64,
pub username: String,
pub is_superuser: bool,
}
#[derive(Debug, thiserror::Error)]
pub enum AuthError {
#[error("database error: {0}")]
Database(#[from] sqlx::Error),
#[error("database error: {0}")]
Exec(#[from] crate::sql::ExecError),
#[error("token is malformed or expired")]
InvalidToken,
#[error("account is inactive")]
Inactive,
#[error("login refused: {0:?}")]
Refused(crate::login_throttle::LoginRefused),
}
impl AuthError {
fn from_hash(e: super::TenancyError) -> Self {
match e {
super::TenancyError::Busy => Self::Refused(crate::login_throttle::LoginRefused::Busy),
_ => Self::InvalidToken,
}
}
}
async fn begin(
parts: &Parts,
scope: fn(String) -> LoginScope,
username: &str,
) -> Result<LoginAttempt, AuthError> {
let Some(slug) = parts.extensions.get::<super::TenantSlug>() else {
tracing::error!("auth backend ran without a TenantSlug; use require_auth / optional_auth");
return Err(AuthError::InvalidToken);
};
let ip = parts
.extensions
.get::<ClientIp>()
.cloned()
.unwrap_or_else(|| ClientIp::from_parts(&parts.extensions, &parts.headers));
crate::login_throttle::shared()
.begin(&scope(slug.0.clone()), &ip, username)
.await
.map_err(AuthError::Refused)
}
#[async_trait]
pub trait AuthBackend: Send + Sync {
async fn authenticate(&self, parts: &Parts, pool: &Pool)
-> Result<Option<AuthUser>, AuthError>;
}
pub type BoxedBackend = Arc<dyn AuthBackend>;
pub struct ModelBackend;
#[async_trait]
impl AuthBackend for ModelBackend {
async fn authenticate(
&self,
parts: &Parts,
pool: &Pool,
) -> Result<Option<AuthUser>, AuthError> {
use crate::core::Column as _;
use crate::sql::FetcherPool as _;
let auth_header = parts
.headers
.get(axum::http::header::AUTHORIZATION)
.and_then(|v| v.to_str().ok());
let (username, password) = match parse_basic_auth(auth_header) {
Some(pair) => pair,
None => return Ok(None),
};
let mut attempt = begin(parts, LoginScope::TenantBasic, &username).await?;
let users = super::auth::User::objects()
.where_(super::auth::User::username.eq(username.clone()))
.fetch(pool)
.await?;
let Some(user) = users.into_iter().next() else {
password::verify_dummy_async_in(HashLane::Credential, &password)
.await
.map_err(AuthError::from_hash)?;
attempt.failed().await;
return Ok(None);
};
attempt
.resolve(&user.username)
.await
.map_err(AuthError::Refused)?;
let ok = password::verify_async_in(HashLane::Credential, &password, &user.password_hash)
.await
.map_err(AuthError::from_hash)?;
if !user.active || !ok {
attempt.failed().await;
return Ok(None);
}
attempt.succeeded().await;
Ok(Some(AuthUser {
id: user.id.get().copied().unwrap_or(0),
username: user.username,
is_superuser: user.is_superuser,
}))
}
}
#[derive(crate::Model, Debug, Clone)]
#[rustango(
table = "rustango_api_keys",
admin(
list_display = "user_id, key_prefix, label, expires_at, created_at",
ordering = "-created_at",
readonly_fields = "key_prefix, key_hash, created_at",
)
)]
pub struct ApiKey {
#[rustango(primary_key)]
pub id: crate::sql::Auto<i64>,
#[rustango(fk = "rustango_users", on = "id", on_delete = "cascade")]
pub user_id: i64,
#[rustango(max_length = 8, index)]
pub key_prefix: String,
#[rustango(max_length = 255)]
pub key_hash: String,
#[rustango(max_length = 100)]
pub label: String,
pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
#[rustango(auto_now_add)]
pub created_at: crate::sql::Auto<chrono::DateTime<chrono::Utc>>,
}
#[cfg(feature = "admin")]
crate::register_admin_object_permission!("rustango_api_keys", "add", |_, _| false);
#[cfg(feature = "postgres")]
pub async fn ensure_api_keys_table(pool: &crate::sql::sqlx::PgPool) -> Result<(), sqlx::Error> {
ensure_api_keys_table_pool(&crate::sql::Pool::from(pool.clone())).await
}
pub async fn ensure_api_keys_table_pool(pool: &Pool) -> Result<(), sqlx::Error> {
use crate::core::Model as _;
let snapshot = crate::migrate::SchemaSnapshot::from_models(&[ApiKey::SCHEMA]);
crate::migrate::apply_idempotent(pool, &snapshot).await?;
Ok(())
}
pub struct ApiKeyBackend;
#[async_trait]
impl AuthBackend for ApiKeyBackend {
async fn authenticate(
&self,
parts: &Parts,
pool: &Pool,
) -> Result<Option<AuthUser>, AuthError> {
use crate::core::Column as _;
use crate::sql::FetcherPool as _;
let bearer = extract_bearer(parts)?;
let Some(token) = bearer else {
return Ok(None);
};
let (prefix, secret) = match token.split_once('.') {
Some(p) => p,
None => return Ok(None), };
if prefix.len() != 8 {
return Ok(None);
}
let attempt = begin(parts, LoginScope::TenantApiKey, "").await?;
let keys = ApiKey::objects()
.where_(ApiKey::key_prefix.eq(prefix.to_owned()))
.fetch(pool)
.await?;
let Some(key) = keys.into_iter().next() else {
password::verify_dummy_async_in(HashLane::Credential, secret)
.await
.map_err(AuthError::from_hash)?;
attempt.failed().await;
return Ok(None);
};
let ok = password::verify_async_in(HashLane::Credential, secret, &key.key_hash)
.await
.map_err(AuthError::from_hash)?;
if !ok {
attempt.failed().await;
return Ok(None);
}
if key.expires_at.is_some_and(|exp| chrono::Utc::now() > exp) {
return Err(AuthError::InvalidToken);
}
attempt.succeeded().await;
let users = super::auth::User::objects()
.where_(super::auth::User::id.eq(key.user_id))
.fetch(pool)
.await?;
let Some(user) = users.into_iter().next() else {
return Ok(None);
};
if !user.active {
return Err(AuthError::Inactive);
}
Ok(Some(AuthUser {
id: user.id.get().copied().unwrap_or(0),
username: user.username,
is_superuser: user.is_superuser,
}))
}
}
pub async fn create_api_key(
user_id: i64,
label: &str,
expires_at: Option<chrono::DateTime<chrono::Utc>>,
pool: &Pool,
) -> Result<String, crate::tenancy::error::TenancyError> {
use crate::sql::Auto;
use rand::rngs::OsRng;
use rand::RngCore;
let mut prefix_bytes = [0u8; 4];
OsRng.fill_bytes(&mut prefix_bytes);
let prefix = to_hex(&prefix_bytes);
let mut secret_bytes = [0u8; 16];
OsRng.fill_bytes(&mut secret_bytes);
let secret = to_hex(&secret_bytes);
let hash = password::hash_async(&secret)
.await
.map_err(|e| crate::tenancy::error::TenancyError::Validation(e.to_string()))?;
let mut key = ApiKey {
id: Auto::default(),
user_id,
key_prefix: prefix.clone(),
key_hash: hash,
label: label.to_owned(),
expires_at,
created_at: Auto::default(),
};
key.save_pool(pool).await?;
Ok(format!("{prefix}.{secret}"))
}
pub struct JwtBackend {
secret: Vec<u8>,
pub ttl_secs: i64,
jti_store: Option<std::sync::Arc<dyn crate::jti_store::JtiStore>>,
warned_unchecked: std::sync::atomic::AtomicBool,
}
impl JwtBackend {
#[must_use]
pub fn new(secret: Vec<u8>) -> Self {
assert!(
secret.len() >= 32,
"JwtBackend signing key is too short; need >= 32 bytes (a shorter key is forgeable)",
);
Self {
secret,
ttl_secs: 3600,
jti_store: None,
warned_unchecked: std::sync::atomic::AtomicBool::new(false),
}
}
#[must_use]
pub fn with_jti_store(mut self, store: std::sync::Arc<dyn crate::jti_store::JtiStore>) -> Self {
self.jti_store = Some(store);
self
}
fn warn_unchecked_revocation(&self) {
use std::sync::atomic::Ordering;
if !self.warned_unchecked.swap(true, Ordering::Relaxed) {
tracing::warn!(
target: "rustango::tenancy",
"JwtBackend has no JTI store, so tokens are not checked for revocation: a \
logged-out token works until it expires; call `with_jti_store`"
);
}
}
#[cfg(feature = "postgres")]
#[must_use]
pub fn from_session_secret(s: &super::operator_console::SessionSecret) -> Self {
Self::new(s.key().to_vec())
}
#[must_use]
pub fn issue(&self, user_id: i64) -> String {
self.sign_payload(&serde_json::json!({"sub": user_id, "exp": self.exp()}))
}
#[must_use]
pub fn issue_for_tenant(&self, user_id: i64, slug: &str) -> String {
let mut payload = serde_json::json!({"sub": user_id, "exp": self.exp()});
payload[super::jwt_lifecycle::CLAIM_TENANT] = slug.into();
self.sign_payload(&payload)
}
fn exp(&self) -> i64 {
chrono::Utc::now().timestamp() + self.ttl_secs
}
fn sign_payload(&self, payload: &serde_json::Value) -> String {
use base64::Engine;
let payload_b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(serde_json::to_vec(payload).unwrap_or_default());
let sig = hmac_sha256(&self.secret, payload_b64.as_bytes());
let sig_b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(sig);
format!("{payload_b64}.{sig_b64}")
}
fn verify_claims(
&self,
token: &str,
scope: super::jwt_lifecycle::UserTokenScope<'_>,
) -> Option<(i64, Option<String>)> {
use base64::Engine;
use subtle::ConstantTimeEq;
let b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD;
let parts: Vec<&str> = token.split('.').collect();
let (signing_input, payload_b64, sig_b64) = match parts.as_slice() {
[header, payload, sig] => {
let h: serde_json::Value =
serde_json::from_slice(&b64.decode(header).ok()?).ok()?;
if h.get("alg").and_then(serde_json::Value::as_str) != Some("HS256") {
return None;
}
(format!("{header}.{payload}"), *payload, *sig)
}
[payload, sig] => ((*payload).to_owned(), *payload, *sig),
_ => return None,
};
let expected = hmac_sha256(&self.secret, signing_input.as_bytes());
let provided = b64.decode(sig_b64).ok()?;
if expected.ct_eq(&provided[..]).unwrap_u8() == 0 {
return None;
}
let payload: serde_json::Value =
serde_json::from_slice(&b64.decode(payload_b64).ok()?).ok()?;
let exp = payload.get("exp")?.as_i64()?;
if chrono::Utc::now().timestamp() >= exp {
return None; }
if let Some(typ) = payload.get("typ").and_then(serde_json::Value::as_str) {
if typ != "access" {
return None;
}
}
scope.admits(payload.as_object()?).ok()?;
let sub = payload.get("sub")?.as_i64()?;
let jti = payload
.get("jti")
.and_then(serde_json::Value::as_str)
.map(str::to_owned);
Some((sub, jti))
}
}
fn hmac_sha256(secret: &[u8], msg: &[u8]) -> [u8; 32] {
use hmac::{Hmac, Mac};
use sha2::Sha256;
let mut mac = Hmac::<Sha256>::new_from_slice(secret).expect("HMAC accepts any key length");
mac.update(msg);
let bytes = mac.finalize().into_bytes();
let mut out = [0u8; 32];
out.copy_from_slice(&bytes[..32]);
out
}
#[async_trait]
impl AuthBackend for JwtBackend {
async fn authenticate(
&self,
parts: &Parts,
pool: &Pool,
) -> Result<Option<AuthUser>, AuthError> {
use crate::core::Column as _;
use crate::sql::FetcherPool as _;
let bearer = extract_bearer(parts)?;
let Some(token) = bearer else {
return Ok(None);
};
let dots = token.chars().filter(|&c| c == '.').count();
if dots != 1 && dots != 2 {
return Ok(None);
}
if dots == 1 && token.split_once('.').map(|(p, _)| p.len()) == Some(8) {
return Ok(None);
}
let scope = match parts.extensions.get::<super::TenantSlug>() {
Some(slug) => super::jwt_lifecycle::UserTokenScope::Tenant(&slug.0),
None => super::jwt_lifecycle::UserTokenScope::Unscoped,
};
let (user_id, jti) = match self.verify_claims(token, scope) {
Some(c) => c,
None => return Err(AuthError::InvalidToken),
};
match (self.jti_store.as_ref(), jti.as_deref()) {
(Some(store), Some(jti)) if store.is_used(jti).await => {
return Err(AuthError::InvalidToken);
}
(None, Some(_)) => self.warn_unchecked_revocation(),
_ => {}
}
let users = super::auth::User::objects()
.where_(super::auth::User::id.eq(user_id))
.fetch(pool)
.await?;
let Some(user) = users.into_iter().next() else {
return Ok(None);
};
if !user.active {
return Err(AuthError::Inactive);
}
Ok(Some(AuthUser {
id: user.id.get().copied().unwrap_or(0),
username: user.username,
is_superuser: user.is_superuser,
}))
}
}
fn to_hex(bytes: &[u8]) -> String {
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
fn extract_bearer(parts: &Parts) -> Result<Option<&str>, AuthError> {
let Some(value) = parts.headers.get(axum::http::header::AUTHORIZATION) else {
return Ok(None);
};
let s = value.to_str().unwrap_or("");
Ok(s.strip_prefix("Bearer ").map(str::trim))
}
#[cfg(all(test, feature = "sqlite"))]
mod tests {
use super::*;
async fn revocation_warnings(backend: &JwtBackend, token: &str) -> usize {
let buf = crate::testkit::CaptureWriter::default();
let subscriber = tracing_subscriber::fmt()
.with_writer(buf.clone())
.with_ansi(false)
.finish();
let _g = tracing::subscriber::set_default(subscriber);
let pool = Pool::connect("sqlite::memory:").await.unwrap();
for _ in 0..2 {
let (parts, ()) = axum::http::Request::builder()
.header("authorization", format!("Bearer {token}"))
.body(())
.unwrap()
.into_parts();
let _ = backend.authenticate(&parts, &pool).await;
}
buf.contents().matches("not checked for revocation").count()
}
#[tokio::test]
async fn a_revocable_token_without_a_jti_store_warns_once() {
let token = super::super::jwt_lifecycle::JwtLifecycle::new(vec![7u8; 32])
.issue_pair(42)
.access;
let backend = JwtBackend::new(vec![7u8; 32]);
assert_eq!(revocation_warnings(&backend, &token).await, 1);
}
#[tokio::test]
async fn a_token_without_a_jti_stays_silent() {
let backend = JwtBackend::new(vec![7u8; 32]);
let token = backend.issue(42);
assert_eq!(revocation_warnings(&backend, &token).await, 0);
}
}