starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-06 08:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: store hashed tokens in the token table instead of plaintext

Browse Source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-03-04 07:47:54 +00:00
co-authored by Claude Opus 4.6
parent ef7b2ec81c
commit c65b4b2bb4
42 changed files with 686 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-8be2919c3511575c89b882b112b987fd5724c299cb285f819a2561260404e513.json → backend/.sqlx/query-1bf4a93cb85c6eed313a2f393da9408dd2aa4e47ef7a38a0d3ccca944a09f5bb.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, last_used_at, scopes FROM token WHERE email = $1 AND (label != 'ephemeral-script' OR label IS NULL)\n ORDER BY created_at DESC LIMIT $2 OFFSET $3",
"query": "SELECT label, token_prefix, expiration, created_at, last_used_at, scopes FROM token WHERE email = $1 AND (label != 'ephemeral-script' OR label IS NULL)\n ORDER BY created_at DESC LIMIT $2 OFFSET $3",
"describe": {
"columns": [
{
@@ -11,7 +11,7 @@
{
"ordinal": 1,
"name": "token_prefix",
"type_info": "Text"
"type_info": "Varchar"
},
{
"ordinal": 2,
@@ -43,12 +43,12 @@
},
"nullable": [
true,
null,
false,
true,
false,
false,
true
]
},
"hash": "8be2919c3511575c89b882b112b987fd5724c299cb285f819a2561260404e513"
"hash": "1bf4a93cb85c6eed313a2f393da9408dd2aa4e47ef7a38a0d3ccca944a09f5bb"
}
backend/.sqlx/query-1eba3d46110917b248da0b00fc63d63ad273994fa8508c8aadaefdb7aa7e5185.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM token\n WHERE token_prefix = $1\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "1eba3d46110917b248da0b00fc63d63ad273994fa8508c8aadaefdb7aa7e5185"
}
backend/.sqlx/query-6a254de9005594dc75a59a545546417c8a5aa7635be1dc0b37dc29d0f9e7c163.json → backend/.sqlx/query-207106aa8267fe756989f3ee1eadb7e169d07463f67f1da79c8bc23c1079c185.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT label,\n concat(substring(token for 10)) AS token_prefix,\n expiration,\n created_at,\n last_used_at,\n scopes,\n email\n FROM token\n WHERE workspace_id = $1\n AND (\n scopes @> ARRAY['jobs:run:flows:' || $2]::text[]\n OR scopes @> ARRAY['run:flow/' || $2]::text[]\n )\n ",
"query": "\n SELECT label,\n token_prefix,\n expiration,\n created_at,\n last_used_at,\n scopes,\n email\n FROM token\n WHERE workspace_id = $1\n AND (\n scopes @> ARRAY['jobs:run:scripts:' || $2]::text[]\n OR scopes @> ARRAY['run:script/' || $2]::text[]\n )\n ",
"describe": {
"columns": [
{
@@ -11,7 +11,7 @@
{
"ordinal": 1,
"name": "token_prefix",
"type_info": "Text"
"type_info": "Varchar"
},
{
"ordinal": 2,
@@ -47,7 +47,7 @@
},
"nullable": [
true,
null,
false,
true,
false,
false,
@@ -55,5 +55,5 @@
true
]
},
"hash": "6a254de9005594dc75a59a545546417c8a5aa7635be1dc0b37dc29d0f9e7c163"
"hash": "207106aa8267fe756989f3ee1eadb7e169d07463f67f1da79c8bc23c1079c185"
}
backend/.sqlx/query-1bdf186d3b99bbd913cbf95150105470cd5f1d4ddbb147cb8ce46f9d1da5dfaf.json → backend/.sqlx/query-215163b5a2791c51f9b28681c1ca1a47475dcf1a388c613a9e0154aef6582a23.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH email_lookup AS (\n SELECT email FROM token WHERE token = $1\n )\n DELETE FROM token\n WHERE email = (SELECT email FROM email_lookup) AND label = 'session'\n RETURNING email",
"query": "WITH email_lookup AS (\n SELECT email FROM token WHERE token_hash = $1\n )\n DELETE FROM token\n WHERE email = (SELECT email FROM email_lookup) AND label = 'session'\n RETURNING email",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "1bdf186d3b99bbd913cbf95150105470cd5f1d4ddbb147cb8ce46f9d1da5dfaf"
"hash": "215163b5a2791c51f9b28681c1ca1a47475dcf1a388c613a9e0154aef6582a23"
}
backend/.sqlx/query-66e0968fe9f757755945a7010153821cf73ace9d6692750ccc4cca37701ed77a.json → backend/.sqlx/query-4b76c4a387786bc5bb69e4c684c34b936c3ffff44ae58f0709d05ba3ff534f79.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM token WHERE token = $1",
"query": "DELETE FROM token WHERE token_hash = $1",
"describe": {
"columns": [],
"parameters": {
@@ -10,5 +10,5 @@
},
"nullable": []
},
"hash": "66e0968fe9f757755945a7010153821cf73ace9d6692750ccc4cca37701ed77a"
"hash": "4b76c4a387786bc5bb69e4c684c34b936c3ffff44ae58f0709d05ba3ff534f79"
}
backend/.sqlx/query-29673d489fbf45fc249da04c1a2fd60e2364ba87263f962ed7d4329c916620a1.json → backend/.sqlx/query-4c7231f24fd0bcc99004c5bd4065697cd321b397422a7c689b85216bbb1fd525.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT label,\n concat(substring(token for 10)) AS token_prefix,\n expiration,\n created_at,\n last_used_at,\n scopes,\n email\n FROM token\n WHERE workspace_id = $1\n AND (\n scopes @> ARRAY['jobs:run:scripts:' || $2]::text[]\n OR scopes @> ARRAY['run:script/' || $2]::text[]\n )\n ",
"query": "\n SELECT label,\n token_prefix,\n expiration,\n created_at,\n last_used_at,\n scopes,\n email\n FROM token\n WHERE workspace_id = $1\n AND (\n scopes @> ARRAY['jobs:run:flows:' || $2]::text[]\n OR scopes @> ARRAY['run:flow/' || $2]::text[]\n )\n ",
"describe": {
"columns": [
{
@@ -11,7 +11,7 @@
{
"ordinal": 1,
"name": "token_prefix",
"type_info": "Text"
"type_info": "Varchar"
},
{
"ordinal": 2,
@@ -47,7 +47,7 @@
},
"nullable": [
true,
null,
false,
true,
false,
false,
@@ -55,5 +55,5 @@
true
]
},
"hash": "29673d489fbf45fc249da04c1a2fd60e2364ba87263f962ed7d4329c916620a1"
"hash": "4c7231f24fd0bcc99004c5bd4065697cd321b397422a7c689b85216bbb1fd525"
}
backend/.sqlx/query-4e88aec662ebc70e0425a48a1b4e2e60e3183fa81a411622891caea6dc03fa90.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token (token_hash, token_prefix, email, label, super_admin, owner, workspace_id)\n VALUES ($1, $2, 'charlie@windmill.dev', 'Charlie new token', false, 'u/charlie', 'test-workspace')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "4e88aec662ebc70e0425a48a1b4e2e60e3183fa81a411622891caea6dc03fa90"
}
backend/.sqlx/query-2d6607b3c38fe72b5663c32de58dacbabed4c5ae28101e3ae2694f96fd055a91.json → backend/.sqlx/query-52379713a1f7312127bcd13c9a8027a85270c25c5a0f0d4d7670bd602bd3cebf.json
Generated
+3 -3
View File
@@ -1,11 +1,11 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token",
"query": "DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token_prefix",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"name": "token_prefix",
"type_info": "Varchar"
}
],
@@ -18,5 +18,5 @@
false
]
},
"hash": "2d6607b3c38fe72b5663c32de58dacbabed4c5ae28101e3ae2694f96fd055a91"
"hash": "52379713a1f7312127bcd13c9a8027a85270c25c5a0f0d4d7670bd602bd3cebf"
}
backend/.sqlx/query-54756c6c39888feb2206b056df1c84c3bb44adc490309954359845c06b6e607c.json
Generated
-18
View File
@@ -1,18 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token, email, label, expiration, super_admin)\n VALUES ($1, $2, $3, $4, $5)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Timestamptz",
"Bool"
]
},
"nullable": []
},
"hash": "54756c6c39888feb2206b056df1c84c3bb44adc490309954359845c06b6e607c"
}
backend/.sqlx/query-58dc872520beaa914fef8b7f30e578261fb9ebd92a81e1f2c8edaf93cece0819.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n DELETE FROM token\n WHERE token LIKE concat($1::text, '%')\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "58dc872520beaa914fef8b7f30e578261fb9ebd92a81e1f2c8edaf93cece0819"
}
backend/.sqlx/query-5926c452cf9af4e915f45e526667d10306436222bc5b6cb0403507bd465254d8.json
Generated
+21
View File
@@ -0,0 +1,21 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token_hash, token_prefix, email, label, expiration, super_admin, scopes, workspace_id)\n SELECT $1, $2, $3, $4, $5, $6, $7, $8\n WHERE $8::varchar IS NULL OR NOT EXISTS(\n SELECT 1 FROM workspace WHERE id = $8 AND deleted = true\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Timestamptz",
"Bool",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "5926c452cf9af4e915f45e526667d10306436222bc5b6cb0403507bd465254d8"
}
backend/.sqlx/query-5efdf09fc38aed0dea162979444104e3a9bf44ff175993bda6eb1848d155b39c.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token (token_hash, token_prefix, email, label, expiration, scopes, workspace_id)\n SELECT $1::varchar, $2::varchar, $3::varchar, $4::varchar, now() + ($5 || ' seconds')::interval, $6::text[], $7::varchar\n WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $7 AND deleted = true)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Text",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "5efdf09fc38aed0dea162979444104e3a9bf44ff175993bda6eb1848d155b39c"
}
backend/.sqlx/query-97e364c703bdcdfb5345810659cbe0477a28b8199ef0b297f9a22c88a43b6b5c.json → backend/.sqlx/query-722f9da2b3ad1e1129928c52498b994db0dc1728945f90fd23b707ee355d0472.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM token WHERE token = $1 RETURNING email",
"query": "DELETE FROM token WHERE token_hash = $1 RETURNING email",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "97e364c703bdcdfb5345810659cbe0477a28b8199ef0b297f9a22c88a43b6b5c"
"hash": "722f9da2b3ad1e1129928c52498b994db0dc1728945f90fd23b707ee355d0472"
}
backend/.sqlx/query-bfff3d8df18db198d6ebba8a049b00147fc8bcd42f3df37ef81b9ded80974bd0.json → backend/.sqlx/query-8065ed67770101e30eea456c1c682e1900d97721d931cea80a4fe240901b3604.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT true FROM token WHERE token = $1 and expiration IS NOT NULL and expiration > now() + $2::int * '1 sec'::interval",
"query": "SELECT true FROM token WHERE token_hash = $1 and expiration IS NOT NULL and expiration > now() + $2::int * '1 sec'::interval",
"describe": {
"columns": [
{
@@ -19,5 +19,5 @@
null
]
},
"hash": "bfff3d8df18db198d6ebba8a049b00147fc8bcd42f3df37ef81b9ded80974bd0"
"hash": "8065ed67770101e30eea456c1c682e1900d97721d931cea80a4fe240901b3604"
}
backend/.sqlx/query-8515537f73c132e62c4dafad1e8d8e56f0c70dccd0edbc9667f550b31ab54c18.json
Generated
-12
View File
@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token (token, email, label, super_admin, owner, workspace_id)\n VALUES ('CHARLIE_TOKEN_NEW', 'charlie@windmill.dev', 'Charlie new token', false, 'u/charlie', 'test-workspace')",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "8515537f73c132e62c4dafad1e8d8e56f0c70dccd0edbc9667f550b31ab54c18"
}
backend/.sqlx/query-8aebd7f7fd1374f1c3d5389e953ebf080df3f76ac3e6e6373a89c8d46388125d.json
Generated
-18
View File
@@ -1,18 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token, email, label, expiration, super_admin)\n VALUES ($1, $2, $3, now() + ($4 || ' seconds')::interval, $5)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "8aebd7f7fd1374f1c3d5389e953ebf080df3f76ac3e6e6373a89c8d46388125d"
}
backend/.sqlx/query-90092c0b3f7612373fcc8fb7a966200118ab308430d4a0cbb5cb16c397246492.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT token\n FROM token\n WHERE token LIKE concat($1::text, '%')\n LIMIT 1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "token",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "90092c0b3f7612373fcc8fb7a966200118ab308430d4a0cbb5cb16c397246492"
}
backend/.sqlx/query-c7d595d2a12228c49359440ca3a9622f1de5f5ee4bbe5d2b23f6fdb6379cebf3.json → backend/.sqlx/query-93aa569329a85799594606a4f77fe955820f7b2761df6b38a6a6615b518188f9.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE token SET last_used_at = now() WHERE\n token = $1\n AND (expiration > NOW() OR expiration IS NULL)\n AND (workspace_id IS NULL OR workspace_id = $2)\n RETURNING owner, email, super_admin, scopes, label",
"query": "UPDATE token SET last_used_at = now() WHERE\n token_hash = $1\n AND (expiration > NOW() OR expiration IS NULL)\n AND (workspace_id IS NULL OR workspace_id = $2)\n RETURNING owner, email, super_admin, scopes, label",
"describe": {
"columns": [
{
@@ -43,5 +43,5 @@
true
]
},
"hash": "c7d595d2a12228c49359440ca3a9622f1de5f5ee4bbe5d2b23f6fdb6379cebf3"
"hash": "93aa569329a85799594606a4f77fe955820f7b2761df6b38a6a6615b518188f9"
}
backend/.sqlx/query-b7ed94e891ad12a03a6d9083772aa7c9bfcecdd1863b64f9d75bb73805bc72c8.json
Generated
+19
View File
@@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token_hash, token_prefix, email, label, expiration, super_admin)\n VALUES ($1, $2, $3, $4, now() + ($5 || ' seconds')::interval, $6)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "b7ed94e891ad12a03a6d9083772aa7c9bfcecdd1863b64f9d75bb73805bc72c8"
}
backend/.sqlx/query-be51860f18da9c516cd926efa8d5b35bcfbd01015274c38deb21ed7edb718983.json
Generated
+19
View File
@@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token_hash, token_prefix, email, label, expiration, super_admin)\n VALUES ($1, $2, $3, $4, $5, $6)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Timestamptz",
"Bool"
]
},
"nullable": []
},
"hash": "be51860f18da9c516cd926efa8d5b35bcfbd01015274c38deb21ed7edb718983"
}
backend/.sqlx/query-55cf43cb9219b43f8e9f94b23b62846cd0b1ef5f64d20b0d975d0058730f427b.json → backend/.sqlx/query-ca8997323e27f99bfc5ad8c4a54224d43eaab99b9f1b7d55eff25b0225bb1504.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT job FROM token WHERE token = $1",
"query": "SELECT job FROM token WHERE token_hash = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "55cf43cb9219b43f8e9f94b23b62846cd0b1ef5f64d20b0d975d0058730f427b"
"hash": "ca8997323e27f99bfc5ad8c4a54224d43eaab99b9f1b7d55eff25b0225bb1504"
}
backend/.sqlx/query-d32448f6b329cf98dad42b218a630c0cf40a99edb4ae9fe3e9be485ab1077b3a.json
Generated
-19
View File
@@ -1,19 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token (token, email, label, expiration, scopes, workspace_id)\n SELECT $1::varchar, $2::varchar, $3::varchar, now() + ($4 || ' seconds')::interval, $5::text[], $6::varchar\n WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $6 AND deleted = true)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Text",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "d32448f6b329cf98dad42b218a630c0cf40a99edb4ae9fe3e9be485ab1077b3a"
}
backend/.sqlx/query-e33be0991702ae3a295db7defc6d19d914307a95d72bb0fb447e5b367d52f6a0.json
Generated
-20
View File
@@ -1,20 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO token\n (token, email, label, expiration, super_admin, scopes, workspace_id)\n SELECT $1, $2, $3, $4, $5, $6, $7\n WHERE $7::varchar IS NULL OR NOT EXISTS(\n SELECT 1 FROM workspace WHERE id = $7 AND deleted = true\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Timestamptz",
"Bool",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "e33be0991702ae3a295db7defc6d19d914307a95d72bb0fb447e5b367d52f6a0"
}
backend/.sqlx/query-3b746f73abbaea3570b9c79af21d4d0f60232098d69b71c21fd3da985f7a5905.json → backend/.sqlx/query-ebc2eed287f93e184ed683feb20432caa6e6682620c90f38b29dd32b9a8fe633.json
Generated
+4 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, last_used_at, scopes FROM token WHERE email = $1\n ORDER BY created_at DESC LIMIT $2 OFFSET $3",
"query": "SELECT label, token_prefix, expiration, created_at, last_used_at, scopes FROM token WHERE email = $1\n ORDER BY created_at DESC LIMIT $2 OFFSET $3",
"describe": {
"columns": [
{
@@ -11,7 +11,7 @@
{
"ordinal": 1,
"name": "token_prefix",
"type_info": "Text"
"type_info": "Varchar"
},
{
"ordinal": 2,
@@ -43,12 +43,12 @@
},
"nullable": [
true,
null,
false,
true,
false,
false,
true
]
},
"hash": "3b746f73abbaea3570b9c79af21d4d0f60232098d69b71c21fd3da985f7a5905"
"hash": "ebc2eed287f93e184ed683feb20432caa6e6682620c90f38b29dd32b9a8fe633"
}
backend/migrations/20260304000000_token_hash.down.sql
+7
View File
@@ -0,0 +1,7 @@
-- Reverse of step 1: drop indexes and columns
DROP INDEX IF EXISTS idx_token_prefix;
DROP INDEX IF EXISTS token_hash_unique;
ALTER TABLE token DROP COLUMN token_hash;
ALTER TABLE token DROP COLUMN token_prefix;
backend/migrations/20260304000000_token_hash.up.sql
+26
View File
@@ -0,0 +1,26 @@
-- Step 1: Add columns, backfill, build indexes.
-- This migration does the heavy work but avoids ACCESS EXCLUSIVE during index build
-- by creating the unique index first, then using it for the PK swap in the next migration.
-- Add new columns (instant metadata change)
ALTER TABLE token ADD COLUMN token_hash VARCHAR(64);
ALTER TABLE token ADD COLUMN token_prefix VARCHAR(10);
-- Backfill existing tokens using built-in sha256() (no extension needed).
-- Takes ROW EXCLUSIVE lock — concurrent reads and non-token writes proceed normally.
UPDATE token
SET token_hash = encode(sha256(token::bytea), 'hex'),
token_prefix = substring(token for 10)
WHERE token_hash IS NULL;
-- Mark NOT NULL (instant on PG 12+ when all rows already satisfy the constraint)
ALTER TABLE token ALTER COLUMN token_hash SET NOT NULL;
ALTER TABLE token ALTER COLUMN token_prefix SET NOT NULL;
-- Build the unique index that the next migration will promote to PK.
-- Takes SHARE lock (reads OK, writes wait) but only for the duration of the build,
-- which is fast since token tables are typically small.
CREATE UNIQUE INDEX token_hash_unique ON token (token_hash);
-- Index on prefix for deletion/listing
CREATE INDEX idx_token_prefix ON token (token_prefix);
backend/migrations/20260304000001_token_hash_pk_swap.down.sql
+23
View File
@@ -0,0 +1,23 @@
-- Reverse step 2: restore old PK and trigger
-- Restore the original trigger
CREATE OR REPLACE FUNCTION notify_token_invalidation()
RETURNS TRIGGER AS $$
BEGIN
IF OLD.label = 'session' AND OLD.email IS NOT NULL THEN
INSERT INTO notify_event (channel, payload)
VALUES ('notify_token_invalidation', OLD.token);
END IF;
RETURN OLD;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;
-- Make token NOT NULL again
ALTER TABLE token ALTER COLUMN token SET NOT NULL;
-- Swap PK back: drop token_hash PK, restore token PK
ALTER TABLE token DROP CONSTRAINT token_pkey;
ALTER TABLE token ADD PRIMARY KEY (token);
-- Re-create the unique index on token_hash (was consumed by ADD CONSTRAINT ... USING INDEX)
CREATE UNIQUE INDEX token_hash_unique ON token (token_hash);
backend/migrations/20260304000001_token_hash_pk_swap.up.sql
+22
View File
@@ -0,0 +1,22 @@
-- Step 2: Swap PK and update trigger.
-- All operations here are instant metadata changes (no data/index rebuild).
-- The ACCESS EXCLUSIVE lock is held for only milliseconds.
-- Swap primary key: drop old, promote existing unique index (instant)
ALTER TABLE token DROP CONSTRAINT token_pkey;
ALTER TABLE token ADD CONSTRAINT token_pkey PRIMARY KEY USING INDEX token_hash_unique;
-- Make old token column nullable (no longer written for new tokens)
ALTER TABLE token ALTER COLUMN token DROP NOT NULL;
-- Update the cache invalidation trigger to send prefix instead of plaintext
CREATE OR REPLACE FUNCTION notify_token_invalidation()
RETURNS TRIGGER AS $$
BEGIN
IF OLD.label = 'session' AND OLD.email IS NOT NULL THEN
INSERT INTO notify_event (channel, payload)
VALUES ('notify_token_invalidation', OLD.token_prefix);
END IF;
RETURN OLD;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;
backend/src/main.rs
+1 -1
View File
@@ -1563,7 +1563,7 @@ async fn process_notify_event(
}
"notify_token_invalidation" => {
tracing::info!(
"Token invalidation detected for token: {}...",
"Token invalidation detected for prefix: {}...",
payload.get(..8).unwrap_or(payload)
);
windmill_api::auth::invalidate_token_from_cache(payload);
backend/src/monitor.rs
+1 -1
View File
@@ -847,7 +847,7 @@ struct LogFile {
pub async fn delete_expired_items(db: &DB) -> () {
let tokens_deleted_r: std::result::Result<Vec<String>, _> = sqlx::query_scalar(
"DELETE FROM token WHERE expiration <= now()
RETURNING concat(substring(token for 10), '*****')",
RETURNING concat(token_prefix, '*****')",
)
.fetch_all(db)
.await;
backend/tests/fixtures/permissions_test.sql
Vendored
+6 -6
View File
@@ -39,13 +39,13 @@ ON CONFLICT (email) DO NOTHING;
-- Tokens associated with emails (workspace-scoped)
-- The auth system will look up the user by email in the usr table
-- Note: tokens must be at least 10 characters (TOKEN_PREFIX_LEN)
INSERT INTO token (token, email, label, super_admin, owner, workspace_id)
INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, owner, workspace_id)
VALUES
('ADMIN_TOKEN_TEST', 'admin@windmill.dev', 'Admin token', false, 'u/admin', 'test-workspace'),
('ALICE_TOKEN_TEST', 'alice@windmill.dev', 'Alice token', false, 'u/alice', 'test-workspace'),
('BOB_TOKEN_TEST12', 'bob@windmill.dev', 'Bob token', false, 'u/bob', 'test-workspace'),
('CHARLIE_TOKEN_01', 'charlie@windmill.dev', 'Charlie token', false, 'u/charlie', 'test-workspace'),
('OPERATOR_TOKEN_1', 'operator@windmill.dev', 'Operator token', false, 'u/operator', 'test-workspace');
(encode(sha256('ADMIN_TOKEN_TEST'::bytea), 'hex'), 'ADMIN_TOKE', 'ADMIN_TOKEN_TEST', 'admin@windmill.dev', 'Admin token', false, 'u/admin', 'test-workspace'),
(encode(sha256('ALICE_TOKEN_TEST'::bytea), 'hex'), 'ALICE_TOKE', 'ALICE_TOKEN_TEST', 'alice@windmill.dev', 'Alice token', false, 'u/alice', 'test-workspace'),
(encode(sha256('BOB_TOKEN_TEST12'::bytea), 'hex'), 'BOB_TOKEN_', 'BOB_TOKEN_TEST12', 'bob@windmill.dev', 'Bob token', false, 'u/bob', 'test-workspace'),
(encode(sha256('CHARLIE_TOKEN_01'::bytea), 'hex'), 'CHARLIE_TO', 'CHARLIE_TOKEN_01', 'charlie@windmill.dev', 'Charlie token', false, 'u/charlie', 'test-workspace'),
(encode(sha256('OPERATOR_TOKEN_1'::bytea), 'hex'), 'OPERATOR_T', 'OPERATOR_TOKEN_1', 'operator@windmill.dev', 'Operator token', false, 'u/operator', 'test-workspace');
-- ============================================
-- GROUPS
backend/tests/fixtures/preserve_on_behalf_of.sql
Vendored
+9 -5
View File
@@ -65,14 +65,18 @@ INSERT INTO usr_to_group(workspace_id, group_, usr) VALUES
('test-workspace', 'wm_deployers', 'deployer-user')
ON CONFLICT DO NOTHING;
-- Tokens for all users
INSERT INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN', 'test@windmill.dev', 'test token', true)
-- Tokens for all users (token_hash = sha256 hex, token_prefix = first 10 chars)
INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin)
VALUES (encode(sha256('SECRET_TOKEN'::bytea), 'hex'), 'SECRET_TOK', 'SECRET_TOKEN', 'test@windmill.dev', 'test token', true)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN_2', 'test2@windmill.dev', 'test token 2', false)
INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin)
VALUES (encode(sha256('SECRET_TOKEN_2'::bytea), 'hex'), 'SECRET_TOK', 'SECRET_TOKEN_2', 'test2@windmill.dev', 'test token 2', false)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('DEPLOYER_TOKEN', 'deployer@windmill.dev', 'deployer token', false)
INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin)
VALUES (encode(sha256('DEPLOYER_TOKEN'::bytea), 'hex'), 'DEPLOYER_T', 'DEPLOYER_TOKEN', 'deployer@windmill.dev', 'deployer token', false)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('ORIGINAL_TOKEN', 'original@windmill.dev', 'original token', false)
INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin)
VALUES (encode(sha256('ORIGINAL_TOKEN'::bytea), 'hex'), 'ORIGINAL_T', 'ORIGINAL_TOKEN', 'original@windmill.dev', 'original token', false)
ON CONFLICT DO NOTHING;
GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_admin;
backend/windmill-api-auth/src/auth.rs
+17 -12
View File
@@ -26,7 +26,9 @@ use tokio::sync::RwLock;
use windmill_common::DB;
use windmill_common::{
auth::{get_folders_for_user, get_groups_for_user, JWTAuthClaims, TOKEN_PREFIX_LEN},
auth::{
get_folders_for_user, get_groups_for_user, hash_token, JWTAuthClaims, TOKEN_PREFIX_LEN,
},
error::{Error, JsonResult},
jwt,
users::{COOKIE_NAME, SUPERADMIN_SECRET_EMAIL},
@@ -37,13 +39,15 @@ lazy_static::lazy_static! {
pub static ref AUTH_CACHE: Cache<(String, String), ExpiringAuthCache> = Cache::new(300);
}
// Global function to invalidate a specific token from cache
pub fn invalidate_token_from_cache(token: &str) {
// Remove all cache entries for this token (across all workspaces)
AUTH_CACHE.retain(|(_workspace_id, cached_token), _cached_value| cached_token != token);
// 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 from auth cache: {}...",
&token[..token.len().min(8)]
"Invalidated token(s) from auth cache with prefix: {}...",
&token_prefix[..token_prefix.len().min(8)]
);
}
@@ -173,13 +177,14 @@ impl AuthCache {
}
}
_ => {
let t_hash = hash_token(token);
let user_o = sqlx::query!(
"UPDATE token SET last_used_at = now() WHERE
token = $1
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",
token,
t_hash,
w_id.as_ref(),
)
.map(|x| (x.owner, x.email, x.super_admin, x.scopes, x.label))
@@ -679,7 +684,7 @@ fn username_override_from_label(label: Option<String>) -> Option<String> {
#[derive(FromRow, Serialize)]
pub struct TruncatedTokenWithEmail {
pub label: Option<String>,
pub token_prefix: 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>,
@@ -698,7 +703,7 @@ pub async fn list_tokens_internal(
TruncatedTokenWithEmail,
r#"
SELECT label,
concat(substring(token for 10)) AS token_prefix,
token_prefix,
expiration,
created_at,
last_used_at,
@@ -721,7 +726,7 @@ pub async fn list_tokens_internal(
TruncatedTokenWithEmail,
r#"
SELECT label,
concat(substring(token for 10)) AS token_prefix,
token_prefix,
expiration,
created_at,
last_used_at,
backend/windmill-api-auth/src/lib.rs
+13 -7
View File
@@ -18,7 +18,10 @@ use http::request::Parts;
use windmill_audit::audit_oss::AuditAuthorable;
use windmill_common::{
auth::{fetch_authed_from_permissioned_as, is_devops_email, is_super_admin_email},
auth::{
fetch_authed_from_permissioned_as, hash_token, is_devops_email, is_super_admin_email,
TOKEN_PREFIX_LEN,
},
db::{Authable, Authed, AuthedRef},
error::{self, Error, Result},
users::username_to_permissioned_as,
@@ -514,6 +517,8 @@ pub async fn create_token_internal(
use windmill_common::{utils::rd_string, worker::CLOUD_HOSTED};
let token = rd_string(32);
let t_hash = hash_token(&token);
let t_prefix = &token[..TOKEN_PREFIX_LEN];
let is_super_admin = sqlx::query_scalar!(
"SELECT super_admin FROM password WHERE email = $1",
@@ -536,12 +541,13 @@ pub async fn create_token_internal(
}
let rows = sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin, scopes, workspace_id)
SELECT $1, $2, $3, $4, $5, $6, $7
WHERE $7::varchar IS NULL OR NOT EXISTS(
SELECT 1 FROM workspace WHERE id = $7 AND deleted = true
(token_hash, token_prefix, email, label, expiration, super_admin, scopes, workspace_id)
SELECT $1, $2, $3, $4, $5, $6, $7, $8
WHERE $8::varchar IS NULL OR NOT EXISTS(
SELECT 1 FROM workspace WHERE id = $8 AND deleted = true
)",
token,
t_hash,
t_prefix,
authed.email,
token_config.label,
token_config.expiration,
@@ -563,7 +569,7 @@ pub async fn create_token_internal(
"users.token.create",
ActionKind::Create,
&"global",
Some(&token[0..10]),
Some(t_prefix),
None,
)
.instrument(tracing::info_span!("token", email = &authed.email))
backend/windmill-api-integration-tests/tests/fixtures/permissions_test.sql
Vendored
+6 -6
View File
@@ -39,13 +39,13 @@ ON CONFLICT (email) DO NOTHING;
-- Tokens associated with emails (workspace-scoped)
-- The auth system will look up the user by email in the usr table
-- Note: tokens must be at least 10 characters (TOKEN_PREFIX_LEN)
INSERT INTO token (token, email, label, super_admin, owner, workspace_id)
INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, owner, workspace_id)
VALUES
('ADMIN_TOKEN_TEST', 'admin@windmill.dev', 'Admin token', false, 'u/admin', 'test-workspace'),
('ALICE_TOKEN_TEST', 'alice@windmill.dev', 'Alice token', false, 'u/alice', 'test-workspace'),
('BOB_TOKEN_TEST12', 'bob@windmill.dev', 'Bob token', false, 'u/bob', 'test-workspace'),
('CHARLIE_TOKEN_01', 'charlie@windmill.dev', 'Charlie token', false, 'u/charlie', 'test-workspace'),
('OPERATOR_TOKEN_1', 'operator@windmill.dev', 'Operator token', false, 'u/operator', 'test-workspace');
(encode(sha256('ADMIN_TOKEN_TEST'::bytea), 'hex'), 'ADMIN_TOKE', 'ADMIN_TOKEN_TEST', 'admin@windmill.dev', 'Admin token', false, 'u/admin', 'test-workspace'),
(encode(sha256('ALICE_TOKEN_TEST'::bytea), 'hex'), 'ALICE_TOKE', 'ALICE_TOKEN_TEST', 'alice@windmill.dev', 'Alice token', false, 'u/alice', 'test-workspace'),
(encode(sha256('BOB_TOKEN_TEST12'::bytea), 'hex'), 'BOB_TOKEN_', 'BOB_TOKEN_TEST12', 'bob@windmill.dev', 'Bob token', false, 'u/bob', 'test-workspace'),
(encode(sha256('CHARLIE_TOKEN_01'::bytea), 'hex'), 'CHARLIE_TO', 'CHARLIE_TOKEN_01', 'charlie@windmill.dev', 'Charlie token', false, 'u/charlie', 'test-workspace'),
(encode(sha256('OPERATOR_TOKEN_1'::bytea), 'hex'), 'OPERATOR_T', 'OPERATOR_TOKEN_1', 'operator@windmill.dev', 'Operator token', false, 'u/operator', 'test-workspace');
-- ============================================
-- GROUPS
backend/windmill-api-integration-tests/tests/permissions.rs
+94 -19
View File
@@ -517,9 +517,14 @@ async fn test_group_permission_inheritance(db: Pool<Postgres>) -> anyhow::Result
// for (workspace_id, token) tuples. Since we can't easily clear it from tests,
// we use a different token or wait for cache expiry. For this test, we create
// a new token for Charlie.
let charlie_token = "CHARLIE_TOKEN_NEW";
let charlie_token_hash = windmill_common::utils::calculate_hash(charlie_token);
let charlie_token_prefix = &charlie_token[..10.min(charlie_token.len())];
sqlx::query!(
"INSERT INTO token (token, email, label, super_admin, owner, workspace_id)
VALUES ('CHARLIE_TOKEN_NEW', 'charlie@windmill.dev', 'Charlie new token', false, 'u/charlie', 'test-workspace')"
"INSERT INTO token (token_hash, token_prefix, email, label, super_admin, owner, workspace_id)
VALUES ($1, $2, 'charlie@windmill.dev', 'Charlie new token', false, 'u/charlie', 'test-workspace')",
charlie_token_hash,
charlie_token_prefix,
)
.execute(&db)
.await?;
@@ -563,28 +568,100 @@ async fn test_all_item_types_permissions(db: Pool<Postgres>) -> anyhow::Result<(
let bob_client = create_client_for_user(port, "BOB_TOKEN_TEST12").await;
// Test Scripts - uses /scripts/get/p/{path}
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/scripts/get/p/u/alice/my_script")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/scripts/get/p/u/alice/my_script")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/scripts/get/p/u/alice/my_script")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/scripts/get/p/u/alice/my_script")
)
.await
);
// Test Flows - uses /flows/get/{path} (no /p/)
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/flows/get/u/alice/my_flow")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/flows/get/u/alice/my_flow")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/flows/get/u/alice/my_flow")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/flows/get/u/alice/my_flow")
)
.await
);
// Test Resources - uses /resources/get/{path} (no /p/)
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/resources/get/u/alice/my_resource")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/resources/get/u/alice/my_resource")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/resources/get/u/alice/my_resource")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/resources/get/u/alice/my_resource")
)
.await
);
// Test Variables - uses /variables/get/{path} (no /p/)
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/variables/get/u/alice/my_variable")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/variables/get/u/alice/my_variable")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/variables/get/u/alice/my_variable")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/variables/get/u/alice/my_variable")
)
.await
);
// Test Schedules - uses /schedules/get/{path} (no /p/)
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/schedules/get/u/alice/my_schedule")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/schedules/get/u/alice/my_schedule")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/schedules/get/u/alice/my_schedule")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/schedules/get/u/alice/my_schedule")
)
.await
);
// Test Apps - uses /apps/get/p/{path}
assert!(can_read(&alice_client, &format!("{base_url}/w/test-workspace/apps/get/p/u/alice/my_app")).await);
assert!(!can_read(&bob_client, &format!("{base_url}/w/test-workspace/apps/get/p/u/alice/my_app")).await);
assert!(
can_read(
&alice_client,
&format!("{base_url}/w/test-workspace/apps/get/p/u/alice/my_app")
)
.await
);
assert!(
!can_read(
&bob_client,
&format!("{base_url}/w/test-workspace/apps/get/p/u/alice/my_app")
)
.await
);
Ok(())
}
@@ -747,11 +824,9 @@ async fn test_operator_cannot_create_update(db: Pool<Postgres>) -> anyhow::Resul
.await?;
// Update app versions
sqlx::query!(
"UPDATE app SET versions = ARRAY[3001::bigint] WHERE id = 3001"
)
.execute(&db)
.await?;
sqlx::query!("UPDATE app SET versions = ARRAY[3001::bigint] WHERE id = 3001")
.execute(&db)
.await?;
let update_app = json!({
"path": "u/operator/existing_app",
backend/windmill-api-users/src/users.rs
+32 -20
View File
@@ -235,7 +235,7 @@ pub struct EditLoginType {
#[derive(FromRow, Serialize)]
pub struct TruncatedToken {
pub label: Option<String>,
pub token_prefix: 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>,
@@ -527,23 +527,28 @@ async fn logout(
}
cookies.remove(cookie);
let mut tx = db.begin().await?;
let t_hash = windmill_common::auth::hash_token(&token);
let t_prefix = &token[..TOKEN_PREFIX_LEN];
let email = if *INVALIDATE_ALL_SESSIONS_ON_LOGOUT {
sqlx::query_scalar!(
"WITH email_lookup AS (
SELECT email FROM token WHERE token = $1
SELECT email FROM token WHERE token_hash = $1
)
DELETE FROM token
WHERE email = (SELECT email FROM email_lookup) AND label = 'session'
RETURNING email",
token
t_hash
)
.fetch_optional(&mut *tx)
.await?
} else {
sqlx::query_scalar!("DELETE FROM token WHERE token = $1 RETURNING email", token)
.fetch_optional(&mut *tx)
.await?
sqlx::query_scalar!(
"DELETE FROM token WHERE token_hash = $1 RETURNING email",
t_hash
)
.fetch_optional(&mut *tx)
.await?
};
if let Some(email) = email {
@@ -559,7 +564,7 @@ async fn logout(
email: email.clone(),
username: email,
username_override: None,
token_prefix: Some(token[0..TOKEN_PREFIX_LEN].to_string()),
token_prefix: Some(t_prefix.to_string()),
},
audit_message,
ActionKind::Delete,
@@ -1689,7 +1694,8 @@ async fn refresh_token(
let mut tx = db.begin().await?;
if let Some(thresh_s) = query.if_expiring_in_less_than_s {
let not_expired = sqlx::query_scalar!("SELECT true FROM token WHERE token = $1 and expiration IS NOT NULL and expiration > now() + $2::int * '1 sec'::interval", &token, thresh_s)
let t_hash = windmill_common::auth::hash_token(&token);
let not_expired = sqlx::query_scalar!("SELECT true FROM token WHERE token_hash = $1 and expiration IS NOT NULL and expiration > now() + $2::int * '1 sec'::interval", &t_hash, thresh_s)
.fetch_optional(&db)
.await?
.flatten()
@@ -1741,6 +1747,8 @@ pub async fn create_session_token<'c>(
cookies: Cookies,
) -> Result<String> {
let token = rd_string(32);
let t_hash = windmill_common::auth::hash_token(&token);
let t_prefix = &token[..TOKEN_PREFIX_LEN];
if *INVALIDATE_OLD_SESSIONS {
sqlx::query!(
@@ -1756,7 +1764,7 @@ pub async fn create_session_token<'c>(
email: email.to_string(),
username: email.to_string(),
username_override: None,
token_prefix: Some(token[0..TOKEN_PREFIX_LEN].to_string()),
token_prefix: Some(t_prefix.to_string()),
},
"users.token.invalidate_old_sessions",
ActionKind::Delete,
@@ -1770,9 +1778,10 @@ pub async fn create_session_token<'c>(
sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin)
VALUES ($1, $2, $3, now() + ($4 || ' seconds')::interval, $5)",
token,
(token_hash, token_prefix, email, label, expiration, super_admin)
VALUES ($1, $2, $3, $4, now() + ($5 || ' seconds')::interval, $6)",
t_hash,
t_prefix,
email,
"session",
&MAX_SESSION_VALIDITY_SECONDS.to_string(),
@@ -1818,6 +1827,8 @@ async fn impersonate(
Json(new_token): Json<NewToken>,
) -> Result<(StatusCode, String)> {
let token = rd_string(32);
let t_hash = windmill_common::auth::hash_token(&token);
let t_prefix = &token[..TOKEN_PREFIX_LEN];
require_super_admin(&db, &authed.email).await?;
if new_token.impersonate_email.is_none() {
@@ -1839,9 +1850,10 @@ async fn impersonate(
sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin)
VALUES ($1, $2, $3, $4, $5)",
token,
(token_hash, token_prefix, email, label, expiration, super_admin)
VALUES ($1, $2, $3, $4, $5, $6)",
t_hash,
t_prefix,
impersonated,
new_token.label,
new_token.expiration,
@@ -1856,7 +1868,7 @@ async fn impersonate(
"users.impersonate",
ActionKind::Delete,
&"global",
Some(&token[0..10]),
Some(t_prefix),
Some([("impersonated", &format!("{impersonated}")[..])].into()),
)
.instrument(tracing::info_span!("token", email = &impersonated))
@@ -1880,7 +1892,7 @@ async fn list_tokens(
let rows = if query.exclude_ephemeral.unwrap_or(false) {
sqlx::query_as!(
TruncatedToken,
"SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, \
"SELECT label, token_prefix, expiration, created_at, \
last_used_at, scopes FROM token WHERE email = $1 AND (label != 'ephemeral-script' OR label IS NULL)
ORDER BY created_at DESC LIMIT $2 OFFSET $3",
email,
@@ -1892,7 +1904,7 @@ async fn list_tokens(
} else {
sqlx::query_as!(
TruncatedToken,
"SELECT label, concat(substring(token for 10)) as token_prefix, expiration, created_at, \
"SELECT label, token_prefix, expiration, created_at, \
last_used_at, scopes FROM token WHERE email = $1
ORDER BY created_at DESC LIMIT $2 OFFSET $3",
email,
@@ -1915,8 +1927,8 @@ async fn delete_token(
let tokens_deleted: Vec<String> = sqlx::query_scalar(
"DELETE FROM token
WHERE email = $1
AND token LIKE concat($2::text, '%')
RETURNING concat(substring(token for 10), '*****')",
AND token_prefix = $2
RETURNING concat(token_prefix, '*****')",
)
.bind(&authed.email)
.bind(&token_prefix)
backend/windmill-api-workspaces/src/workspaces.rs
+1 -1
View File
@@ -3577,7 +3577,7 @@ pub(crate) async fn archive_workspace_impl(
// Delete non-session tokens scoped to this workspace
let deleted_tokens = sqlx::query_scalar!(
"DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token",
"DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token_prefix",
w_id
)
.fetch_all(&mut *tx)
backend/windmill-api/src/mcp/oauth_server.rs
+26 -16
View File
@@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use sqlx::FromRow;
use windmill_common::{
auth::{hash_token, TOKEN_PREFIX_LEN},
error::{Error, Result},
utils::rd_string,
BASE_URL, DB,
@@ -383,16 +384,19 @@ async fn handle_authorization_code_grant(
}
let access_token = rd_string(32);
let access_token_hash = hash_token(&access_token);
let access_token_prefix = &access_token[..TOKEN_PREFIX_LEN];
let refresh_token = rd_string(32);
let token_family = sqlx::types::Uuid::new_v4();
let scopes = auth_code.scopes;
// Create access token (rejects archived workspaces inline)
let rows = sqlx::query!(
"INSERT INTO token (token, email, label, expiration, scopes, workspace_id)
SELECT $1::varchar, $2::varchar, $3::varchar, now() + ($4 || ' seconds')::interval, $5::text[], $6::varchar
WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $6 AND deleted = true)",
access_token,
"INSERT INTO token (token_hash, token_prefix, email, label, expiration, scopes, workspace_id)
SELECT $1::varchar, $2::varchar, $3::varchar, $4::varchar, now() + ($5 || ' seconds')::interval, $6::text[], $7::varchar
WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $7 AND deleted = true)",
access_token_hash,
access_token_prefix,
auth_code.user_email,
format!("mcp-oauth-{}", auth_code.client_id),
MCP_OAUTH_TOKEN_EXPIRATION_SECS.to_string(),
@@ -411,13 +415,13 @@ async fn handle_authorization_code_grant(
));
}
// Create refresh token
// Create refresh token — store the hash of the access token so we can delete it later
let refresh_token_result = sqlx::query!(
"INSERT INTO mcp_oauth_refresh_token
(refresh_token, access_token, client_id, user_email, workspace_id, scopes, token_family, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, now() + ($8 || ' seconds')::interval)",
refresh_token,
access_token,
access_token_hash,
auth_code.client_id,
auth_code.user_email,
auth_code.workspace_id,
@@ -504,10 +508,13 @@ async fn handle_refresh_token_grant(
}
};
// Delete old access token
if let Err(e) = sqlx::query!("DELETE FROM token WHERE token = $1", token_row.access_token)
.execute(db)
.await
// Delete old access token — access_token column in mcp_oauth_refresh_token now stores the hash
if let Err(e) = sqlx::query!(
"DELETE FROM token WHERE token_hash = $1",
token_row.access_token
)
.execute(db)
.await
{
tracing::error!("Failed to delete old access token: {}", e);
// Non-fatal, continue with token creation
@@ -515,15 +522,18 @@ async fn handle_refresh_token_grant(
// Generate new tokens
let new_access_token = rd_string(32);
let new_access_token_hash = hash_token(&new_access_token);
let new_access_token_prefix = &new_access_token[..TOKEN_PREFIX_LEN];
let new_refresh_token = rd_string(32);
let scopes = token_row.scopes;
// Create new access token (rejects archived workspaces inline)
let rows = sqlx::query!(
"INSERT INTO token (token, email, label, expiration, scopes, workspace_id)
SELECT $1::varchar, $2::varchar, $3::varchar, now() + ($4 || ' seconds')::interval, $5::text[], $6::varchar
WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $6 AND deleted = true)",
new_access_token,
"INSERT INTO token (token_hash, token_prefix, email, label, expiration, scopes, workspace_id)
SELECT $1::varchar, $2::varchar, $3::varchar, $4::varchar, now() + ($5 || ' seconds')::interval, $6::text[], $7::varchar
WHERE NOT EXISTS(SELECT 1 FROM workspace WHERE id = $7 AND deleted = true)",
new_access_token_hash,
new_access_token_prefix,
token_row.user_email,
format!("mcp-oauth-{}", token_row.client_id),
MCP_OAUTH_TOKEN_EXPIRATION_SECS.to_string(),
@@ -542,13 +552,13 @@ async fn handle_refresh_token_grant(
));
}
// Create new refresh token (same token family for tracking)
// Create new refresh token (same token family for tracking) — store hash of access token
if let Err(e) = sqlx::query!(
"INSERT INTO mcp_oauth_refresh_token
(refresh_token, access_token, client_id, user_email, workspace_id, scopes, token_family, expires_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, now() + ($8 || ' seconds')::interval)",
new_refresh_token,
new_access_token,
new_access_token_hash,
token_row.client_id,
token_row.user_email,
token_row.workspace_id,
backend/windmill-common/src/auth.rs
+6
View File
@@ -18,6 +18,12 @@ use crate::{
DB,
};
/// Hash a raw token using SHA-256 (hex-encoded, 64 chars).
/// Used to store and look up tokens without keeping plaintext in the DB.
pub fn hash_token(token: &str) -> String {
crate::utils::calculate_hash(token)
}
#[derive(Debug)]
pub struct IdToken {
token: String,
backend/windmill-common/tests/notify_events.rs
+250 -85
View File
@@ -40,13 +40,20 @@ async fn count_events_for_channel(db: &Pool<Postgres>, channel: &str) -> i64 {
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_get_latest_event_id_returns_valid_id(db: Pool<Postgres>) {
// Get current latest id
let latest_id = get_latest_event_id(&db).await.expect("Should get latest event id");
let latest_id = get_latest_event_id(&db)
.await
.expect("Should get latest event id");
assert!(latest_id >= 0, "Latest id should be non-negative");
// Insert a new event and verify latest_id increases
let new_id = insert_test_event(&db, "test_latest_id", "payload").await;
let new_latest_id = get_latest_event_id(&db).await.expect("Should get latest event id");
assert!(new_latest_id >= new_id, "Latest id should be >= new event id");
let new_latest_id = get_latest_event_id(&db)
.await
.expect("Should get latest event id");
assert!(
new_latest_id >= new_id,
"Latest id should be >= new event id"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -55,7 +62,9 @@ async fn test_get_latest_event_id_with_events(db: Pool<Postgres>) {
let _id2 = insert_test_event(&db, "test_channel_2", "payload2").await;
let id3 = insert_test_event(&db, "test_channel_3", "payload3").await;
let latest_id = get_latest_event_id(&db).await.expect("Should get latest event id");
let latest_id = get_latest_event_id(&db)
.await
.expect("Should get latest event id");
assert!(latest_id >= id3, "Latest id should be >= last inserted id");
}
@@ -65,8 +74,13 @@ async fn test_poll_notify_events_no_new_events(db: Pool<Postgres>) {
let latest_id = get_latest_event_id(&db).await.unwrap();
// Poll from the latest id - should return empty since no new events
let events = poll_notify_events(&db, latest_id).await.expect("Should poll events");
assert!(events.is_empty(), "Should return empty vec when polling from latest id");
let events = poll_notify_events(&db, latest_id)
.await
.expect("Should poll events");
assert!(
events.is_empty(),
"Should return empty vec when polling from latest id"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -76,7 +90,9 @@ async fn test_poll_notify_events_returns_new_events(db: Pool<Postgres>) {
let _id1 = insert_test_event(&db, "test_poll_channel", "payload1").await;
let _id2 = insert_test_event(&db, "test_poll_channel", "payload2").await;
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
assert!(events.len() >= 2, "Should return at least 2 new events");
// Verify the events we inserted are present
@@ -87,7 +103,10 @@ async fn test_poll_notify_events_returns_new_events(db: Pool<Postgres>) {
assert_eq!(our_events.len(), 2, "Should have exactly our 2 test events");
// Verify ordering (ascending by id)
assert!(our_events[0].id < our_events[1].id, "Events should be ordered by id ascending");
assert!(
our_events[0].id < our_events[1].id,
"Events should be ordered by id ascending"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -97,14 +116,19 @@ async fn test_poll_notify_events_respects_last_event_id(db: Pool<Postgres>) {
let _id3 = insert_test_event(&db, "test_respect_id", "payload3").await;
// Poll from id1 should only return id2 and id3
let events = poll_notify_events(&db, id1).await.expect("Should poll events");
let events = poll_notify_events(&db, id1)
.await
.expect("Should poll events");
let our_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "test_respect_id")
.collect();
assert_eq!(our_events.len(), 2, "Should only return events after id1");
assert!(our_events.iter().all(|e| e.id > id1), "All events should have id > id1");
assert!(
our_events.iter().all(|e| e.id > id1),
"All events should have id > id1"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -124,21 +148,24 @@ async fn test_cleanup_old_events(db: Pool<Postgres>) {
.expect("Failed to insert old event");
// Insert a recent event
sqlx::query(
"INSERT INTO notify_event (channel, payload) VALUES ($1, $2)",
)
.bind(&recent_channel)
.bind("recent_payload")
.execute(&db)
.await
.expect("Failed to insert recent event");
sqlx::query("INSERT INTO notify_event (channel, payload) VALUES ($1, $2)")
.bind(&recent_channel)
.bind("recent_payload")
.execute(&db)
.await
.expect("Failed to insert recent event");
// Count before cleanup
let old_count_before = count_events_for_channel(&db, &old_channel).await;
assert_eq!(old_count_before, 1, "Should have 1 old event before cleanup");
assert_eq!(
old_count_before, 1,
"Should have 1 old event before cleanup"
);
// Cleanup events older than 10 minutes
let deleted = cleanup_old_events(&db, 10).await.expect("Should cleanup events");
let deleted = cleanup_old_events(&db, 10)
.await
.expect("Should cleanup events");
assert!(deleted >= 1, "Should delete at least 1 old event");
// Verify old event is gone
@@ -167,13 +194,18 @@ async fn test_trigger_notify_config_change(db: Pool<Postgres>) {
.await
.expect("Failed to insert config");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let config_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_config_change" && e.payload == "test_config_trigger")
.collect();
assert!(!config_events.is_empty(), "Should have notify_config_change event");
assert!(
!config_events.is_empty(),
"Should have notify_config_change event"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -190,13 +222,18 @@ async fn test_trigger_notify_global_setting_change_insert(db: Pool<Postgres>) {
.await
.expect("Failed to insert global setting");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let setting_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name)
.collect();
assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on insert");
assert!(
!setting_events.is_empty(),
"Should have notify_global_setting_change event on insert"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -220,13 +257,18 @@ async fn test_trigger_notify_global_setting_change_update(db: Pool<Postgres>) {
.await
.expect("Failed to update global setting");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let setting_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name)
.collect();
assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on update");
assert!(
!setting_events.is_empty(),
"Should have notify_global_setting_change event on update"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -250,13 +292,18 @@ async fn test_trigger_notify_global_setting_change_delete(db: Pool<Postgres>) {
.await
.expect("Failed to delete global setting");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let setting_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name)
.collect();
assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on delete");
assert!(
!setting_events.is_empty(),
"Should have notify_global_setting_change event on delete"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -272,13 +319,18 @@ async fn test_trigger_notify_workspace_envs_change(db: Pool<Postgres>) {
.await
.expect("Failed to insert workspace env");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let env_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_workspace_envs_change" && e.payload == "test-workspace")
.collect();
assert!(!env_events.is_empty(), "Should have notify_workspace_envs_change event");
assert!(
!env_events.is_empty(),
"Should have notify_workspace_envs_change event"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -294,44 +346,57 @@ async fn test_trigger_notify_workspace_key_change(db: Pool<Postgres>) {
.await
.expect("Failed to insert workspace key");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let key_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_workspace_key_change" && e.payload == "test-workspace")
.collect();
assert!(!key_events.is_empty(), "Should have notify_workspace_key_change event");
assert!(
!key_events.is_empty(),
"Should have notify_workspace_key_change event"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_trigger_notify_token_invalidation(db: Pool<Postgres>) {
// First insert a session token
// First insert a session token with token_hash and token_prefix
let token = format!("test_token_{}", uuid::Uuid::new_v4());
let token_hash = windmill_common::utils::calculate_hash(&token);
let token_prefix = &token[..10];
sqlx::query(
"INSERT INTO token (token, label, email, workspace_id, owner, expiration)
VALUES ($1, 'session', 'test@test.com', 'test-workspace', 'test-user', now() + interval '1 hour')",
"INSERT INTO token (token_hash, token_prefix, label, email, workspace_id, owner, expiration)
VALUES ($1, $2, 'session', 'test@test.com', 'test-workspace', 'test-user', now() + interval '1 hour')",
)
.bind(&token)
.bind(&token_hash)
.bind(token_prefix)
.execute(&db)
.await
.expect("Failed to insert token");
let before_id = get_latest_event_id(&db).await.unwrap();
// Delete the token (should trigger notification)
sqlx::query("DELETE FROM token WHERE token = $1")
.bind(&token)
// Delete the token (should trigger notification with prefix)
sqlx::query("DELETE FROM token WHERE token_hash = $1")
.bind(&token_hash)
.execute(&db)
.await
.expect("Failed to delete token");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let token_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_token_invalidation" && e.payload == token)
.filter(|e| e.channel == "notify_token_invalidation" && e.payload == token_prefix)
.collect();
assert!(!token_events.is_empty(), "Should have notify_token_invalidation event");
assert!(
!token_events.is_empty(),
"Should have notify_token_invalidation event"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -344,13 +409,18 @@ async fn test_trigger_notify_webhook_change(db: Pool<Postgres>) {
.await
.expect("Failed to update webhook");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let webhook_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_webhook_change" && e.payload == "test-workspace")
.collect();
assert!(!webhook_events.is_empty(), "Should have notify_webhook_change event");
assert!(
!webhook_events.is_empty(),
"Should have notify_webhook_change event"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -363,13 +433,18 @@ async fn test_trigger_notify_workspace_premium_change(db: Pool<Postgres>) {
.await
.expect("Failed to update workspace premium");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let premium_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_workspace_premium_change" && e.payload == "test-workspace")
.collect();
assert!(!premium_events.is_empty(), "Should have notify_workspace_premium_change event");
assert!(
!premium_events.is_empty(),
"Should have notify_workspace_premium_change event"
);
}
// ============================================================================
@@ -392,14 +467,19 @@ async fn test_trigger_notify_http_trigger_change(db: Pool<Postgres>) {
.await
.expect("Failed to insert HTTP trigger");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let http_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_http_trigger_change")
.filter(|e| e.payload.contains("test-workspace") && e.payload.contains(&trigger_path))
.collect();
assert!(!http_events.is_empty(), "Should have notify_http_trigger_change event");
assert!(
!http_events.is_empty(),
"Should have notify_http_trigger_change event"
);
}
// ============================================================================
@@ -431,19 +511,27 @@ async fn test_trigger_notify_runnable_version_change_script(db: Pool<Postgres>)
.await
.expect("Failed to update script lock");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let script_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_runnable_version_change")
.filter(|e| e.payload.contains("test-workspace") && e.payload.contains("script"))
.collect();
assert!(!script_events.is_empty(), "Should have notify_runnable_version_change event for script");
assert!(
!script_events.is_empty(),
"Should have notify_runnable_version_change event for script"
);
// Verify payload format: workspace_id:source_type:path:kind
let parts: Vec<&str> = script_events[0].payload.split(':').collect();
assert!(parts.len() >= 4, "Payload should have at least 4 parts");
assert_eq!(parts[0], "test-workspace", "First part should be workspace_id");
assert_eq!(
parts[0], "test-workspace",
"First part should be workspace_id"
);
assert_eq!(parts[1], "script", "Second part should be 'script'");
}
@@ -472,19 +560,27 @@ async fn test_trigger_notify_runnable_version_change_flow(db: Pool<Postgres>) {
.await
.expect("Failed to update flow versions");
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let flow_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "notify_runnable_version_change")
.filter(|e| e.payload.contains("test-workspace") && e.payload.contains("flow"))
.collect();
assert!(!flow_events.is_empty(), "Should have notify_runnable_version_change event for flow");
assert!(
!flow_events.is_empty(),
"Should have notify_runnable_version_change event for flow"
);
// Verify payload format
let parts: Vec<&str> = flow_events[0].payload.split(':').collect();
assert!(parts.len() >= 4, "Payload should have at least 4 parts");
assert_eq!(parts[0], "test-workspace", "First part should be workspace_id");
assert_eq!(
parts[0], "test-workspace",
"First part should be workspace_id"
);
assert_eq!(parts[1], "flow", "Second part should be 'flow'");
}
@@ -521,13 +617,16 @@ async fn test_concurrent_event_insertion(db: Pool<Postgres>) {
handle.await.expect("Task should complete");
}
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let concurrent_events: Vec<_> = events
.iter()
.filter(|e| e.channel == channel)
.collect();
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let concurrent_events: Vec<_> = events.iter().filter(|e| e.channel == channel).collect();
assert_eq!(concurrent_events.len(), 10, "Should have all 10 concurrent events");
assert_eq!(
concurrent_events.len(),
10,
"Should have all 10 concurrent events"
);
// Verify all events have unique IDs
let ids: std::collections::HashSet<i64> = concurrent_events.iter().map(|e| e.id).collect();
@@ -571,18 +670,45 @@ async fn test_polling_isolation(db: Pool<Postgres>) {
.expect("Failed to insert event");
// Two different "consumers" polling from different points
let events_from_baseline = poll_notify_events(&db, baseline_id).await.expect("Should poll events");
let events_from_id1 = poll_notify_events(&db, id1).await.expect("Should poll events");
let events_from_id2 = poll_notify_events(&db, id2).await.expect("Should poll events");
let events_from_baseline = poll_notify_events(&db, baseline_id)
.await
.expect("Should poll events");
let events_from_id1 = poll_notify_events(&db, id1)
.await
.expect("Should poll events");
let events_from_id2 = poll_notify_events(&db, id2)
.await
.expect("Should poll events");
// Filter to our test events
let from_baseline: Vec<_> = events_from_baseline.iter().filter(|e| e.channel == channel).collect();
let from_id1: Vec<_> = events_from_id1.iter().filter(|e| e.channel == channel).collect();
let from_id2: Vec<_> = events_from_id2.iter().filter(|e| e.channel == channel).collect();
let from_baseline: Vec<_> = events_from_baseline
.iter()
.filter(|e| e.channel == channel)
.collect();
let from_id1: Vec<_> = events_from_id1
.iter()
.filter(|e| e.channel == channel)
.collect();
let from_id2: Vec<_> = events_from_id2
.iter()
.filter(|e| e.channel == channel)
.collect();
assert_eq!(from_baseline.len(), 3, "Polling from baseline should include all 3 events");
assert_eq!(from_id1.len(), 2, "Polling from id1 should include id2 and id3");
assert_eq!(from_id2.len(), 1, "Polling from id2 should include only id3");
assert_eq!(
from_baseline.len(),
3,
"Polling from baseline should include all 3 events"
);
assert_eq!(
from_id1.len(),
2,
"Polling from id1 should include id2 and id3"
);
assert_eq!(
from_id2.len(),
1,
"Polling from id2 should include only id3"
);
}
// ============================================================================
@@ -595,14 +721,23 @@ async fn test_empty_payload(db: Pool<Postgres>) {
insert_test_event(&db, "test_empty_payload", "").await;
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let empty_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "test_empty_payload")
.collect();
assert_eq!(empty_events.len(), 1, "Should have event with empty payload");
assert_eq!(empty_events[0].payload, "", "Payload should be empty string");
assert_eq!(
empty_events.len(),
1,
"Should have event with empty payload"
);
assert_eq!(
empty_events[0].payload, "",
"Payload should be empty string"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -613,14 +748,24 @@ async fn test_large_payload(db: Pool<Postgres>) {
let large_payload = "x".repeat(1024);
insert_test_event(&db, "test_large_payload", &large_payload).await;
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let large_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "test_large_payload")
.collect();
assert_eq!(large_events.len(), 1, "Should have event with large payload");
assert_eq!(large_events[0].payload.len(), 1024, "Payload should be preserved");
assert_eq!(
large_events.len(),
1,
"Should have event with large payload"
);
assert_eq!(
large_events[0].payload.len(),
1024,
"Payload should be preserved"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -630,14 +775,23 @@ async fn test_special_characters_in_payload(db: Pool<Postgres>) {
let special_payload = r#"{"key": "value with \"quotes\" and 'apostrophes'", "unicode": "日本語", "newline": "line1\nline2"}"#;
insert_test_event(&db, "test_special_chars", special_payload).await;
let events = poll_notify_events(&db, before_id).await.expect("Should poll events");
let events = poll_notify_events(&db, before_id)
.await
.expect("Should poll events");
let special_events: Vec<_> = events
.iter()
.filter(|e| e.channel == "test_special_chars")
.collect();
assert_eq!(special_events.len(), 1, "Should have event with special characters");
assert_eq!(special_events[0].payload, special_payload, "Special characters should be preserved");
assert_eq!(
special_events.len(),
1,
"Should have event with special characters"
);
assert_eq!(
special_events[0].payload, special_payload,
"Special characters should be preserved"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
@@ -664,7 +818,9 @@ async fn test_cleanup_with_no_old_events(db: Pool<Postgres>) {
assert_eq!(before_count, 2, "Should have 2 recent events");
// Cleanup old events (none of our events should be deleted since they're recent)
let _deleted = cleanup_old_events(&db, 10).await.expect("Should cleanup events");
let _deleted = cleanup_old_events(&db, 10)
.await
.expect("Should cleanup events");
let after_count = count_events_for_channel(&db, &channel).await;
assert_eq!(after_count, 2, "Recent events should not be deleted");
@@ -732,7 +888,11 @@ impl ServerProcess {
}
fn logs_contain(&self, needle: &str) -> bool {
self.log_lines.lock().unwrap().iter().any(|l| l.contains(needle))
self.log_lines
.lock()
.unwrap()
.iter()
.any(|l| l.contains(needle))
}
fn dump_logs(&self) -> String {
@@ -783,12 +943,17 @@ async fn test_two_server_processes_both_receive_event() {
let mut server_b = ServerProcess::start(19200, &db_url);
// Wait for both servers to be ready
let (ready_a, ready_b) = tokio::join!(
wait_for_server(19100, 30),
wait_for_server(19200, 30),
let (ready_a, ready_b) = tokio::join!(wait_for_server(19100, 30), wait_for_server(19200, 30),);
assert!(
ready_a,
"Server A (port 19100) failed to start. Logs:\n{}",
server_a.dump_logs()
);
assert!(
ready_b,
"Server B (port 19200) failed to start. Logs:\n{}",
server_b.dump_logs()
);
assert!(ready_a, "Server A (port 19100) failed to start. Logs:\n{}", server_a.dump_logs());
assert!(ready_b, "Server B (port 19200) failed to start. Logs:\n{}", server_b.dump_logs());
// Give servers a moment to complete their first poll cycle
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
backend/windmill-native-triggers/src/lib.rs
+7 -17
View File
@@ -719,24 +719,14 @@ async fn update_oauth_token_resource(
}
}
/// Look up the full token from the token table using its prefix
/// Look up the full token from the token table using its prefix.
/// With hashed token storage, plaintext tokens are no longer stored in the DB,
/// so this always returns None, forcing callers to create a new token.
pub async fn get_token_by_prefix<'c, E: sqlx::Executor<'c, Database = Postgres>>(
db: E,
token_prefix: &str,
_db: E,
_token_prefix: &str,
) -> Result<Option<String>> {
let token = sqlx::query_scalar!(
r#"
SELECT token
FROM token
WHERE token LIKE concat($1::text, '%')
LIMIT 1
"#,
token_prefix
)
.fetch_optional(db)
.await?;
Ok(token)
Ok(None)
}
/// Delete a token from the token table using its prefix
@@ -747,7 +737,7 @@ pub async fn delete_token_by_prefix<'c, E: sqlx::Executor<'c, Database = Postgre
let deleted = sqlx::query!(
r#"
DELETE FROM token
WHERE token LIKE concat($1::text, '%')
WHERE token_prefix = $1
"#,
token_prefix
)