mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-04 08:02:23 +00:00
AlexRV12andClaude Opus 5
42655ff5ef
* fix: memoize the resolved authed so one request resolves identity once Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * test: pin that the memo preserves job token provenance Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
1385 lines
61 KiB
Rust
1385 lines
61 KiB
Rust
#[cfg(feature = "enterprise")]
|
|
use crate::ee_oss::ExternalJwks;
|
|
use axum::{
|
|
extract::{FromRequestParts, OriginalUri, Query},
|
|
Extension, Json,
|
|
};
|
|
use chrono::TimeZone;
|
|
use http::{request::Parts, StatusCode};
|
|
use quick_cache::sync::Cache;
|
|
use serde::{Deserialize, Serialize};
|
|
use sqlx::FromRow;
|
|
use tower_cookies::Cookies;
|
|
use tracing::Span;
|
|
|
|
use crate::{ApiAuthed, OptJobAuthed};
|
|
use std::{
|
|
str::FromStr,
|
|
sync::{
|
|
atomic::{AtomicI64, AtomicU64, Ordering},
|
|
Arc,
|
|
},
|
|
};
|
|
#[cfg(feature = "enterprise")]
|
|
use tokio::sync::RwLock;
|
|
use windmill_common::DB;
|
|
|
|
use windmill_common::{
|
|
auth::{
|
|
get_folders_for_user, get_groups_for_user, hash_token, is_session_label, safe_token_prefix,
|
|
JWTAuthClaims,
|
|
},
|
|
error::{Error, JsonResult},
|
|
jwt,
|
|
usernames::get_instance_username_or_fallback_to_email,
|
|
users::{COOKIE_NAME, SUPERADMIN_SECRET_EMAIL},
|
|
};
|
|
|
|
lazy_static::lazy_static! {
|
|
// Global auth cache accessible from main.rs for direct invalidation
|
|
pub static ref AUTH_CACHE: Cache<(String, String), ExpiringAuthCache> = Cache::new(300);
|
|
// Cache for token -> email lookups (for non-workspace-member authenticated users)
|
|
static ref TOKEN_EMAIL_CACHE: Cache<String, (Option<String>, std::time::Instant)> = Cache::new(500);
|
|
}
|
|
|
|
/// A token keeps its identity when a superadmin moves the account to another address, so entries
|
|
/// here must expire on their own; nothing invalidates them by token hash.
|
|
const TOKEN_EMAIL_CACHE_TTL_SECS: u64 = 60;
|
|
|
|
/// Get email from a valid token, with caching.
|
|
/// Used for WM_END_USER_EMAIL when user is authenticated but not a workspace member.
|
|
async fn get_email_from_token(db: &DB, token: &str) -> Option<String> {
|
|
let t_hash = hash_token(token);
|
|
if let Some((cached, cached_at)) = TOKEN_EMAIL_CACHE.get(&t_hash) {
|
|
if cached_at.elapsed().as_secs() < TOKEN_EMAIL_CACHE_TTL_SECS {
|
|
return cached;
|
|
}
|
|
}
|
|
|
|
let email = sqlx::query_scalar!(
|
|
"SELECT email FROM token WHERE token_hash = $1 AND (expiration > NOW() OR expiration IS NULL)",
|
|
t_hash
|
|
)
|
|
.fetch_optional(db)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.flatten(); // email column is nullable, so we get Option<Option<String>>
|
|
|
|
TOKEN_EMAIL_CACHE.insert(t_hash, (email.clone(), std::time::Instant::now()));
|
|
email
|
|
}
|
|
|
|
/// Get end user email from authenticated user or token.
|
|
/// Returns email if user is authenticated (workspace member) or has valid instance token.
|
|
pub async fn get_end_user_email(
|
|
db: &DB,
|
|
opt_authed: Option<&ApiAuthed>,
|
|
token: Option<&str>,
|
|
) -> Option<String> {
|
|
if let Some(authed) = opt_authed {
|
|
return Some(authed.email.clone());
|
|
}
|
|
if let Some(token) = token {
|
|
return get_email_from_token(db, token).await;
|
|
}
|
|
None
|
|
}
|
|
// Global function to invalidate tokens from cache by prefix
|
|
pub fn invalidate_token_from_cache(token_prefix: &str) {
|
|
// Remove all cache entries whose raw token starts with this prefix (across all workspaces)
|
|
AUTH_CACHE.retain(|(_workspace_id, cached_token), _cached_value| {
|
|
!cached_token.starts_with(token_prefix)
|
|
});
|
|
tracing::info!(
|
|
"Invalidated token(s) from auth cache with prefix: {}...",
|
|
&token_prefix[..token_prefix.len().min(8)]
|
|
);
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct ExpiringAuthCache {
|
|
pub authed: ApiAuthed,
|
|
pub expiry: chrono::DateTime<chrono::Utc>,
|
|
pub job_id: Option<uuid::Uuid>,
|
|
}
|
|
|
|
pub struct AuthCache {
|
|
db: DB,
|
|
superadmin_secret: Option<String>,
|
|
#[cfg(feature = "enterprise")]
|
|
ext_jwks: Option<Arc<RwLock<ExternalJwks>>>,
|
|
}
|
|
|
|
impl AuthCache {
|
|
pub fn new(
|
|
db: DB,
|
|
superadmin_secret: Option<String>,
|
|
#[cfg(feature = "enterprise")] ext_jwks: Option<Arc<RwLock<ExternalJwks>>>,
|
|
) -> Self {
|
|
AuthCache {
|
|
db,
|
|
superadmin_secret,
|
|
#[cfg(feature = "enterprise")]
|
|
ext_jwks,
|
|
}
|
|
}
|
|
|
|
pub async fn invalidate(&self, w_id: &str, token: String) {
|
|
AUTH_CACHE.remove(&(w_id.to_string(), token));
|
|
}
|
|
|
|
pub async fn get_authed(&self, w_id: Option<String>, token: &str) -> Option<ApiAuthed> {
|
|
Some(self.get_opt_job_authed(w_id, token).await?.authed)
|
|
}
|
|
|
|
pub async fn get_opt_job_authed(
|
|
&self,
|
|
w_id: Option<String>,
|
|
token: &str,
|
|
) -> Option<OptJobAuthed> {
|
|
let mut opt_job_authed = self.get_opt_job_authed_inner(w_id.clone(), token).await?;
|
|
// Single source of truth: mirror the resolved job_id onto the authed so
|
|
// every consumer (require_super_admin, ...) sees that this identity came
|
|
// from a job's WM_TOKEN, even on an AUTH_CACHE hit whose cached authed
|
|
// predates this field.
|
|
opt_job_authed.authed.job_id = opt_job_authed.job_id;
|
|
// The workspace's guest switch is enforced here, once, for every guest request
|
|
// — not per handler, where each guest-reachable route would have to remember
|
|
// it. Uncached, so turning guests off takes effect on the next request of every
|
|
// guest session and every token derived from one.
|
|
if crate::scopes::has_guest_sentinel(opt_job_authed.authed.scopes.as_deref()) {
|
|
let Some(w_id) = w_id else { return None };
|
|
let email = &opt_job_authed.authed.email;
|
|
match windmill_common::workspaces::guest_session_stands(&self.db, &w_id, email).await {
|
|
Ok(true) => {}
|
|
Ok(false) => return None,
|
|
Err(e) => {
|
|
tracing::error!("guest session check failed for {w_id}: {e:#}");
|
|
return None;
|
|
}
|
|
}
|
|
}
|
|
Some(opt_job_authed)
|
|
}
|
|
|
|
async fn get_opt_job_authed_inner(
|
|
&self,
|
|
w_id: Option<String>,
|
|
token: &str,
|
|
) -> Option<OptJobAuthed> {
|
|
// In no-auth mode there are no real tokens: resolve directly as the
|
|
// admin superadmin so direct cache callers (e.g. get_all_runnables,
|
|
// which re-validates the request token per workspace) don't reject the
|
|
// fabricated token.
|
|
if is_no_auth() {
|
|
return Some(OptJobAuthed { authed: no_auth_admin_authed(), job_id: None });
|
|
}
|
|
// Reject an oversized guest bearer before the cache key is built from it: the key
|
|
// copies and hashes the whole token, so the cap should bound that work too. Log it
|
|
// like the other guest refusals, since get_opt_job_authed turns None into a bare 401.
|
|
if token.starts_with(windmill_common::guest_jwt::BEARER_PREFIX)
|
|
&& token.len() > windmill_common::guest_jwt::MAX_GUEST_JWT_LEN
|
|
{
|
|
tracing::error!(
|
|
"guest JWT refused: bearer is longer than {} bytes",
|
|
windmill_common::guest_jwt::MAX_GUEST_JWT_LEN
|
|
);
|
|
return None;
|
|
}
|
|
let key = (
|
|
w_id.as_ref().unwrap_or(&"".to_string()).to_string(),
|
|
token.to_string(),
|
|
);
|
|
let s = AUTH_CACHE.get(&key).map(|c| c.to_owned());
|
|
match s {
|
|
Some(ExpiringAuthCache { authed, expiry, job_id }) if expiry > chrono::Utc::now() => {
|
|
Some(OptJobAuthed { authed, job_id })
|
|
}
|
|
#[cfg(feature = "enterprise")]
|
|
_ if token.starts_with("jwt_ext_") => {
|
|
let authed_and_exp = match crate::ee_oss::jwt_ext_auth(
|
|
w_id.as_ref(),
|
|
token.trim_start_matches("jwt_ext_"),
|
|
self.ext_jwks.clone(),
|
|
&self.db,
|
|
)
|
|
.await
|
|
{
|
|
Ok(r) => Some(r),
|
|
Err(e) => {
|
|
tracing::error!("JWT_EXT auth error: {:?}", e);
|
|
None
|
|
}
|
|
};
|
|
|
|
if let Some((authed, exp, job_id)) = authed_and_exp.clone() {
|
|
AUTH_CACHE.insert(
|
|
key,
|
|
ExpiringAuthCache {
|
|
authed: authed.clone(),
|
|
expiry: chrono::Utc.timestamp_nanos(exp as i64 * 1_000_000_000),
|
|
job_id,
|
|
},
|
|
);
|
|
|
|
Some(OptJobAuthed { authed, job_id })
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
_ if token.starts_with(windmill_common::guest_jwt::BEARER_PREFIX) => {
|
|
// A workspace-less route never accepts a guest JWT: the identity is
|
|
// pinned to the workspace its claim names, like a DB guest session.
|
|
let Some(w_id) = w_id.as_deref() else {
|
|
return None;
|
|
};
|
|
// Strip exactly one prefix: `trim_start_matches` would strip repeated prefixes,
|
|
// so `jwt_guest_jwt_guest_<jwt>` would reduce to a valid token that verifies and
|
|
// is then cached under the full, non-canonical bearer key.
|
|
let jwt = token
|
|
.strip_prefix(windmill_common::guest_jwt::BEARER_PREFIX)
|
|
.unwrap_or(token);
|
|
let claims =
|
|
match windmill_common::guest_jwt::verify_for_workspace(&self.db, w_id, jwt)
|
|
.await
|
|
{
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
tracing::error!("guest JWT auth error for {w_id}: {e:#}");
|
|
return None;
|
|
}
|
|
};
|
|
// The workspace switch, the instance switch and the app being in guest
|
|
// mode, in one answer (guest_app_admits). The door re-reads the switches
|
|
// and the no-account rule per request through the sentinel below
|
|
// (guest_session_stands), so turning any of them off stops a cached JWT
|
|
// session on its next call.
|
|
match windmill_common::workspaces::guest_app_admits(
|
|
&self.db,
|
|
w_id,
|
|
&claims.app_path,
|
|
)
|
|
.await
|
|
{
|
|
Ok(true) => {}
|
|
Ok(false) => return None,
|
|
Err(e) => {
|
|
tracing::error!("guest JWT admit check failed for {w_id}: {e:#}");
|
|
return None;
|
|
}
|
|
}
|
|
// Resolve on the lowercased email: accounts are stored lowercased, so a
|
|
// mixed-case claim would otherwise slip past the no-account gate and
|
|
// resolve an account holder to a guest, and split the activity rows the
|
|
// seat count reads.
|
|
let email = claims.email.to_lowercase();
|
|
// A guest is someone with no account at all; an account holder is refused,
|
|
// never downgraded (the same rule as the signed-in guest mint).
|
|
match windmill_common::users::has_any_account(&self.db, &email).await {
|
|
Ok(false) => {}
|
|
Ok(true) => return None,
|
|
Err(e) => {
|
|
tracing::error!("guest JWT account check failed: {e:#}");
|
|
return None;
|
|
}
|
|
}
|
|
// The instance allowance, checked and recorded transactionally. A stranger
|
|
// past the cap on a capped instance is refused here; a returning guest
|
|
// always passes. Recording an account holder is avoided by the check above.
|
|
if !admit_and_record_guest_jwt(&self.db, w_id, &email, &claims.app_path).await {
|
|
return None;
|
|
}
|
|
// guest_session_scopes already carries the sentinel, and it is the whole
|
|
// grant; a JWT has no label, so the sentinel is what governs it. It also
|
|
// re-checks the path holds no scope metacharacter (verify already did).
|
|
let scopes = match crate::scopes::guest_session_scopes(&claims.app_path) {
|
|
Ok(s) => Some(s),
|
|
Err(e) => {
|
|
tracing::error!("guest JWT app_path cannot be scoped for {w_id}: {e:#}");
|
|
return None;
|
|
}
|
|
};
|
|
// The JWT's own expiry caps a token minted from this session. The auth
|
|
// cache entry itself is capped far shorter (GUEST_JWT_CACHE_TTL) so a
|
|
// rotated or cleared key stops the session on re-verification, within
|
|
// minutes, rather than only at exp (up to 24h away).
|
|
let credential_expiry =
|
|
chrono::Utc.timestamp_nanos(claims.exp as i64 * 1_000_000_000);
|
|
let cache_expiry = credential_expiry.min(chrono::Utc::now() + GUEST_JWT_CACHE_TTL);
|
|
let authed = ApiAuthed {
|
|
username: email.clone(),
|
|
email,
|
|
is_admin: false,
|
|
is_operator: true,
|
|
groups: vec![],
|
|
folders: vec![],
|
|
scopes,
|
|
username_override: None,
|
|
username_override_is_token_label: false,
|
|
is_session_token: false,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only: false,
|
|
job_id: None,
|
|
credential_expiry: Some(credential_expiry),
|
|
};
|
|
AUTH_CACHE.insert(
|
|
key,
|
|
ExpiringAuthCache {
|
|
authed: authed.clone(),
|
|
expiry: cache_expiry,
|
|
job_id: None,
|
|
},
|
|
);
|
|
Some(OptJobAuthed { authed, job_id: None })
|
|
}
|
|
_ if token.starts_with("jwt_") => {
|
|
let jwt_token = token.trim_start_matches("jwt_");
|
|
|
|
let jwt_result = jwt::decode_with_internal_secret::<JWTAuthClaims>(jwt_token).await;
|
|
|
|
match jwt_result {
|
|
Ok(claims) => {
|
|
if w_id.is_some_and(|w_id| !claims.allowed_in_workspace(&w_id)) {
|
|
tracing::error!("JWT auth error: workspace_id mismatch");
|
|
return None;
|
|
}
|
|
let is_session_token = is_session_label(claims.label.as_deref());
|
|
let (username_override, username_override_is_token_label) =
|
|
username_override_from_label(claims.label);
|
|
|
|
let authed = ApiAuthed {
|
|
email: claims.email,
|
|
username: claims.username,
|
|
is_admin: claims.is_admin,
|
|
is_operator: claims.is_operator,
|
|
groups: claims.groups,
|
|
folders: claims.folders,
|
|
// Honor the scopes embedded in the JWT (mirrors the EE
|
|
// jwt_ext_ branch). The route middleware only enforces
|
|
// scopes when Some, so a None-scoped JWT (e.g. the job
|
|
// WM_TOKEN) keeps full user privileges as before.
|
|
scopes: claims.scopes,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: claims.audit_span,
|
|
read_only: false,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
};
|
|
// Fail closed: a `job_id` claim that does not parse must reject
|
|
// the token rather than resolve to `None`, which would clear the
|
|
// job provenance and uncap the token (GHSA-hfh4-cx4h-3fcr).
|
|
let job_id = match claims.job_id {
|
|
Some(j) => match uuid::Uuid::from_str(&j) {
|
|
Ok(job_id) => Some(job_id),
|
|
Err(_) => {
|
|
tracing::error!("JWT auth error: job_id claim is not a uuid");
|
|
return None;
|
|
}
|
|
},
|
|
None => None,
|
|
};
|
|
AUTH_CACHE.insert(
|
|
key,
|
|
ExpiringAuthCache {
|
|
authed: authed.clone(),
|
|
expiry: chrono::Utc
|
|
.timestamp_nanos(claims.exp as i64 * 1_000_000_000),
|
|
job_id,
|
|
},
|
|
);
|
|
|
|
Some(OptJobAuthed { authed, job_id })
|
|
}
|
|
Err(err) => {
|
|
tracing::error!("JWT auth error: {:?}", err);
|
|
None
|
|
}
|
|
}
|
|
}
|
|
_ => {
|
|
let t_hash = hash_token(token);
|
|
let user_o = sqlx::query!(
|
|
"UPDATE token SET last_used_at = now() WHERE
|
|
token_hash = $1
|
|
AND (expiration > NOW() OR expiration IS NULL)
|
|
AND (workspace_id IS NULL OR workspace_id = $2)
|
|
RETURNING owner, email, super_admin, scopes, label, read_only",
|
|
t_hash,
|
|
w_id.as_ref(),
|
|
)
|
|
.map(|x| {
|
|
(
|
|
x.owner,
|
|
x.email,
|
|
x.super_admin,
|
|
x.scopes,
|
|
x.label,
|
|
x.read_only,
|
|
)
|
|
})
|
|
.fetch_optional(&self.db)
|
|
.await
|
|
.ok()
|
|
.flatten();
|
|
|
|
if let Some(user) = user_o {
|
|
let authed_o = {
|
|
match user {
|
|
(Some(owner), Some(email), super_admin, _, label, read_only)
|
|
if w_id.is_some() =>
|
|
{
|
|
let is_session_token = is_session_label(label.as_deref());
|
|
let (username_override, username_override_is_token_label) =
|
|
username_override_from_label(label);
|
|
if let Some((prefix, name)) = owner.split_once('/') {
|
|
if prefix == "u" {
|
|
let lookup = if super_admin {
|
|
Some((true, false))
|
|
} else {
|
|
sqlx::query!(
|
|
"SELECT is_admin, operator FROM usr where username = $1 AND \
|
|
workspace_id = $2 AND disabled = false",
|
|
name,
|
|
&w_id.as_ref().unwrap()
|
|
)
|
|
.fetch_optional(&self.db)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.map(|r| (r.is_admin, r.operator))
|
|
};
|
|
|
|
if let Some((is_admin, is_operator)) = lookup {
|
|
let w_id = &w_id.unwrap();
|
|
let groups =
|
|
get_groups_for_user(w_id, &name, &email, &self.db)
|
|
.await
|
|
.ok()
|
|
.unwrap_or_default();
|
|
|
|
let folders = get_folders_for_user(
|
|
w_id, &name, &groups, &self.db,
|
|
)
|
|
.await
|
|
.ok()
|
|
.unwrap_or_default();
|
|
|
|
Some(ApiAuthed {
|
|
email: email,
|
|
username: name.to_string(),
|
|
is_admin,
|
|
is_operator,
|
|
groups,
|
|
folders,
|
|
scopes: None,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
})
|
|
} else {
|
|
tracing::warn!(
|
|
"Token owner u/{} is not a member of workspace {}; rejecting auth",
|
|
name,
|
|
w_id.as_deref().unwrap_or("")
|
|
);
|
|
None
|
|
}
|
|
} else if prefix == "g" {
|
|
let group_exists = if super_admin {
|
|
true
|
|
} else {
|
|
sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM group_ WHERE workspace_id = $1 AND name = $2)",
|
|
&w_id.as_ref().unwrap(),
|
|
name,
|
|
)
|
|
.fetch_one(&self.db)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or(false)
|
|
};
|
|
|
|
if group_exists {
|
|
let groups = vec![name.to_string()];
|
|
let folders = get_folders_for_user(
|
|
&w_id.unwrap(),
|
|
"",
|
|
&groups,
|
|
&self.db,
|
|
)
|
|
.await
|
|
.ok()
|
|
.unwrap_or_default();
|
|
Some(ApiAuthed {
|
|
email: email,
|
|
username: format!(
|
|
"{}{name}",
|
|
windmill_common::users::USERNAME_GROUP_PREFIX
|
|
),
|
|
is_admin: false,
|
|
groups,
|
|
is_operator: false,
|
|
folders,
|
|
scopes: None,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
})
|
|
} else {
|
|
tracing::warn!(
|
|
"Token owner g/{} is not a group in workspace {}; rejecting auth",
|
|
name,
|
|
w_id.as_deref().unwrap_or("")
|
|
);
|
|
None
|
|
}
|
|
} else {
|
|
tracing::warn!(
|
|
"Token owner '{}' has unrecognised prefix '{}'; rejecting auth",
|
|
owner,
|
|
prefix
|
|
);
|
|
None
|
|
}
|
|
} else {
|
|
tracing::warn!(
|
|
"Token owner '{}' is missing a prefix (expected u/ or g/); rejecting auth",
|
|
owner
|
|
);
|
|
None
|
|
}
|
|
}
|
|
(_, Some(email), super_admin, scopes, label, read_only) => {
|
|
let is_session_token = is_session_label(label.as_deref());
|
|
let is_guest_session =
|
|
windmill_common::auth::is_guest_session_label(label.as_deref());
|
|
let (username_override, username_override_is_token_label) =
|
|
username_override_from_label(label);
|
|
if w_id.is_some() {
|
|
let row_o = sqlx::query!(
|
|
"SELECT username, is_admin, operator FROM usr WHERE
|
|
email = $1 AND workspace_id = $2 AND disabled = false",
|
|
&email,
|
|
w_id.as_ref().unwrap()
|
|
)
|
|
.map(|x| (x.username, x.is_admin, x.operator))
|
|
.fetch_optional(&self.db)
|
|
.await
|
|
.unwrap_or(Some(("error".to_string(), false, false)));
|
|
|
|
match row_o {
|
|
Some((username, is_admin, is_operator)) => {
|
|
let groups = get_groups_for_user(
|
|
&w_id.as_ref().unwrap(),
|
|
&username,
|
|
&email,
|
|
&self.db,
|
|
)
|
|
.await
|
|
.ok()
|
|
.unwrap_or_default();
|
|
|
|
let folders = get_folders_for_user(
|
|
&w_id.unwrap(),
|
|
&username,
|
|
&groups,
|
|
&self.db,
|
|
)
|
|
.await
|
|
.ok()
|
|
.unwrap_or_default();
|
|
Some(ApiAuthed {
|
|
email,
|
|
username,
|
|
is_admin: is_admin || super_admin,
|
|
is_operator,
|
|
groups,
|
|
folders,
|
|
scopes,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
})
|
|
}
|
|
None if super_admin => {
|
|
// Fail closed on a DB error rather than
|
|
// letting the email leak in as the username.
|
|
match get_instance_username_or_fallback_to_email(
|
|
&self.db, &email,
|
|
)
|
|
.await
|
|
{
|
|
Ok(username) => Some(ApiAuthed {
|
|
email,
|
|
username,
|
|
is_admin: super_admin,
|
|
is_operator: false,
|
|
groups: vec![],
|
|
folders: vec![],
|
|
scopes,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
}),
|
|
Err(e) => {
|
|
tracing::error!(
|
|
"Failed to resolve instance username for superadmin {email}: {e:#}"
|
|
);
|
|
None
|
|
}
|
|
}
|
|
}
|
|
// A guest session: IdP-authenticated, member of
|
|
// nothing. No `usr` lookup, groups or folders, so
|
|
// every ACL denies it and the token's scopes are
|
|
// its whole grant. After the superadmin arm, so
|
|
// that token is never demoted into this one.
|
|
None if is_guest_session => {
|
|
// The server-minted label is the grant, never
|
|
// the `guest` scope (a user-minted token's
|
|
// scopes are whatever the caller typed); the
|
|
// sentinel is pinned on here so every guest
|
|
// control downstream sees a guest regardless.
|
|
let scopes = Some(crate::scopes::with_guest_sentinel(
|
|
scopes.unwrap_or_default(),
|
|
));
|
|
Some(ApiAuthed {
|
|
username: email.clone(),
|
|
email,
|
|
is_admin: false,
|
|
is_operator: true,
|
|
groups: vec![],
|
|
folders: vec![],
|
|
scopes,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
})
|
|
}
|
|
None => None,
|
|
}
|
|
} else {
|
|
Some(ApiAuthed {
|
|
email: email.to_string(),
|
|
username: email,
|
|
is_admin: super_admin,
|
|
is_operator: true,
|
|
groups: Vec::new(),
|
|
folders: Vec::new(),
|
|
scopes,
|
|
username_override,
|
|
username_override_is_token_label,
|
|
is_session_token,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
})
|
|
}
|
|
}
|
|
_ => None,
|
|
}
|
|
};
|
|
if let Some(authed) = authed_o.as_ref() {
|
|
AUTH_CACHE.insert(
|
|
key,
|
|
ExpiringAuthCache {
|
|
authed: authed.clone(),
|
|
expiry: chrono::Utc::now()
|
|
+ chrono::Duration::try_seconds(120).unwrap(),
|
|
job_id: None,
|
|
},
|
|
);
|
|
}
|
|
authed_o.map(|authed| OptJobAuthed { authed, job_id: None })
|
|
} else if self
|
|
.superadmin_secret
|
|
.as_ref()
|
|
.map(|x| x == token)
|
|
.unwrap_or(false)
|
|
{
|
|
let authed = ApiAuthed {
|
|
email: SUPERADMIN_SECRET_EMAIL.to_string(),
|
|
username: "superadmin_secret".to_string(),
|
|
is_admin: true,
|
|
is_operator: false,
|
|
groups: Vec::new(),
|
|
folders: Vec::new(),
|
|
scopes: None,
|
|
username_override: None,
|
|
username_override_is_token_label: false,
|
|
is_session_token: false,
|
|
token_prefix: Some(safe_token_prefix(token)),
|
|
read_only: false,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
};
|
|
Some(OptJobAuthed { authed, job_id: None })
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// How long a guest JWT resolves from the auth cache before the arm re-runs (and
|
|
/// re-reads the key). A guest JWT is not revocable except by the workspace switch or
|
|
/// by rotating the key, so the entry must be short enough that a rotated key bites
|
|
/// soon, unlike a normal token whose row can be deleted. Also what makes the
|
|
/// day-keyed activity dedupe below reachable across a midnight.
|
|
const GUEST_JWT_CACHE_TTL: chrono::Duration = chrono::Duration::minutes(5);
|
|
|
|
/// A refused JWT (a stranger past the allowance) is remembered this long so a replayed
|
|
/// bearer does not take the instance-wide allowance advisory lock on every request.
|
|
/// Short, so a stranger admitted once the window frees is re-checked soon.
|
|
const GUEST_JWT_REFUSED_TTL: std::time::Duration = std::time::Duration::from_secs(30);
|
|
|
|
lazy_static::lazy_static! {
|
|
// One `guest_activity` upsert and one `users.login_guest` audit per email,
|
|
// workspace and day: the arm re-runs every GUEST_JWT_CACHE_TTL, and neither the
|
|
// seat scan nor the audit trail wants a write each time. LRU-bounded; the day is in
|
|
// the key, so a new day writes again.
|
|
static ref GUEST_JWT_ACTIVITY_CACHE: Cache<String, ()> = Cache::new(2000);
|
|
static ref GUEST_JWT_REFUSED_CACHE: Cache<String, std::time::Instant> = Cache::new(2000);
|
|
}
|
|
|
|
/// Admit a JWT guest against the instance allowance and record today's activity, in one
|
|
/// transaction so the advisory lock in `guest_admission` spans the count check and the
|
|
/// row that changes it. Returns false when the allowance refuses the email or on a DB
|
|
/// error, both of which deny the guest. Cached per email, workspace and day: a bearer
|
|
/// replayed every request runs this at most once a day, and a refused one is remembered
|
|
/// briefly so it does not re-take the allowance lock. `email` is already lowercased.
|
|
async fn admit_and_record_guest_jwt(db: &DB, w_id: &str, email: &str, app_path: &str) -> bool {
|
|
let cache_key = format!("{email}|{w_id}|{}", chrono::Utc::now().date_naive());
|
|
if GUEST_JWT_ACTIVITY_CACHE.get(&cache_key).is_some() {
|
|
return true;
|
|
}
|
|
if GUEST_JWT_REFUSED_CACHE
|
|
.get(&cache_key)
|
|
.is_some_and(|at| at.elapsed() < GUEST_JWT_REFUSED_TTL)
|
|
{
|
|
return false;
|
|
}
|
|
let mut tx = match db.begin().await {
|
|
Ok(tx) => tx,
|
|
Err(e) => {
|
|
tracing::error!("guest JWT tx begin failed for {w_id}: {e:#}");
|
|
return false;
|
|
}
|
|
};
|
|
// The allowance and the row that changes it, in one transaction: guest_admission
|
|
// takes a transaction-scoped advisory lock, so the count check and the insert cannot
|
|
// race two strangers past the cap. Only a real allowance refusal is negative-cached;
|
|
// a transient DB error denies this request but must not lock the email out for 30s.
|
|
match windmill_common::workspaces::guest_admission(&mut *tx, email).await {
|
|
Ok(()) => {}
|
|
Err(e @ windmill_common::error::Error::PermissionDenied(_)) => {
|
|
// The guest hits a bare 401 (the reason must not leak to an unauthenticated caller);
|
|
// warn so an admin sees the cap in logs, since it is the actionable signal here.
|
|
tracing::warn!("guest JWT refused (guest allowance) for {w_id}: {e:#}");
|
|
GUEST_JWT_REFUSED_CACHE.insert(cache_key, std::time::Instant::now());
|
|
return false;
|
|
}
|
|
Err(e) => {
|
|
tracing::error!("guest JWT allowance check failed for {w_id}: {e:#}");
|
|
return false;
|
|
}
|
|
}
|
|
// The conditional `WHERE NOT jwt_entry` flips the flag only on its false-to-true
|
|
// transition, so the upsert returns a row exactly once per email per day: on the
|
|
// fresh insert, or on the first JWT after an identity-provider sign-in created
|
|
// today's row with `jwt_entry = false`. The audit is gated on that, decided
|
|
// atomically by the conflicting tuple, so concurrent first requests (a metered
|
|
// instance takes no advisory lock) audit at most once.
|
|
let first_jwt = sqlx::query_scalar!(
|
|
r#"INSERT INTO guest_activity (email, workspace_id, day, jwt_entry)
|
|
VALUES ($1, $2, CURRENT_DATE, true)
|
|
ON CONFLICT (email, workspace_id, day)
|
|
DO UPDATE SET jwt_entry = true, last_seen_at = now()
|
|
WHERE NOT guest_activity.jwt_entry
|
|
RETURNING 1 AS "audited!""#,
|
|
email,
|
|
w_id,
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.await;
|
|
let first_jwt = match first_jwt {
|
|
Ok(v) => v.is_some(),
|
|
Err(e) => {
|
|
tracing::error!("recording guest JWT activity for {w_id}: {e:#}");
|
|
return false;
|
|
}
|
|
};
|
|
if let Err(e) = tx.commit().await {
|
|
tracing::error!("guest JWT tx commit failed for {w_id}: {e:#}");
|
|
return false;
|
|
}
|
|
GUEST_JWT_ACTIVITY_CACHE.insert(cache_key, ());
|
|
// Audit last, best-effort, on its own connection: the EE writer swallows an
|
|
// `audit_partitioned` failure but that failing statement still aborts the
|
|
// transaction it runs in, so auditing before the commit would let the whole
|
|
// activity row roll back while this returned success, admitting an uncounted guest.
|
|
if first_jwt {
|
|
let author = windmill_common::audit::AuditAuthor {
|
|
email: email.to_string(),
|
|
username: email.to_string(),
|
|
username_override: None,
|
|
token_prefix: None,
|
|
};
|
|
if let Err(e) = windmill_audit::audit_oss::audit_log(
|
|
db,
|
|
&author,
|
|
"users.login_guest",
|
|
windmill_audit::ActionKind::Create,
|
|
w_id,
|
|
Some(app_path),
|
|
Some([("entry", "jwt")].into()),
|
|
)
|
|
.await
|
|
{
|
|
tracing::error!("auditing guest JWT login for {w_id}: {e:#}");
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
pub(crate) async fn extract_token<S: Send + Sync>(parts: &mut Parts, state: &S) -> Option<String> {
|
|
let auth_header = parts
|
|
.headers
|
|
.get(http::header::AUTHORIZATION)
|
|
.and_then(|value| value.to_str().ok())
|
|
.and_then(|s| s.strip_prefix("Bearer "));
|
|
|
|
let from_cookie = match auth_header {
|
|
Some(x) => Some(x.to_owned()),
|
|
None => Extension::<Cookies>::from_request_parts(parts, state)
|
|
.await
|
|
.ok()
|
|
.and_then(|cookies| {
|
|
cookies
|
|
.get(COOKIE_NAME)
|
|
.map(|c| c.value_trimmed().to_owned())
|
|
}),
|
|
};
|
|
|
|
#[derive(Deserialize)]
|
|
struct Token {
|
|
token: Option<String>,
|
|
}
|
|
match from_cookie {
|
|
Some(token) => Some(token),
|
|
None => Query::<Token>::from_request_parts(parts, state)
|
|
.await
|
|
.ok()
|
|
.and_then(|token| token.token.clone()),
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct Tokened {
|
|
pub token: String,
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct OptTokened {
|
|
#[allow(dead_code)]
|
|
pub token: Option<String>,
|
|
}
|
|
|
|
struct BruteForceCounter {
|
|
counter: AtomicU64,
|
|
last_reset: AtomicI64,
|
|
}
|
|
|
|
lazy_static::lazy_static! {
|
|
static ref BRUTE_FORCE_COUNTER: BruteForceCounter =
|
|
BruteForceCounter { last_reset: AtomicI64::new(0), counter: AtomicU64::new(0) };
|
|
}
|
|
|
|
impl BruteForceCounter {
|
|
async fn increment(&self) {
|
|
let now = time::OffsetDateTime::now_utc().unix_timestamp();
|
|
if self.counter.fetch_add(1, Ordering::Relaxed) > 10000 {
|
|
tracing::error!(
|
|
"Brute force attack to find valid token detected, sleeping unauthorized response for 2 seconds"
|
|
);
|
|
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
|
|
}
|
|
if now - self.last_reset.load(Ordering::Relaxed) > 60 {
|
|
self.counter.store(0, Ordering::Relaxed);
|
|
self.last_reset.store(now, Ordering::Relaxed);
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<S> FromRequestParts<S> for Tokened
|
|
where
|
|
S: Send + Sync,
|
|
{
|
|
type Rejection = (StatusCode, String);
|
|
|
|
async fn from_request_parts(
|
|
parts: &mut Parts,
|
|
state: &S,
|
|
) -> std::result::Result<Self, Self::Rejection> {
|
|
if parts.method == http::Method::OPTIONS {
|
|
return Ok(Tokened { token: "".to_string() });
|
|
};
|
|
let already_tokened = parts.extensions.get::<Tokened>();
|
|
if let Some(tokened) = already_tokened {
|
|
Ok(tokened.clone())
|
|
} else {
|
|
let token_o = extract_token(parts, state).await;
|
|
if let Some(token) = token_o {
|
|
let tokened = Self { token };
|
|
parts.extensions.insert(tokened.clone());
|
|
Ok(tokened)
|
|
} else if is_no_auth() {
|
|
// In `--no-auth` mode requests carry no token, but handlers that
|
|
// also require Tokened (e.g. global_whoami) must still resolve.
|
|
let tokened = Self { token: "no_auth".to_string() };
|
|
parts.extensions.insert(tokened.clone());
|
|
Ok(tokened)
|
|
} else {
|
|
BRUTE_FORCE_COUNTER.increment().await;
|
|
Err((StatusCode::UNAUTHORIZED, "Unauthorized".to_owned()))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<S> FromRequestParts<S> for OptTokened
|
|
where
|
|
S: Send + Sync,
|
|
{
|
|
type Rejection = (StatusCode, String);
|
|
|
|
async fn from_request_parts(
|
|
parts: &mut Parts,
|
|
state: &S,
|
|
) -> std::result::Result<Self, Self::Rejection> {
|
|
if parts.method == http::Method::OPTIONS {
|
|
return Ok(OptTokened { token: None });
|
|
};
|
|
let already_tokened = parts.extensions.get::<Tokened>();
|
|
if let Some(tokened) = already_tokened {
|
|
Ok(OptTokened { token: Some(tokened.token.clone()) })
|
|
} else {
|
|
let token_o = extract_token(parts, state).await;
|
|
Ok(OptTokened { token: token_o })
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn transform_old_scope_to_new_scope(scopes: Option<&mut Vec<String>>) {
|
|
if let Some(scopes) = scopes {
|
|
for scope in scopes.iter_mut() {
|
|
if scope.starts_with("run:") {
|
|
let (_, part_scope) = scope.split_once(":").unwrap();
|
|
|
|
if let Some((kind, path)) = part_scope.split_once("/") {
|
|
//appending a 's' as runnable kind is singular while new scope format expect it to be plural
|
|
*scope = format!("jobs:run:{}s:{}", kind, path);
|
|
}
|
|
} else if scope.starts_with("jobs:") {
|
|
// Map old jobs scopes to new format
|
|
let new_scope = match scope.as_str() {
|
|
"jobs:listjobs" => "jobs:read",
|
|
"jobs:runscript" => "jobs:run:scripts",
|
|
"jobs:runflow" => "jobs:run:flows",
|
|
"jobs:resumeflow" => "jobs:run:flows",
|
|
"jobs:deletejob" => "jobs:write",
|
|
_ => continue,
|
|
};
|
|
|
|
*scope = new_scope.to_string();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn maybe_get_workspace_id_from_path(path_vec: &[&str]) -> Option<String> {
|
|
let workspace_id = if path_vec.len() >= 4 && path_vec[0] == "" && path_vec[2] == "w" {
|
|
Some(path_vec[3].to_owned())
|
|
} else if path_vec.len() >= 5
|
|
&& path_vec[0] == ""
|
|
&& path_vec[1] == "api"
|
|
&& path_vec[2] == "mcp"
|
|
&& path_vec[3] == "w"
|
|
{
|
|
Some(path_vec[4].to_owned())
|
|
} else {
|
|
if path_vec.len() >= 5 && path_vec[0] == "" && path_vec[2] == "srch" && path_vec[3] == "w" {
|
|
Some(path_vec[4].to_owned())
|
|
} else {
|
|
None
|
|
}
|
|
};
|
|
|
|
workspace_id
|
|
}
|
|
|
|
/// `--no-auth` mode: compiled-in `oss` builds, or the `NO_AUTH` runtime flag on
|
|
/// any build (the runtime flag is force-disabled on CLOUD_HOSTED). When on,
|
|
/// every request resolves as the admin superadmin so a fronting gateway can
|
|
/// handle authentication instead.
|
|
pub fn is_no_auth() -> bool {
|
|
cfg!(feature = "no_auth") || *windmill_common::worker::NO_AUTH
|
|
}
|
|
|
|
/// The synthetic superadmin identity returned for every request in no-auth mode.
|
|
fn no_auth_admin_authed() -> ApiAuthed {
|
|
ApiAuthed {
|
|
email: "admin@windmill.dev".to_string(),
|
|
username: "admin".to_string(),
|
|
is_admin: true,
|
|
is_operator: false,
|
|
groups: Vec::new(),
|
|
folders: Vec::new(),
|
|
scopes: None,
|
|
username_override: None,
|
|
username_override_is_token_label: false,
|
|
is_session_token: false,
|
|
token_prefix: None,
|
|
read_only: false,
|
|
job_id: None,
|
|
credential_expiry: None,
|
|
}
|
|
}
|
|
|
|
/// Resolves OptJobAuthed from request parts.
|
|
/// Takes ownership of Parts and returns them back, memoized into the extensions so a
|
|
/// second extraction in the same request is free.
|
|
#[allow(unreachable_code, unused_mut)]
|
|
pub async fn resolve_opt_job_authed(
|
|
mut parts: Parts,
|
|
) -> std::result::Result<(OptJobAuthed, Parts), (Error, Parts)> {
|
|
if parts.method == http::Method::OPTIONS {
|
|
return Ok((OptJobAuthed::default(), parts));
|
|
};
|
|
|
|
if is_no_auth() {
|
|
return Ok((
|
|
OptJobAuthed { authed: no_auth_admin_authed(), job_id: None },
|
|
parts,
|
|
));
|
|
}
|
|
|
|
// Safe only while what the route checks below read is fixed before the first
|
|
// extraction: path and method always are, `workspace_id` only while every
|
|
// `GatewayWorkspaceId` writer stays layered outside auth extraction.
|
|
if let Some(opt_job_authed) = parts.extensions.get::<OptJobAuthed>().cloned() {
|
|
return Ok((opt_job_authed, parts));
|
|
}
|
|
|
|
let already_tokened = parts.extensions.get::<Tokened>().cloned();
|
|
let token_o = if let Some(token) = already_tokened {
|
|
Some(token.token.clone())
|
|
} else {
|
|
extract_token(&mut parts, &()).await
|
|
};
|
|
if let Some(token) = token_o {
|
|
if let Ok(Extension(cache)) =
|
|
Extension::<Arc<AuthCache>>::from_request_parts(&mut parts, &()).await
|
|
{
|
|
let original_uri = OriginalUri::from_request_parts(&mut parts, &())
|
|
.await
|
|
.ok()
|
|
.map(|x| x.0)
|
|
.unwrap_or_default();
|
|
let path_vec: Vec<&str> = original_uri.path().split("/").collect();
|
|
let workspace_id = maybe_get_workspace_id_from_path(&path_vec).or_else(|| {
|
|
parts
|
|
.extensions
|
|
.get::<windmill_common::db::GatewayWorkspaceId>()
|
|
.map(|g| g.0.clone())
|
|
});
|
|
|
|
if let Some(mut opt_job_authed) =
|
|
cache.get_opt_job_authed(workspace_id.clone(), &token).await
|
|
{
|
|
let path = original_uri.path();
|
|
let method = parts.method.as_str();
|
|
if workspace_id.is_none() && opt_job_authed.job_id.is_some() {
|
|
if let Err(err) = crate::scopes::check_job_token_for_global_route(path, method)
|
|
{
|
|
return Err((err, parts));
|
|
}
|
|
}
|
|
let authed = &mut opt_job_authed.authed;
|
|
if authed.scopes.is_some() {
|
|
transform_old_scope_to_new_scope(authed.scopes.as_mut());
|
|
|
|
if let Err(err) = crate::scopes::check_scopes_for_route(
|
|
authed.scopes.as_deref(),
|
|
path,
|
|
method,
|
|
) {
|
|
return Err((err, parts));
|
|
}
|
|
}
|
|
if authed.read_only {
|
|
// MCP transport runs over POST (streamable HTTP / SSE handshake),
|
|
// so the middleware can't safely reject mutating methods on it —
|
|
// the MCP runner itself filters out write tools and rejects
|
|
// mutating tool calls for read-only tokens. Narrow to the actual
|
|
// transport endpoints: anything else under `/api/mcp/*` (OAuth
|
|
// approve, token exchange, client registration) must still go
|
|
// through the read-only check, otherwise a read-only token
|
|
// could approve an OAuth flow that mints a new non-read-only
|
|
// token.
|
|
let is_mcp_transport = path == "/api/mcp/gateway"
|
|
|| (path.starts_with("/api/mcp/w/")
|
|
&& (path.ends_with("/mcp")
|
|
|| path.ends_with("/sse")
|
|
|| path.ends_with("/list_tools")));
|
|
if !is_mcp_transport {
|
|
if let Err(err) = crate::scopes::check_read_only_for_route(path, method) {
|
|
return Err((err, parts));
|
|
}
|
|
}
|
|
}
|
|
parts.extensions.insert(authed.clone());
|
|
|
|
Span::current().record("username", &authed.username.as_str());
|
|
Span::current().record("email", &authed.email);
|
|
|
|
// Mirror into the per-request LogContext so exported OTEL
|
|
// LogRecords carry the same identifiers (the log bridge
|
|
// doesn't walk span fields — see windmill_common::log_context).
|
|
let username_copy = authed.username.clone();
|
|
let email_copy = authed.email.clone();
|
|
let workspace_copy = workspace_id.clone();
|
|
windmill_common::log_context::update_log_context(move |c| {
|
|
windmill_common::log_context::LogContext {
|
|
username: Some(username_copy),
|
|
email: Some(email_copy),
|
|
workspace_id: workspace_copy.or_else(|| c.workspace_id.clone()),
|
|
..c.clone()
|
|
}
|
|
});
|
|
|
|
if let Some(workspace_id) = workspace_id {
|
|
Span::current().record("workspace_id", &workspace_id);
|
|
}
|
|
// Separate from the ApiAuthed insert above, which handlers taking
|
|
// `Extension<ApiAuthed>` read.
|
|
parts.extensions.insert(opt_job_authed.clone());
|
|
return Ok((opt_job_authed, parts));
|
|
}
|
|
}
|
|
}
|
|
BRUTE_FORCE_COUNTER.increment().await;
|
|
Err((Error::NotAuthorized("Unauthorized".to_string()), parts))
|
|
}
|
|
|
|
/// Returns the override and whether it names the token's *label* rather than the entity that
|
|
/// fired the request. Callers must not re-derive the second element from the first: the
|
|
/// `ephemeral-script-end-user-` arm forwards a `created_by` verbatim, and `created_by` is
|
|
/// unconstrained, so it may itself look like any of these shapes.
|
|
///
|
|
/// Only namespaces `create_token` rejects (`is_server_minted_label`) are trusted to name the
|
|
/// entity acting, so the label can only have come from a server-side mint. Tokens minted
|
|
/// before that guard existed are the remaining hole; closing it needs the token row to record
|
|
/// who minted it rather than inferring it from the label.
|
|
///
|
|
/// Note that a trigger whose identity is set server-side — the SMTP one builds an `email-*`
|
|
/// override directly — does not rely on this at all, so its prefix must not be trusted here.
|
|
pub(crate) fn username_override_from_label(label: Option<String>) -> (Option<String>, bool) {
|
|
match label {
|
|
Some(label) if label.starts_with("ephemeral-webhook-") => (Some(label), false),
|
|
Some(label) if label.starts_with("ephemeral-script-end-user-") => (
|
|
Some(
|
|
label
|
|
.trim_start_matches("ephemeral-script-end-user-")
|
|
.to_string(),
|
|
),
|
|
false,
|
|
),
|
|
// User-mintable, so they name nobody in particular — the trigger panels merely
|
|
// pre-fill `webhook-`/`http-`, and the editor mints the lsp one. The override keeps
|
|
// its value because `require_job_read_access` matches it against the `created_by` of
|
|
// jobs launched under it, which these shapes produced while they were trusted.
|
|
Some(label) if label == "Ephemeral lsp token" => (Some("lsp".to_string()), true),
|
|
Some(label)
|
|
if label.starts_with("webhook-")
|
|
|| label.starts_with("http-")
|
|
|| label.starts_with("email-")
|
|
|| label.starts_with("ws-") =>
|
|
{
|
|
(Some(label), true)
|
|
}
|
|
Some(label)
|
|
if label != "ephemeral-script"
|
|
&& label != "session"
|
|
&& label != windmill_common::auth::GUEST_SESSION_LABEL
|
|
&& !label.is_empty() =>
|
|
{
|
|
(
|
|
Some(format!("{}{label}", crate::GENERIC_TOKEN_LABEL_PREFIX)),
|
|
true,
|
|
)
|
|
}
|
|
_ => (None, false),
|
|
}
|
|
}
|
|
|
|
#[derive(FromRow, Serialize)]
|
|
pub struct TruncatedTokenWithEmail {
|
|
pub label: Option<String>,
|
|
pub token_prefix: String,
|
|
pub expiration: Option<chrono::DateTime<chrono::Utc>>,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
pub last_used_at: chrono::DateTime<chrono::Utc>,
|
|
pub scopes: Option<Vec<String>>,
|
|
pub email: Option<String>,
|
|
}
|
|
|
|
pub async fn list_tokens_internal(
|
|
db: &DB,
|
|
w_id: &str,
|
|
path: &str,
|
|
is_flow: bool,
|
|
) -> JsonResult<Vec<TruncatedTokenWithEmail>> {
|
|
let tokens = if is_flow {
|
|
sqlx::query_as!(
|
|
TruncatedTokenWithEmail,
|
|
r#"
|
|
SELECT label,
|
|
token_prefix,
|
|
expiration,
|
|
created_at,
|
|
last_used_at,
|
|
scopes,
|
|
email
|
|
FROM token
|
|
WHERE workspace_id = $1
|
|
AND (
|
|
scopes @> ARRAY['jobs:run:flows:' || $2]::text[]
|
|
OR scopes @> ARRAY['run:flow/' || $2]::text[]
|
|
)
|
|
"#,
|
|
w_id,
|
|
path
|
|
)
|
|
.fetch_all(db)
|
|
.await?
|
|
} else {
|
|
sqlx::query_as!(
|
|
TruncatedTokenWithEmail,
|
|
r#"
|
|
SELECT label,
|
|
token_prefix,
|
|
expiration,
|
|
created_at,
|
|
last_used_at,
|
|
scopes,
|
|
email
|
|
FROM token
|
|
WHERE workspace_id = $1
|
|
AND (
|
|
scopes @> ARRAY['jobs:run:scripts:' || $2]::text[]
|
|
OR scopes @> ARRAY['run:script/' || $2]::text[]
|
|
)
|
|
"#,
|
|
w_id,
|
|
path
|
|
)
|
|
.fetch_all(db)
|
|
.await?
|
|
};
|
|
|
|
Ok(Json(tokens))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// Dropping the `AUTH_CACHE` entry between the two resolutions is what makes a memo
|
|
/// hit observable: only a working memo can answer the second one.
|
|
#[tokio::test]
|
|
async fn resolve_opt_job_authed_is_memoized() {
|
|
// Never connected: an AUTH_CACHE hit resolves without a query, and AuthCache
|
|
// only needs the handle to exist. Without the timeout a regression — which
|
|
// falls through to a real lookup — spends sqlx's 30s default failing.
|
|
let db = sqlx::postgres::PgPoolOptions::new()
|
|
.acquire_timeout(std::time::Duration::from_millis(50))
|
|
.connect_lazy("postgres://memo@127.0.0.1:1/memo")
|
|
.expect("lazy pool");
|
|
|
|
let token = "memo_guard_token";
|
|
// Workspace-scoped: a job token on a workspace-less route is refused by
|
|
// check_job_token_for_global_route before the memo is ever taken.
|
|
let key = ("memo_workspace".to_string(), token.to_string());
|
|
let job_id = uuid::Uuid::from_u128(0x6d656d6f);
|
|
AUTH_CACHE.insert(
|
|
key.clone(),
|
|
ExpiringAuthCache {
|
|
authed: ApiAuthed {
|
|
email: "memo@windmill.dev".to_string(),
|
|
username: "memo".to_string(),
|
|
..Default::default()
|
|
},
|
|
expiry: chrono::Utc::now() + chrono::Duration::hours(1),
|
|
job_id: Some(job_id),
|
|
},
|
|
);
|
|
|
|
#[cfg(feature = "enterprise")]
|
|
let cache = AuthCache::new(db, None, None);
|
|
#[cfg(not(feature = "enterprise"))]
|
|
let cache = AuthCache::new(db, None);
|
|
|
|
let mut parts = http::Request::builder()
|
|
.uri("/api/w/memo_workspace/jobs/list")
|
|
.header(http::header::AUTHORIZATION, format!("Bearer {token}"))
|
|
.body(())
|
|
.unwrap()
|
|
.into_parts()
|
|
.0;
|
|
parts.extensions.insert(Arc::new(cache));
|
|
|
|
let (_, parts) = resolve_opt_job_authed(parts)
|
|
.await
|
|
.map_err(|(e, _)| e)
|
|
.expect("first resolution");
|
|
|
|
AUTH_CACHE.remove(&key);
|
|
|
|
let (second, _) = resolve_opt_job_authed(parts)
|
|
.await
|
|
.map_err(|(e, _)| e)
|
|
.expect("second resolution must hit the memo, not re-resolve");
|
|
assert_eq!(second.authed.username, "memo");
|
|
// The wrapper must survive whole: memoizing only the inner ApiAuthed and
|
|
// rebuilding around it drops the provenance forbid_elevated_job_token gates on.
|
|
assert_eq!(second.job_id, Some(job_id));
|
|
}
|
|
}
|