mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-20 00:02:19 +00:00
ee3d82f01f
* fix(native-triggers): serialize Google channel renewal across replicas `sync_all_triggers` runs every 5 minutes on every windmill-app replica with no leader election. Multiple replicas were each rotating the webhook token, creating a new Google watch channel, and racing the trigger UPDATE — leaving the loser's new token (in `token`) and channel (in Google) orphaned. Cloud was accumulating ~5 leaked tokens/week without the silent best-effort `delete_token_by_hash` ever logging a warning. Wrap each per-trigger renewal in a transaction and acquire the row with `SELECT … FOR UPDATE SKIP LOCKED`. Contending replicas skip the row instead of duplicating the work. The lock spans `rotate_webhook_token` → Google API call → `update_native_trigger_service_config` and is only released on commit. Re-checks `should_renew_channel` after acquiring the lock so a replica that committed seconds earlier doesn't trigger a duplicate renewal. The pattern matches existing batch-cleanup paths in `monitor.rs` (job-retention sweep) and other `FOR UPDATE SKIP LOCKED` call sites. Also logs at `debug!` when `delete_token_by_hash` finds no matching row, so future investigations can distinguish "deleted" from "not found" without changing the `Ok(false)` contract. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fixup! fix(native-triggers): serialize Google channel renewal across replicas * fixup! fix(native-triggers): serialize Google channel renewal across replicas fixup! fix(native-triggers): serialize Google channel renewal across replicas Address claude review: - #5: per-skip log info -> debug (expected outcome under SKIP LOCKED) - #2: warn moved out of delete_token_by_hash to the call site that knows the expected state (try_renew_channel_locked); other callers are race-prone and shouldn't warn - #3: NULL service_config now warns (anomalous case) - #4: post-Google-API DB-update + commit failures log distinctly so the channel-orphan case is grep-able Plus: add 14d expiry to Google webhook tokens via ServiceName::webhook_token_expiration, mint fresh ephemeral-webhook-{service}-{rd5} labels at create + rotate so the existing 'ephemeral-' filter excludes them from user-token email/critical-alert paths (no filter changes in 3 places). Orphans now self-clean via the existing expiry sweep in monitor.rs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fixup! fix(native-triggers): serialize Google channel renewal across replicas fixup! fix(native-triggers): serialize Google channel renewal across replicas Address second-round review: - Claude #1 (P2): username_override_from_label now strips the 'ephemeral-' prefix for ephemeral-webhook-* labels, so created_by stays webhook-{service}-{rd5} instead of changing to label-ephemeral-webhook-... (preserves audit/job-list filter compatibility) - Codex (P2): updated renew_channel doc — labels are no longer copied; rotate mints fresh ephemeral-webhook-google-{rd5} with 14d expiration - Claude #3 (optional): test_rotate_webhook_token now asserts the rotated Google token has an ephemeral-webhook-google-* label and a populated expiration Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fixup! fix(native-triggers): serialize Google channel renewal across replicas fixup! fix(native-triggers): serialize Google channel renewal across replicas Reconsider the previous fixup: stripping the 'ephemeral-' prefix made created_by no longer match token.label exactly, defeating the linking purpose. Just allowlist 'ephemeral-webhook-' alongside the other recognized webhook/email/ws prefixes — created_by becomes ephemeral-webhook-google-XXXXX, matching token.label exactly. The 'ephemeral-' substring also informs operators that this is a system-managed auto-expiring token vs a user-managed webhook trigger. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
413 lines
14 KiB
Rust
413 lines
14 KiB
Rust
//! Tests for the token hash migration.
|
|
//!
|
|
//! Verifies that:
|
|
//! - Rust hash_token() matches PostgreSQL's encode(sha256(...),'hex')
|
|
//! - Newly created tokens can authenticate immediately
|
|
//! - Token list/delete-by-prefix works with the new token_prefix column
|
|
//! - Logout invalidates tokens via hash-based deletion
|
|
//! - Backward compat: plaintext column is populated when old workers exist
|
|
//! - rotate_webhook_token produces valid tokens and defers old token deletion
|
|
|
|
use serde_json::json;
|
|
use sqlx::{Pool, Postgres};
|
|
use windmill_common::auth::{hash_token, TOKEN_PREFIX_LEN};
|
|
use windmill_test_utils::*;
|
|
|
|
fn client() -> reqwest::Client {
|
|
reqwest::Client::new()
|
|
}
|
|
|
|
fn authed(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
|
builder.header("Authorization", "Bearer SECRET_TOKEN")
|
|
}
|
|
|
|
fn authed_with(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder {
|
|
builder.header("Authorization", format!("Bearer {}", token))
|
|
}
|
|
|
|
/// Test 1: Verify that Rust's hash_token() produces the same hash as PostgreSQL's
|
|
/// encode(sha256(token::bytea), 'hex'). This is the foundational invariant.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_hash_consistency(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
|
|
// Compute hash in Rust
|
|
let rust_hash = hash_token("SECRET_TOKEN");
|
|
|
|
// Compute hash in PostgreSQL
|
|
let pg_hash: String =
|
|
sqlx::query_scalar!("SELECT encode(sha256('SECRET_TOKEN'::bytea), 'hex') AS hash")
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap();
|
|
|
|
assert_eq!(
|
|
rust_hash, pg_hash,
|
|
"Rust hash_token() must match PostgreSQL sha256()"
|
|
);
|
|
|
|
// Also verify it matches what's stored in the fixture
|
|
let stored_hash: String = sqlx::query_scalar!(
|
|
"SELECT token_hash FROM token WHERE email = 'test@windmill.dev' AND label = 'test token'"
|
|
)
|
|
.fetch_one(&db)
|
|
.await?;
|
|
|
|
assert_eq!(
|
|
rust_hash, stored_hash,
|
|
"hash_token() must match the fixture's pre-computed hash"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 2: Create a token via API, then immediately use it to authenticate.
|
|
/// Verifies create_token_internal stores the hash correctly and auth lookups work.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_create_token_and_auth(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
let server = ApiServer::start(db.clone()).await?;
|
|
let port = server.addr.port();
|
|
let base = format!("http://localhost:{port}/api/users");
|
|
|
|
// Create a new token
|
|
let resp = authed(client().post(format!("{base}/tokens/create")))
|
|
.json(&json!({"label": "test-hash-token"}))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 201);
|
|
let new_token = resp.text().await?;
|
|
assert!(!new_token.is_empty());
|
|
|
|
// Use the new token to call whoami
|
|
let resp = authed_with(client().get(format!("{base}/whoami")), &new_token)
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 200, "newly created token must authenticate");
|
|
let body = resp.json::<serde_json::Value>().await?;
|
|
assert_eq!(body["email"], "test@windmill.dev");
|
|
|
|
// Verify the hash is stored correctly in DB
|
|
let expected_hash = hash_token(&new_token);
|
|
let db_hash: Option<String> = sqlx::query_scalar!(
|
|
"SELECT token_hash FROM token WHERE token_hash = $1",
|
|
expected_hash
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?;
|
|
assert!(db_hash.is_some(), "token_hash must be stored in DB");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 3: Create a token, list tokens (verify prefix), delete by prefix, confirm invalid.
|
|
/// Covers the change from WHERE token LIKE to WHERE token_prefix = $1.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_token_list_and_delete_by_prefix(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
let server = ApiServer::start(db.clone()).await?;
|
|
let port = server.addr.port();
|
|
let base = format!("http://localhost:{port}/api/users");
|
|
|
|
// Create a token
|
|
let resp = authed(client().post(format!("{base}/tokens/create")))
|
|
.json(&json!({"label": "prefix-test-token"}))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 201);
|
|
let new_token = resp.text().await?;
|
|
let prefix = &new_token[..TOKEN_PREFIX_LEN];
|
|
|
|
// List tokens and find our token by prefix
|
|
let resp = authed(client().get(format!("{base}/tokens/list")))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 200);
|
|
let tokens = resp.json::<Vec<serde_json::Value>>().await?;
|
|
let found = tokens
|
|
.iter()
|
|
.any(|t| t["token_prefix"].as_str() == Some(prefix));
|
|
assert!(found, "token with prefix {prefix} must appear in list");
|
|
|
|
// Verify the new token works
|
|
let resp = authed_with(client().get(format!("{base}/whoami")), &new_token)
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 200);
|
|
|
|
// Delete by prefix
|
|
let resp = authed(client().delete(format!("{base}/tokens/delete/{prefix}")))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 200, "delete token: {}", resp.text().await?);
|
|
|
|
// Confirm the token is gone from the DB (auth cache may still serve 200 briefly)
|
|
let token_hash = hash_token(&new_token);
|
|
let deleted: bool = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM token WHERE token_hash = $1) AS exists",
|
|
token_hash
|
|
)
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap_or(true);
|
|
assert!(
|
|
!deleted,
|
|
"token must be deleted from DB after delete-by-prefix"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 4: Logout invalidates a token via hash-based deletion.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_logout_invalidates_token(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
let server = ApiServer::start(db.clone()).await?;
|
|
let port = server.addr.port();
|
|
let base = format!("http://localhost:{port}/api/users");
|
|
let auth_base = format!("http://localhost:{port}/api/auth");
|
|
|
|
// Create a fresh token (don't burn the fixture token)
|
|
let resp = authed(client().post(format!("{base}/tokens/create")))
|
|
.json(&json!({"label": "logout-test-token"}))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 201);
|
|
let token = resp.text().await?;
|
|
|
|
// Verify it works
|
|
let resp = authed_with(client().get(format!("{base}/whoami")), &token)
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 200);
|
|
|
|
// Logout with the token
|
|
let resp = authed_with(client().post(format!("{auth_base}/logout")), &token)
|
|
.send()
|
|
.await?;
|
|
assert!(
|
|
resp.status() == 200 || resp.status() == 303,
|
|
"logout: unexpected status {}",
|
|
resp.status()
|
|
);
|
|
|
|
// Confirm the token is gone from the DB (auth cache may still serve 200 briefly)
|
|
let token_hash = hash_token(&token);
|
|
let exists: bool = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM token WHERE token_hash = $1) AS exists",
|
|
token_hash
|
|
)
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap_or(true);
|
|
assert!(!exists, "token must be deleted from DB after logout");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 5: Backward compatibility — plaintext column behavior based on MIN_VERSION.
|
|
/// When old workers exist (MIN_VERSION < 1.650.0), plaintext must be written so
|
|
/// old workers running WHERE token = $1 can still authenticate.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_plaintext_backward_compat(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
|
|
use windmill_common::min_version::{MIN_VERSION, MIN_VERSION_SUPPORTS_TOKEN_HASH};
|
|
|
|
let server = ApiServer::start(db.clone()).await?;
|
|
let port = server.addr.port();
|
|
let base = format!("http://localhost:{port}/api/users");
|
|
|
|
// --- Phase 1: Simulate old workers present (version < 1.650.0) ---
|
|
// Set MIN_VERSION to one minor below the token hash feature version
|
|
let mut old_version = MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone();
|
|
old_version.minor -= 1;
|
|
MIN_VERSION.store(std::sync::Arc::new(old_version));
|
|
|
|
let resp = authed(client().post(format!("{base}/tokens/create")))
|
|
.json(&json!({"label": "old-worker-compat-token"}))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 201);
|
|
let old_compat_token = resp.text().await?;
|
|
let old_compat_hash = hash_token(&old_compat_token);
|
|
|
|
// Plaintext should be stored (for old workers)
|
|
let plaintext: Option<String> = sqlx::query_scalar!(
|
|
"SELECT token FROM token WHERE token_hash = $1",
|
|
old_compat_hash
|
|
)
|
|
.fetch_one(&db)
|
|
.await?;
|
|
assert!(
|
|
plaintext.is_some(),
|
|
"plaintext must be stored when old workers exist"
|
|
);
|
|
assert_eq!(plaintext.unwrap(), old_compat_token);
|
|
|
|
// Old-style query (what old workers run) must find the token
|
|
let old_style_email: Option<String> = sqlx::query_scalar!(
|
|
"SELECT email FROM token WHERE token = $1 AND (expiration > NOW() OR expiration IS NULL)",
|
|
&old_compat_token
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.flatten();
|
|
assert_eq!(
|
|
old_style_email.as_deref(),
|
|
Some("test@windmill.dev"),
|
|
"old-style WHERE token = $1 must find the token"
|
|
);
|
|
|
|
// New-style query must also work
|
|
let new_style_email: Option<String> = sqlx::query_scalar!(
|
|
"SELECT email FROM token WHERE token_hash = $1 AND (expiration > NOW() OR expiration IS NULL)",
|
|
old_compat_hash
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.flatten();
|
|
assert_eq!(
|
|
new_style_email.as_deref(),
|
|
Some("test@windmill.dev"),
|
|
"new-style WHERE token_hash = $1 must also work"
|
|
);
|
|
|
|
// --- Phase 2: All workers upgraded (version >= 1.650.0) ---
|
|
MIN_VERSION.store(std::sync::Arc::new(
|
|
MIN_VERSION_SUPPORTS_TOKEN_HASH.version().clone(),
|
|
));
|
|
|
|
let resp = authed(client().post(format!("{base}/tokens/create")))
|
|
.json(&json!({"label": "new-worker-token"}))
|
|
.send()
|
|
.await?;
|
|
assert_eq!(resp.status(), 201);
|
|
let new_token = resp.text().await?;
|
|
let new_hash = hash_token(&new_token);
|
|
|
|
// Plaintext should NOT be stored
|
|
let plaintext: Option<String> =
|
|
sqlx::query_scalar!("SELECT token FROM token WHERE token_hash = $1", new_hash)
|
|
.fetch_one(&db)
|
|
.await?;
|
|
assert!(
|
|
plaintext.is_none(),
|
|
"plaintext must be NULL when all workers support hash"
|
|
);
|
|
|
|
// Old-style query should NOT find this token
|
|
let old_style_result: Option<String> =
|
|
sqlx::query_scalar!("SELECT email FROM token WHERE token = $1", &new_token)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.flatten();
|
|
assert!(
|
|
old_style_result.is_none(),
|
|
"old-style query must not find token when plaintext is NULL"
|
|
);
|
|
|
|
// New-style query must still work
|
|
let resp = authed_with(client().get(format!("{base}/whoami")), &new_token)
|
|
.send()
|
|
.await?;
|
|
assert_eq!(
|
|
resp.status(),
|
|
200,
|
|
"new token must authenticate via hash lookup"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Test 6: rotate_webhook_token creates a new token and keeps the old one alive.
|
|
/// Callers delete the old token after successfully updating the trigger.
|
|
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
|
async fn test_rotate_webhook_token(db: Pool<Postgres>) -> anyhow::Result<()> {
|
|
initialize_tracing().await;
|
|
|
|
use windmill_native_triggers::{delete_token_by_hash, rotate_webhook_token, ServiceName};
|
|
|
|
// Insert a token directly with known values
|
|
let original_token = "test-webhook-token-original-1234";
|
|
let original_hash = hash_token(original_token);
|
|
let original_prefix = &original_token[..TOKEN_PREFIX_LEN];
|
|
|
|
sqlx::query!(
|
|
"INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin)
|
|
VALUES ($1, $2, $3, 'test@windmill.dev', 'webhook-test', false)",
|
|
original_hash,
|
|
original_prefix,
|
|
original_token,
|
|
)
|
|
.execute(&db)
|
|
.await?;
|
|
|
|
// Rotate the token
|
|
let rotated = rotate_webhook_token(&db, &original_hash, ServiceName::Google)
|
|
.await?
|
|
.expect("rotate must return Some for existing token");
|
|
|
|
// New token should be different
|
|
assert_ne!(rotated.new_token, original_token);
|
|
assert_eq!(rotated.old_token_hash, original_hash);
|
|
|
|
// New token's hash should exist in DB with the per-service label and expiration
|
|
let new_hash = hash_token(&rotated.new_token);
|
|
let new_row = sqlx::query!(
|
|
"SELECT label, expiration FROM token WHERE token_hash = $1",
|
|
new_hash
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.expect("new token hash must exist in DB after rotation");
|
|
assert!(
|
|
new_row
|
|
.label
|
|
.as_deref()
|
|
.is_some_and(|l| l.starts_with("ephemeral-webhook-google-")),
|
|
"rotated token must carry an ephemeral-webhook-google-* label, got {:?}",
|
|
new_row.label
|
|
);
|
|
assert!(
|
|
new_row.expiration.is_some(),
|
|
"rotated Google token must carry an expiration"
|
|
);
|
|
|
|
// Old token should still exist (deletion deferred to caller)
|
|
let old_exists: bool = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM token WHERE token_hash = $1) AS exists",
|
|
original_hash
|
|
)
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap_or(false);
|
|
assert!(
|
|
old_exists,
|
|
"old token must still exist until caller deletes it"
|
|
);
|
|
|
|
// Caller deletes old token after successful trigger update
|
|
let deleted = delete_token_by_hash(&db, &rotated.old_token_hash).await?;
|
|
assert!(deleted, "old token must be deletable");
|
|
|
|
// Old token should now be gone
|
|
let old_gone: bool = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM token WHERE token_hash = $1) AS exists",
|
|
original_hash
|
|
)
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap_or(true);
|
|
assert!(!old_gone, "old token must be gone after explicit deletion");
|
|
|
|
// Rotating a non-existent hash should return None
|
|
let result = rotate_webhook_token(&db, "nonexistent_hash", ServiceName::Google).await?;
|
|
assert!(
|
|
result.is_none(),
|
|
"rotating a non-existent token must return None"
|
|
);
|
|
|
|
Ok(())
|
|
}
|