mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-18 08:03:11 +00:00
Merge remote-tracking branch 'origin/main' into datatable-connect-as-viewer-roles
# Conflicts: # backend/src/monitor.rs
This commit is contained in:
Ruben Fiszel
@@ -18,16 +18,19 @@ jobs:
|
||||
env:
|
||||
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
|
||||
publish_chat_sdk:
|
||||
runs-on: ubicloud-standard-8
|
||||
# Published by npm trusted publishing (OIDC), which npm accepts only from
|
||||
# GitHub-hosted runners and npm >= 11.5.1: keep both, there is no token to fall back on.
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
id-token: write
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-node@v7
|
||||
with:
|
||||
node-version: "20.x"
|
||||
node-version: "24.x"
|
||||
registry-url: "https://registry.npmjs.org"
|
||||
- run: cd chat-sdk && npm ci && npm run build && npm publish --access public
|
||||
env:
|
||||
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
|
||||
publish_cli:
|
||||
runs-on: ubicloud-standard-8
|
||||
steps:
|
||||
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM token WHERE email = $1 AND label = 'session' AND token_hash != $2",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "c3c51c3536eaa6a5e702d306344db63d19f1c9ab0366e01c0092aa653ab51877"
|
||||
}
|
||||
Generated
+1
@@ -16333,6 +16333,7 @@ dependencies = [
|
||||
"git-version",
|
||||
"globset",
|
||||
"hex",
|
||||
"hkdf",
|
||||
"hmac 0.12.1",
|
||||
"hyper 1.12.0",
|
||||
"indexmap 2.14.2",
|
||||
|
||||
@@ -599,6 +599,7 @@ p256 = "0.13"
|
||||
spki = { version = "0.7", features = ["pem"] }
|
||||
pkcs1 = "0.7"
|
||||
aes-gcm = "0.10.3"
|
||||
hkdf = "0.12.4"
|
||||
async_zip = { version = "0.0.17", features = ["tokio", "tokio-fs", "deflate", "chrono"] }
|
||||
once_cell = "1.17.1"
|
||||
dashmap = "6.1.0"
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
DO $$
|
||||
DECLARE
|
||||
too_long BIGINT;
|
||||
BEGIN
|
||||
SELECT count(*) INTO too_long FROM variable WHERE length(value) > 15000;
|
||||
IF too_long > 0 THEN
|
||||
RAISE EXCEPTION 'cannot narrow variable.value to VARCHAR(15000): % row(s) are longer than 15000 characters', too_long;
|
||||
END IF;
|
||||
END $$;
|
||||
|
||||
ALTER TABLE variable DROP CONSTRAINT IF EXISTS variable_value_length;
|
||||
ALTER TABLE variable ALTER COLUMN value TYPE VARCHAR(15000);
|
||||
@@ -0,0 +1,8 @@
|
||||
-- AES-256-GCM ciphertext is 32 bytes (magic, nonce, tag) longer than legacy CBC before base64,
|
||||
-- so the longest CBC value that fits 15000 chars is exactly 15040 chars once re-encrypted.
|
||||
ALTER TABLE variable ALTER COLUMN value TYPE VARCHAR(15040);
|
||||
|
||||
-- Only GCM secrets (base64 of the `wmA2` magic starts `d21BM`) may use the extra room: a longer
|
||||
-- CBC value could never be re-encrypted to fit, and would fail every rotation of its workspace key.
|
||||
ALTER TABLE variable ADD CONSTRAINT variable_value_length
|
||||
CHECK (length(value) <= 15000 OR (is_secret AND value LIKE 'd21BM%'));
|
||||
@@ -4467,6 +4467,24 @@ pub async fn monitor_db(
|
||||
}
|
||||
};
|
||||
|
||||
// Re-encrypt the legacy-format secret variables once new values are written in AES-256-GCM.
|
||||
// Every ~5 min until one pass has gone through every workspace; spawned so a large instance
|
||||
// does not hold up the join below.
|
||||
let legacy_secret_upgrade_f = async {
|
||||
if server_mode
|
||||
&& iteration.is_some()
|
||||
&& iteration.as_ref().unwrap().should_run(30)
|
||||
&& windmill_common::variables::legacy_secret_upgrade_pending()
|
||||
{
|
||||
if let Some(db) = conn.as_sql() {
|
||||
let db = db.clone();
|
||||
tokio::spawn(async move {
|
||||
windmill_common::variables::upgrade_legacy_secret_variables(&db).await
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Delete the AI session backups older than their workspace's retention. Every ~40 min
|
||||
// (240 iterations at the default 10 s, the most a u8 `should_run` counts): the retention
|
||||
// counts in days. Spawned for the same reason as the credential maintenance above, a
|
||||
@@ -4540,6 +4558,7 @@ pub async fn monitor_db(
|
||||
git_credential_maintenance_f,
|
||||
datatable_role_set_sweep_f,
|
||||
ai_session_retention_f,
|
||||
legacy_secret_upgrade_f,
|
||||
pipeline_freshness_watchdog_f,
|
||||
reconcile_unarmed_schedules_f,
|
||||
stranded_jobs_f,
|
||||
|
||||
@@ -470,26 +470,16 @@ async fn update_igroup(
|
||||
require_super_admin(&db, &authed).await?;
|
||||
let mut tx: Transaction<'_, Postgres> = db.begin().await?;
|
||||
|
||||
let exists_opt = sqlx::query("SELECT 1 FROM instance_group WHERE name = $1")
|
||||
.bind(name.clone())
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
not_found_if_none(exists_opt, "instance_group", name.clone())?;
|
||||
let old_role_opt = sqlx::query_scalar!(
|
||||
"SELECT instance_role FROM instance_group WHERE name = $1 FOR UPDATE",
|
||||
&name
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let old_role = not_found_if_none(old_role_opt, "instance_group", name.clone())?;
|
||||
|
||||
let validated_role = validate_instance_role(&igroup_update.instance_role)?;
|
||||
|
||||
// Fetch old role before updating so we can detect changes
|
||||
let old_role = if igroup_update.instance_role.is_some() {
|
||||
sqlx::query_scalar!(
|
||||
"SELECT instance_role FROM instance_group WHERE name = $1",
|
||||
&name
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
sqlx::query("UPDATE instance_group SET summary = $1, instance_role = $2 WHERE name = $3")
|
||||
.bind(igroup_update.new_summary)
|
||||
.bind(&validated_role)
|
||||
@@ -497,8 +487,9 @@ async fn update_igroup(
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// If instance_role actually changed, propagate to all group members
|
||||
if igroup_update.instance_role.is_some() && old_role != validated_role {
|
||||
// An omitted or null instance_role is stored as NULL above, so it revokes like any other
|
||||
// change: members must be recomputed whenever the stored role differs.
|
||||
if old_role != validated_role {
|
||||
propagate_instance_group_roles(&name, &mut tx).await?;
|
||||
}
|
||||
|
||||
@@ -664,21 +655,19 @@ async fn delete_igroup(
|
||||
|
||||
// FOR UPDATE: the group row is the group-level mutex, taken before the workspace
|
||||
// advisory locks (see reconcile_workspace_instance_groups).
|
||||
let group_role = sqlx::query_scalar!(
|
||||
sqlx::query_scalar!(
|
||||
"SELECT instance_role FROM instance_group WHERE name = $1 FOR UPDATE",
|
||||
&name
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten();
|
||||
.await?;
|
||||
|
||||
let affected_members: Vec<String> = if group_role.is_some() {
|
||||
// Collected whatever the group's current role: a member can still hold a role this group
|
||||
// granted before it was cleared.
|
||||
let affected_members: Vec<String> =
|
||||
sqlx::query_scalar!("SELECT email FROM email_to_igroup WHERE igroup = $1", &name)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?
|
||||
} else {
|
||||
vec![]
|
||||
};
|
||||
.await?;
|
||||
|
||||
// Captured and advisory-locked before the settings update strips the group from them.
|
||||
#[cfg(feature = "private")]
|
||||
|
||||
@@ -497,6 +497,72 @@ async fn test_remove_user_from_instance_group_rederives_role(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Clearing an instance group's role, or deleting a group whose role was cleared out from under
|
||||
/// its members, must take the instance-level role back from them.
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
||||
async fn test_instance_group_role_revoked_on_clear_and_delete(
|
||||
db: Pool<Postgres>,
|
||||
) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let port = server.addr.port();
|
||||
let global_base = format!("http://localhost:{port}/api/groups");
|
||||
let email = "test2@windmill.dev";
|
||||
|
||||
let is_super_admin = || async {
|
||||
sqlx::query_scalar::<_, bool>("SELECT super_admin FROM password WHERE email = $1")
|
||||
.bind(email)
|
||||
.fetch_one(&db)
|
||||
.await
|
||||
};
|
||||
|
||||
let resp = authed(client().post(format!("{global_base}/create")))
|
||||
.json(&json!({ "name": "admins_grp" }))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 200, "create: {}", resp.text().await?);
|
||||
let resp = authed(client().post(format!("{global_base}/update/admins_grp")))
|
||||
.json(&json!({ "new_summary": "", "instance_role": "superadmin" }))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 200, "set role: {}", resp.text().await?);
|
||||
let resp = authed(client().post(format!("{global_base}/adduser/admins_grp")))
|
||||
.json(&json!({ "email": email }))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 200, "adduser: {}", resp.text().await?);
|
||||
assert!(is_super_admin().await?, "group should grant superadmin");
|
||||
|
||||
let resp = authed(client().post(format!("{global_base}/update/admins_grp")))
|
||||
.json(&json!({ "new_summary": "", "instance_role": null }))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 200, "update: {}", resp.text().await?);
|
||||
assert!(
|
||||
!is_super_admin().await?,
|
||||
"clearing the group's role must revoke superadmin"
|
||||
);
|
||||
|
||||
// A member left elevated while the group's stored role is already NULL.
|
||||
sqlx::query(
|
||||
"UPDATE password SET super_admin = true, role_source = 'instance_group' WHERE email = $1",
|
||||
)
|
||||
.bind(email)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
|
||||
let resp = authed(client().delete(format!("{global_base}/delete/admins_grp")))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 200, "delete: {}", resp.text().await?);
|
||||
assert!(
|
||||
!is_super_admin().await?,
|
||||
"deleting a role-cleared group must revoke superadmin"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// An overwrite import that moves a member from a dropped group to a retained one must keep
|
||||
/// their workspace data.
|
||||
///
|
||||
|
||||
@@ -25,11 +25,11 @@ use axum::{
|
||||
};
|
||||
use chrono::{DateTime, Utc};
|
||||
use lazy_static::lazy_static;
|
||||
use magic_crypt::MagicCrypt256;
|
||||
use regex::Regex;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use windmill_api_auth::{is_effectively_unscoped, ApiAuthed, Tokened};
|
||||
use windmill_audit::{audit_oss::audit_log, ActionKind};
|
||||
use windmill_common::variables::WorkspaceCrypt;
|
||||
use windmill_common::{
|
||||
auth::hash_token,
|
||||
error::{error_source_chain, Error, Result},
|
||||
@@ -676,8 +676,8 @@ async fn proxy(
|
||||
pub(crate) async fn reencrypt_tokens(
|
||||
conn: &mut sqlx::PgConnection,
|
||||
w_id: &str,
|
||||
old: &MagicCrypt256,
|
||||
new: &MagicCrypt256,
|
||||
old: &WorkspaceCrypt,
|
||||
new: &WorkspaceCrypt,
|
||||
) -> Result<()> {
|
||||
let rows = sqlx::query!(
|
||||
"SELECT email, token FROM remote_deploy_token
|
||||
|
||||
@@ -6317,8 +6317,8 @@ async fn set_encryption_key(
|
||||
async fn reencrypt_git_sync_secrets(
|
||||
conn: &mut sqlx::PgConnection,
|
||||
w_id: &str,
|
||||
old: &magic_crypt::MagicCrypt256,
|
||||
new: &magic_crypt::MagicCrypt256,
|
||||
old: &windmill_common::variables::WorkspaceCrypt,
|
||||
new: &windmill_common::variables::WorkspaceCrypt,
|
||||
) -> Result<()> {
|
||||
let Some((mut credentials, mut git_sync)) =
|
||||
sqlx::query_as::<_, (serde_json::Value, Option<serde_json::Value>)>(
|
||||
|
||||
@@ -24,7 +24,7 @@ use axum::{
|
||||
Extension, Json, Router,
|
||||
};
|
||||
use futures::{StreamExt, TryStreamExt};
|
||||
use magic_crypt::{MagicCrypt256, MagicCryptTrait};
|
||||
use magic_crypt::MagicCryptTrait;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::value::RawValue;
|
||||
use std::sync::Arc;
|
||||
@@ -35,7 +35,7 @@ use windmill_api_workspaces::ai_session_backups::{
|
||||
use windmill_api_workspaces::workspaces::sessions_retention_days;
|
||||
use windmill_common::error::{Error, JsonResult, Result};
|
||||
use windmill_common::utils::calculate_hash;
|
||||
use windmill_common::variables::{crypt_from_key_with_suffix, get_workspace_key};
|
||||
use windmill_common::variables::{crypt_from_key_with_suffix, get_workspace_key, WorkspaceCrypt};
|
||||
use windmill_object_store::object_store_reexports::{
|
||||
ObjectStore, ObjectStoreError, Path as ObjectPath, PutPayload,
|
||||
};
|
||||
@@ -135,7 +135,7 @@ enum Read {
|
||||
/// The user's prefix in the workspace storage, plus what reads and writes it.
|
||||
struct Backend {
|
||||
store: Arc<dyn ObjectStore>,
|
||||
mc: MagicCrypt256,
|
||||
mc: WorkspaceCrypt,
|
||||
prefix: String,
|
||||
/// Name the storage and the generation the objects are under, for the browser's sync
|
||||
/// state: a row recorded against another storage or generation is stale, a removal is
|
||||
|
||||
@@ -198,6 +198,7 @@ async fn set_password(
|
||||
Extension(db): Extension<DB>,
|
||||
Extension(argon2): Extension<Arc<Argon2<'_>>>,
|
||||
authed: ApiAuthed,
|
||||
Tokened { token }: Tokened,
|
||||
OptJobAuthed { job_id, .. }: OptJobAuthed,
|
||||
Json(ep): Json<EditPassword>,
|
||||
) -> Result<String> {
|
||||
@@ -206,7 +207,15 @@ async fn set_password(
|
||||
// (GHSA-hfh4-cx4h-3fcr).
|
||||
forbid_elevated_job_token(&db, &authed.email, job_id).await?;
|
||||
let email = authed.email.clone();
|
||||
crate::users_oss::set_password(db, argon2, authed, &email, ep).await
|
||||
let res = crate::users_oss::set_password(db.clone(), argon2, authed, &email, ep).await?;
|
||||
sqlx::query!(
|
||||
"DELETE FROM token WHERE email = $1 AND label = 'session' AND token_hash != $2",
|
||||
&email,
|
||||
windmill_common::auth::hash_token(&token)
|
||||
)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
Ok(res)
|
||||
}
|
||||
|
||||
async fn set_password_of_user(
|
||||
@@ -219,7 +228,14 @@ async fn set_password_of_user(
|
||||
) -> Result<String> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
forbid_superadmin_job_token(&db, &authed.email, job_id).await?;
|
||||
crate::users_oss::set_password(db, argon2, authed, &email, ep).await
|
||||
let res = crate::users_oss::set_password(db.clone(), argon2, authed, &email, ep).await?;
|
||||
sqlx::query!(
|
||||
"DELETE FROM token WHERE email = $1 AND label = 'session'",
|
||||
&email
|
||||
)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
Ok(res)
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
|
||||
@@ -8,7 +8,7 @@ edition.workspace = true
|
||||
default = []
|
||||
enterprise = ["dep:aws-config"]
|
||||
instance_config_schema = ["dep:schemars"]
|
||||
local_reports = ["dep:rsa", "dep:aes-gcm"]
|
||||
local_reports = ["dep:rsa"]
|
||||
private = ["dep:aws-sdk-rds", "dep:aws-sdk-secretsmanager", "dep:aws-config"]
|
||||
jemalloc = ["dep:tikv-jemalloc-ctl"]
|
||||
tantivy = []
|
||||
@@ -119,7 +119,8 @@ rsa = { workspace = true, optional = true }
|
||||
p256 = { workspace = true, optional = true }
|
||||
spki = { workspace = true }
|
||||
pkcs1 = { workspace = true }
|
||||
aes-gcm = { workspace = true, optional = true }
|
||||
aes-gcm.workspace = true
|
||||
hkdf.workspace = true
|
||||
|
||||
semver.workspace = true
|
||||
croner.workspace = true
|
||||
|
||||
@@ -157,6 +157,7 @@ pub mod wac;
|
||||
pub mod webhook;
|
||||
pub mod worker;
|
||||
pub mod worker_group_job_stats;
|
||||
pub mod workspace_crypt;
|
||||
pub mod workspaces;
|
||||
|
||||
pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
|
||||
|
||||
@@ -23,6 +23,11 @@ pub const MIN_VERSION_SUPPORTS_BUN_LOCKFILE_V2: VC = vc(1, 794, 0, "Bun v2 lockf
|
||||
// them, so a restriction set while one is live is silently ignored on the jobs it runs. Must
|
||||
// name the release this ships in.
|
||||
pub const MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES: VC = vc(1, 822, 0, "Restricted job tokens");
|
||||
// Gates writing AES-256-GCM workspace ciphertext and the background pass that rewrites the
|
||||
// stored secret variables. Reading it ships one release earlier (1.831), so a rollback to
|
||||
// that release still reads every secret; below it, secrets written after the switch stay
|
||||
// unreadable until rolled forward. Only worker versions are seen here, not servers.
|
||||
pub const MIN_VERSION_SUPPORTS_AEAD_SECRETS: VC = vc(1, 832, 0, "AES-GCM secret encryption");
|
||||
pub const MIN_VERSION_SUPPORTS_NODE_DEBOUNCING: VC = vc(1, 658, 0, "Flow node debouncing");
|
||||
pub const MIN_VERSION_SUPPORTS_TOKEN_HASH: VC = vc(1, 659, 0, "Token hash storage");
|
||||
pub const MIN_VERSION_SUPPORTS_SYNC_JOBS_DEBOUNCING: VC = vc(1, 602, 0, "Sync jobs debouncing");
|
||||
|
||||
@@ -281,6 +281,9 @@ pub fn sanitize_db_error(e: sqlx::Error) -> Error {
|
||||
return Error::from(e);
|
||||
};
|
||||
match db_err.code().as_deref() {
|
||||
Some("23514") if db_err.constraint() == Some("variable_value_length") => {
|
||||
Error::BadRequest("A field exceeds its maximum length".to_string())
|
||||
}
|
||||
// check_violation — `proper_id` / `proper_name` and friends
|
||||
Some("23514") => Error::BadRequest(
|
||||
"Invalid path or name: it does not match the required format".to_string(),
|
||||
|
||||
@@ -12,11 +12,13 @@ use crate::scripts::ScriptHash;
|
||||
use crate::secret_backend::{get_secret_value, is_external_stored_value};
|
||||
use crate::utils::WarnAfterExt;
|
||||
use crate::worker::Connection;
|
||||
use crate::workspace_crypt::CipherFormat;
|
||||
pub use crate::workspace_crypt::WorkspaceCrypt;
|
||||
use crate::{worker::WORKER_GROUP, BASE_URL, DB};
|
||||
use chrono::{SecondsFormat, Utc};
|
||||
use magic_crypt::{MagicCrypt256, MagicCryptTrait};
|
||||
use quick_cache::sync::Cache;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
pub static ref SECRET_SALT: Option<String> = std::env::var("SECRET_SALT").ok();
|
||||
@@ -211,7 +213,7 @@ pub struct CreateVariable {
|
||||
pub ws_specific: Option<bool>,
|
||||
}
|
||||
|
||||
pub async fn build_crypt(db: &DB, w_id: &str) -> crate::error::Result<MagicCrypt256> {
|
||||
pub async fn build_crypt(db: &DB, w_id: &str) -> crate::error::Result<WorkspaceCrypt> {
|
||||
// Check cache first (300-second staleness)
|
||||
let cached_key_o = WORKSPACE_CRYPT_CACHE.get(w_id).and_then(|(ts, key)| {
|
||||
if ts > chrono::Utc::now().timestamp() - 300 {
|
||||
@@ -235,7 +237,7 @@ pub async fn build_crypt(db: &DB, w_id: &str) -> crate::error::Result<MagicCrypt
|
||||
} else {
|
||||
key
|
||||
};
|
||||
let ncrypt = magic_crypt::new_magic_crypt!(crypt_key, 256);
|
||||
let ncrypt = WorkspaceCrypt::new(&crypt_key);
|
||||
WORKSPACE_CRYPT_CACHE.insert(
|
||||
w_id.to_string(),
|
||||
(chrono::Utc::now().timestamp(), ncrypt.clone()),
|
||||
@@ -250,19 +252,19 @@ pub async fn build_crypt_with_key_suffix(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
key_suffix: &str,
|
||||
) -> crate::error::Result<MagicCrypt256> {
|
||||
) -> crate::error::Result<WorkspaceCrypt> {
|
||||
let key = get_workspace_key(w_id, db).await?;
|
||||
Ok(crypt_from_key_with_suffix(&key, key_suffix))
|
||||
}
|
||||
|
||||
/// The cipher `build_crypt_with_key_suffix` builds, from a key string in hand.
|
||||
pub fn crypt_from_key_with_suffix(key: &str, key_suffix: &str) -> MagicCrypt256 {
|
||||
pub fn crypt_from_key_with_suffix(key: &str, key_suffix: &str) -> WorkspaceCrypt {
|
||||
let crypt_key = if let Some(ref salt) = SECRET_SALT.as_ref() {
|
||||
format!("{}{}{}", key, salt, key_suffix)
|
||||
} else {
|
||||
format!("{}{}", key, key_suffix)
|
||||
};
|
||||
magic_crypt::new_magic_crypt!(crypt_key, 256)
|
||||
WorkspaceCrypt::new(&crypt_key)
|
||||
}
|
||||
|
||||
pub async fn get_workspace_key(w_id: &str, db: &DB) -> crate::error::Result<String> {
|
||||
@@ -371,27 +373,178 @@ pub async fn get_secret_value_as_admin(
|
||||
Ok(r)
|
||||
}
|
||||
|
||||
pub fn encrypt(mc: &MagicCrypt256, value: &str) -> String {
|
||||
mc.encrypt_str_to_base64(value)
|
||||
pub fn encrypt(mc: &WorkspaceCrypt, value: &str) -> String {
|
||||
mc.encrypt_str(value)
|
||||
}
|
||||
|
||||
const DECRYPT_FAILED: &str =
|
||||
"Could not decrypt value. The value may have been encrypted with a different key.";
|
||||
|
||||
/// Every failure returns the same message so responses carry no signal about why decryption
|
||||
/// failed (padding, base64, UTF-8); the cause is only logged.
|
||||
pub fn decrypt(mc: &MagicCrypt256, value: String) -> error::Result<String> {
|
||||
mc.decrypt_base64_to_string(value).map_err(|e| {
|
||||
/// failed (authentication, padding, base64, UTF-8); the kind of failure is only logged.
|
||||
pub fn decrypt(mc: &WorkspaceCrypt, value: String) -> error::Result<String> {
|
||||
mc.decrypt_str(&value).map_err(|e| {
|
||||
tracing::warn!("decryption failed: {e}");
|
||||
error::Error::internal_err(DECRYPT_FAILED.to_string())
|
||||
})
|
||||
}
|
||||
|
||||
static LEGACY_SECRET_UPGRADE_DONE: AtomicBool = AtomicBool::new(false);
|
||||
static LEGACY_SECRET_UPGRADE_RUNNING: AtomicBool = AtomicBool::new(false);
|
||||
const LEGACY_SECRET_UPGRADE_BATCH: i64 = 200;
|
||||
|
||||
/// Whether this process still has to run [`upgrade_legacy_secret_variables`]: once new values
|
||||
/// are written in v2, until one pass has gone through every workspace.
|
||||
pub fn legacy_secret_upgrade_pending() -> bool {
|
||||
CipherFormat::current() == CipherFormat::V2
|
||||
&& !LEGACY_SECRET_UPGRADE_DONE.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
/// Re-encrypt the secret variables still in the legacy format in v2, workspace by workspace.
|
||||
/// Idempotent, and emits no git-sync event: the plaintexts do not change.
|
||||
pub async fn upgrade_legacy_secret_variables(db: &DB) {
|
||||
struct Running;
|
||||
impl Drop for Running {
|
||||
fn drop(&mut self) {
|
||||
LEGACY_SECRET_UPGRADE_RUNNING.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
if !legacy_secret_upgrade_pending()
|
||||
|| LEGACY_SECRET_UPGRADE_RUNNING.swap(true, Ordering::AcqRel)
|
||||
{
|
||||
return;
|
||||
}
|
||||
// Cleared on drop, so a pass that panics does not stop every later one in this process.
|
||||
let _running = Running;
|
||||
upgrade_legacy_secret_variables_pass(db).await;
|
||||
}
|
||||
|
||||
async fn upgrade_legacy_secret_variables_pass(db: &DB) {
|
||||
let workspaces: Vec<String> = match sqlx::query_scalar("SELECT id FROM workspace ORDER BY id")
|
||||
.fetch_all(db)
|
||||
.await
|
||||
{
|
||||
Ok(w) => w,
|
||||
Err(e) => {
|
||||
tracing::error!("secret encryption upgrade: listing workspaces failed: {e:#}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
tracing::info!(
|
||||
"secret encryption upgrade: re-encrypting legacy secret variables of {} workspaces in AES-256-GCM",
|
||||
workspaces.len()
|
||||
);
|
||||
let (mut upgraded, mut skipped, mut failed_workspaces) = (0, 0, 0);
|
||||
for w_id in workspaces {
|
||||
match upgrade_workspace_secret_variables(db, &w_id).await {
|
||||
Ok((u, s)) => {
|
||||
if u > 0 || s > 0 {
|
||||
tracing::info!(
|
||||
"secret encryption upgrade: workspace {w_id}: {u} re-encrypted, {s} skipped"
|
||||
);
|
||||
}
|
||||
upgraded += u;
|
||||
skipped += s;
|
||||
}
|
||||
Err(e) => {
|
||||
failed_workspaces += 1;
|
||||
tracing::error!("secret encryption upgrade: workspace {w_id} failed: {e:#}");
|
||||
}
|
||||
}
|
||||
}
|
||||
tracing::info!(
|
||||
"secret encryption upgrade: done, {upgraded} re-encrypted, {skipped} skipped, {failed_workspaces} workspaces failed"
|
||||
);
|
||||
if failed_workspaces == 0 {
|
||||
LEGACY_SECRET_UPGRADE_DONE.store(true, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// The number of values re-encrypted and of values skipped as they do not decrypt.
|
||||
async fn upgrade_workspace_secret_variables(db: &DB, w_id: &str) -> error::Result<(usize, usize)> {
|
||||
let (mut upgraded, mut skipped) = (0, 0);
|
||||
let mut after = String::new();
|
||||
loop {
|
||||
let mut tx = db.begin().await?;
|
||||
// Shared, so a rotation (which locks the key row for update and rewrites every secret)
|
||||
// never interleaves with a batch, and the key read here is the one the values are under.
|
||||
let key: Option<String> = sqlx::query_scalar(
|
||||
"SELECT key FROM workspace_key WHERE workspace_id = $1 AND kind = 'cloud' FOR SHARE",
|
||||
)
|
||||
.bind(w_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some(key) = key else {
|
||||
return Ok((upgraded, skipped));
|
||||
};
|
||||
let mc = crypt_from_key_with_suffix(&key, "");
|
||||
let rows: Vec<(String, String)> = sqlx::query_as(
|
||||
"SELECT path, value FROM variable WHERE workspace_id = $1 AND is_secret AND path > $2 \
|
||||
ORDER BY path LIMIT $3",
|
||||
)
|
||||
.bind(w_id)
|
||||
.bind(&after)
|
||||
.bind(LEGACY_SECRET_UPGRADE_BATCH)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
let last_page = (rows.len() as i64) < LEGACY_SECRET_UPGRADE_BATCH;
|
||||
for (path, value) in rows {
|
||||
if value.is_empty() || is_external_stored_value(&value) || mc.is_v2(&value) {
|
||||
after = path;
|
||||
continue;
|
||||
}
|
||||
match mc.upgrade(&value) {
|
||||
Some(new_value) => {
|
||||
// In a savepoint, so a value the update rejects (v2 is up to 44 chars longer
|
||||
// than CBC, which a column narrower than 15040 cannot hold) is skipped
|
||||
// without losing the batch.
|
||||
let mut sp = sqlx::Acquire::begin(&mut tx).await?;
|
||||
// Only over the value it was computed from, so a write racing it wins.
|
||||
match sqlx::query(
|
||||
"UPDATE variable SET value = $1 \
|
||||
WHERE workspace_id = $2 AND path = $3 AND value = $4",
|
||||
)
|
||||
.bind(new_value)
|
||||
.bind(w_id)
|
||||
.bind(&path)
|
||||
.bind(&value)
|
||||
.execute(&mut *sp)
|
||||
.await
|
||||
{
|
||||
Ok(r) => {
|
||||
sp.commit().await?;
|
||||
upgraded += r.rows_affected() as usize;
|
||||
}
|
||||
Err(e) => {
|
||||
sp.rollback().await?;
|
||||
tracing::warn!(
|
||||
"secret encryption upgrade: secret variable {path} of workspace {w_id} could not be stored re-encrypted, leaving it as is: {e}"
|
||||
);
|
||||
skipped += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
tracing::warn!(
|
||||
"secret encryption upgrade: secret variable {path} of workspace {w_id} does not decrypt, leaving it as is"
|
||||
);
|
||||
skipped += 1;
|
||||
}
|
||||
}
|
||||
after = path;
|
||||
}
|
||||
tx.commit().await?;
|
||||
if last_page {
|
||||
return Ok((upgraded, skipped));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub const WM_SCHEDULED_FOR: &str = "WM_SCHEDULED_FOR";
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
pub static ref CUSTOM_ENVS_CACHE: Cache<String, (i64, Vec<(String, String)>)> = Cache::new(100);
|
||||
pub static ref WORKSPACE_CRYPT_CACHE: Cache<String, (i64, MagicCrypt256)> = Cache::new(1000);
|
||||
pub static ref WORKSPACE_CRYPT_CACHE: Cache<String, (i64, WorkspaceCrypt)> = Cache::new(1000);
|
||||
|
||||
}
|
||||
|
||||
@@ -753,16 +906,24 @@ mod tests {
|
||||
DECRYPT_FAILED,
|
||||
};
|
||||
use crate::error::Error;
|
||||
use crate::workspace_crypt::WorkspaceCrypt;
|
||||
use base64::Engine;
|
||||
use magic_crypt::MagicCryptTrait;
|
||||
|
||||
#[test]
|
||||
fn decrypt_failures_all_return_the_same_message() {
|
||||
let mc = magic_crypt::new_magic_crypt!("workspace-key", 256);
|
||||
let other = magic_crypt::new_magic_crypt!("another-key", 256);
|
||||
let mc = WorkspaceCrypt::new("workspace-key");
|
||||
let legacy = magic_crypt::new_magic_crypt!("workspace-key", 256);
|
||||
let other = WorkspaceCrypt::new("another-key");
|
||||
let b64 = base64::engine::general_purpose::STANDARD;
|
||||
let v2 = mc.upgrade(&legacy.encrypt_str_to_base64("secret")).unwrap();
|
||||
let mut tampered = b64.decode(v2).unwrap();
|
||||
*tampered.last_mut().unwrap() ^= 1;
|
||||
let failures = [
|
||||
"not base64 at all!".to_string(),
|
||||
mc.encrypt_bytes_to_base64(&[0xff, 0xfe, 0xfd]),
|
||||
legacy.encrypt_bytes_to_base64(&[0xff, 0xfe, 0xfd]),
|
||||
other.encrypt_str_to_base64("secret"),
|
||||
b64.encode(&tampered),
|
||||
];
|
||||
for value in failures {
|
||||
match decrypt(&mc, value.clone()) {
|
||||
|
||||
@@ -0,0 +1,255 @@
|
||||
/*
|
||||
* Author: Ruben Fiszel
|
||||
* Copyright: Windmill Labs, Inc 2026
|
||||
* This file and its contents are licensed under the AGPLv3 License.
|
||||
* Please see the included NOTICE for copyright information and
|
||||
* LICENSE-AGPL for a copy of the license.
|
||||
*/
|
||||
|
||||
//! The cipher behind `variables::encrypt`/`decrypt`.
|
||||
//!
|
||||
//! Values are written as AES-256-GCM ("v2"), standard base64 of
|
||||
//! `MAGIC (4) || nonce (12) || AES-256-GCM(PKCS#7(plaintext)) || tag (16)`.
|
||||
//! Values written before are AES-256-CBC (`magic_crypt`) and are still read.
|
||||
//!
|
||||
//! The plaintext is padded to the block size so a v2 value decodes to a multiple of 16 bytes,
|
||||
//! like a CBC one: the CLI and `validate_already_encrypted_secret` tell ciphertext from
|
||||
//! plaintext by exactly that, and an old CLI that took a v2 value for plaintext would push it
|
||||
//! to be encrypted again, corrupting the secret.
|
||||
|
||||
use aes_gcm::aead::{Aead, AeadCore, KeyInit, OsRng};
|
||||
use aes_gcm::{Aes256Gcm, Key, Nonce};
|
||||
use base64::Engine;
|
||||
use hkdf::Hkdf;
|
||||
use magic_crypt::{MagicCrypt256, MagicCryptTrait};
|
||||
use sha2::Sha256;
|
||||
|
||||
use crate::min_version::MIN_VERSION_SUPPORTS_AEAD_SECRETS;
|
||||
|
||||
pub const V2_MAGIC: [u8; 4] = *b"wmA2";
|
||||
const NONCE_LEN: usize = 12;
|
||||
const TAG_LEN: usize = 16;
|
||||
const BLOCK: usize = 16;
|
||||
|
||||
const AEAD_KEY_INFO: &[u8] = b"windmill workspace secret aes-256-gcm v1";
|
||||
|
||||
const B64: base64::engine::GeneralPurpose = base64::engine::general_purpose::STANDARD;
|
||||
|
||||
/// Why a value did not decrypt. Only ever logged, never returned to a caller.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum DecryptFailure {
|
||||
Encoding,
|
||||
Authentication,
|
||||
Utf8,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for DecryptFailure {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.write_str(match self {
|
||||
DecryptFailure::Encoding => "not base64",
|
||||
DecryptFailure::Authentication => "authentication or padding check failed",
|
||||
DecryptFailure::Utf8 => "plaintext is not UTF-8",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum CipherFormat {
|
||||
LegacyCbc,
|
||||
V2,
|
||||
}
|
||||
|
||||
impl CipherFormat {
|
||||
/// What new values are written as. Old workers read secrets straight from the database with
|
||||
/// the CBC cipher alone, so v2 is written only once every live worker reads it.
|
||||
pub fn current() -> Self {
|
||||
if MIN_VERSION_SUPPORTS_AEAD_SECRETS.met_conservatively() {
|
||||
CipherFormat::V2
|
||||
} else {
|
||||
CipherFormat::LegacyCbc
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct WorkspaceCrypt {
|
||||
legacy: MagicCrypt256,
|
||||
aead: Aes256Gcm,
|
||||
}
|
||||
|
||||
/// Code that drives `magic_crypt` directly on this cipher (app link secrets, AI session
|
||||
/// objects) keeps the legacy format it always had.
|
||||
impl std::ops::Deref for WorkspaceCrypt {
|
||||
type Target = MagicCrypt256;
|
||||
fn deref(&self) -> &MagicCrypt256 {
|
||||
&self.legacy
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkspaceCrypt {
|
||||
/// `crypt_key` is the key material the CBC cipher was always built from (workspace key, salt,
|
||||
/// suffix). The AEAD key is derived from it under its own label, so the two ciphers never
|
||||
/// share a key.
|
||||
pub fn new(crypt_key: &str) -> Self {
|
||||
let mut aead_key = [0u8; 32];
|
||||
Hkdf::<Sha256>::new(None, crypt_key.as_bytes())
|
||||
.expand(AEAD_KEY_INFO, &mut aead_key)
|
||||
.expect("32 bytes is a valid HKDF-SHA256 output length");
|
||||
Self {
|
||||
legacy: magic_crypt::new_magic_crypt!(crypt_key, 256),
|
||||
aead: Aes256Gcm::new(Key::<Aes256Gcm>::from_slice(&aead_key)),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn encrypt_str(&self, plaintext: &str) -> String {
|
||||
self.encrypt_str_in(plaintext, CipherFormat::current())
|
||||
}
|
||||
|
||||
pub fn encrypt_str_in(&self, plaintext: &str, format: CipherFormat) -> String {
|
||||
match format {
|
||||
CipherFormat::V2 => B64.encode(self.seal(plaintext.as_bytes())),
|
||||
CipherFormat::LegacyCbc => self.legacy.encrypt_str_to_base64(plaintext),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn decrypt_str(&self, ciphertext: &str) -> Result<String, DecryptFailure> {
|
||||
let bytes = B64
|
||||
.decode(ciphertext)
|
||||
.map_err(|_| DecryptFailure::Encoding)?;
|
||||
String::from_utf8(self.decrypt_bytes(&bytes)?).map_err(|_| DecryptFailure::Utf8)
|
||||
}
|
||||
|
||||
/// Whether `ciphertext` is a v2 value under this key.
|
||||
pub fn is_v2(&self, ciphertext: &str) -> bool {
|
||||
B64.decode(ciphertext)
|
||||
.is_ok_and(|bytes| self.open(&bytes).is_some())
|
||||
}
|
||||
|
||||
/// The v2 form of a legacy `ciphertext`, checked to decrypt back to the same plaintext.
|
||||
/// `None` for a v2 value and for anything that does not decrypt under this key.
|
||||
pub fn upgrade(&self, ciphertext: &str) -> Option<String> {
|
||||
if self.is_v2(ciphertext) {
|
||||
return None;
|
||||
}
|
||||
let plaintext = self.decrypt_str(ciphertext).ok()?;
|
||||
let upgraded = self.encrypt_str_in(&plaintext, CipherFormat::V2);
|
||||
(self.decrypt_str(&upgraded).ok()? == plaintext).then_some(upgraded)
|
||||
}
|
||||
|
||||
fn decrypt_bytes(&self, bytes: &[u8]) -> Result<Vec<u8>, DecryptFailure> {
|
||||
match self.open(bytes) {
|
||||
Some(plaintext) => Ok(plaintext),
|
||||
// Also for a value with the magic: a CBC ciphertext starts with it one time in 2^32.
|
||||
None => self
|
||||
.legacy
|
||||
.decrypt_bytes_to_bytes(bytes)
|
||||
.map_err(|_| DecryptFailure::Authentication),
|
||||
}
|
||||
}
|
||||
|
||||
fn seal(&self, plaintext: &[u8]) -> Vec<u8> {
|
||||
let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
|
||||
let pad = BLOCK - plaintext.len() % BLOCK;
|
||||
let mut padded = Vec::with_capacity(plaintext.len() + pad);
|
||||
padded.extend_from_slice(plaintext);
|
||||
padded.resize(plaintext.len() + pad, pad as u8);
|
||||
let sealed = self
|
||||
.aead
|
||||
.encrypt(&nonce, padded.as_slice())
|
||||
.expect("AES-GCM encryption fails only past 64 GiB of plaintext");
|
||||
let mut out = Vec::with_capacity(V2_MAGIC.len() + NONCE_LEN + sealed.len());
|
||||
out.extend_from_slice(&V2_MAGIC);
|
||||
out.extend_from_slice(&nonce);
|
||||
out.extend_from_slice(&sealed);
|
||||
out
|
||||
}
|
||||
|
||||
fn open(&self, data: &[u8]) -> Option<Vec<u8>> {
|
||||
let body = data.strip_prefix(&V2_MAGIC)?;
|
||||
if body.len() < NONCE_LEN + BLOCK + TAG_LEN || (body.len() - NONCE_LEN) % BLOCK != 0 {
|
||||
return None;
|
||||
}
|
||||
let (nonce, sealed) = body.split_at(NONCE_LEN);
|
||||
let mut padded = self.aead.decrypt(Nonce::from_slice(nonce), sealed).ok()?;
|
||||
let pad = *padded.last()? as usize;
|
||||
if pad == 0
|
||||
|| pad > BLOCK
|
||||
|| padded[padded.len() - pad..]
|
||||
.iter()
|
||||
.any(|b| *b as usize != pad)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
padded.truncate(padded.len() - pad);
|
||||
Some(padded)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const KEY: &str = "workspace-key";
|
||||
|
||||
#[test]
|
||||
fn v2_round_trips() {
|
||||
let wc = WorkspaceCrypt::new(KEY);
|
||||
for plain in [
|
||||
"",
|
||||
"s",
|
||||
"exactly sixteen!",
|
||||
"a longer secret value with ünïcödé",
|
||||
] {
|
||||
let a = wc.encrypt_str_in(plain, CipherFormat::V2);
|
||||
assert_ne!(a, wc.encrypt_str_in(plain, CipherFormat::V2), "nonce reuse");
|
||||
assert!(wc.is_v2(&a));
|
||||
assert_eq!(wc.decrypt_str(&a).unwrap(), plain);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_cbc_still_decrypts() {
|
||||
let wc = WorkspaceCrypt::new(KEY);
|
||||
let legacy = magic_crypt::new_magic_crypt!(KEY, 256).encrypt_str_to_base64("old secret");
|
||||
assert!(!wc.is_v2(&legacy));
|
||||
assert_eq!(wc.decrypt_str(&legacy).unwrap(), "old secret");
|
||||
let upgraded = wc.upgrade(&legacy).unwrap();
|
||||
assert!(wc.is_v2(&upgraded));
|
||||
assert_eq!(wc.decrypt_str(&upgraded).unwrap(), "old secret");
|
||||
assert_eq!(wc.upgrade(&upgraded), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn magic_prefixed_cbc_still_decrypts() {
|
||||
// Too rare to find by search, so built: with magic_crypt's zero IV and SHA-256(key) AES
|
||||
// key, the first CBC block is AES(P1), so P1 = AES^-1(MAGIC || zeros).
|
||||
let mut block = [0u8; 16];
|
||||
block[..4].copy_from_slice(&V2_MAGIC);
|
||||
let mut plaintext = {
|
||||
use aes_gcm::aes::cipher::{BlockDecrypt, KeyInit};
|
||||
use sha2::Digest;
|
||||
let cipher = aes_gcm::aes::Aes256::new_from_slice(&Sha256::digest(KEY)).unwrap();
|
||||
let mut b = aes_gcm::aes::Block::clone_from_slice(&block);
|
||||
cipher.decrypt_block(&mut b);
|
||||
b.to_vec()
|
||||
};
|
||||
plaintext.extend_from_slice(b"tail of the secret value");
|
||||
let ciphertext = magic_crypt::new_magic_crypt!(KEY, 256).encrypt_bytes_to_bytes(&plaintext);
|
||||
assert!(ciphertext.starts_with(&V2_MAGIC));
|
||||
assert_eq!(
|
||||
WorkspaceCrypt::new(KEY).decrypt_bytes(&ciphertext).unwrap(),
|
||||
plaintext
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v2_passes_the_block_multiple_ciphertext_shape_check() {
|
||||
let wc = WorkspaceCrypt::new(KEY);
|
||||
for len in 0..64 {
|
||||
let decoded = B64
|
||||
.decode(wc.encrypt_str_in(&"x".repeat(len), CipherFormat::V2))
|
||||
.unwrap();
|
||||
assert!(!decoded.is_empty() && decoded.len() % 16 == 0, "len {len}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -184,7 +184,7 @@ pub enum ObjectType {
|
||||
DatatableMigration,
|
||||
}
|
||||
|
||||
pub const LATEST_GIT_SYNC_SCRIPT_PATH: &str = "hub/28994/sync-script-to-git-repo-windmill";
|
||||
pub const LATEST_GIT_SYNC_SCRIPT_PATH: &str = "hub/29151/sync-script-to-git-repo-windmill";
|
||||
|
||||
/// Hub script that applies a repository's state back into a workspace
|
||||
/// (the repo → Windmill / "pull" direction). Same script the UI runs from
|
||||
@@ -192,7 +192,7 @@ pub const LATEST_GIT_SYNC_SCRIPT_PATH: &str = "hub/28994/sync-script-to-git-repo
|
||||
/// ignores the slug, so the slug is kept free of characters that would be
|
||||
/// percent-encoded into the run URL (a `:` becomes `%3A`, which some hardened
|
||||
/// reverse proxies reject as double-encoding when the client re-encodes it).
|
||||
pub const GIT_SYNC_PULL_SCRIPT_PATH: &str = "hub/28993/git-sync-init-repository-windmill";
|
||||
pub const GIT_SYNC_PULL_SCRIPT_PATH: &str = "hub/29150/git-sync-init-repository-windmill";
|
||||
|
||||
/// Prefix used to identify fork workspaces. A workspace whose id starts with this string is a
|
||||
/// fork of another workspace.
|
||||
|
||||
@@ -563,7 +563,8 @@ async fn check_path_conflict(db: &DB, w_id: &str, path: &str) -> Result<()> {
|
||||
///
|
||||
/// The check is purely structural and never decrypts, so it cannot act as a
|
||||
/// decryption/padding oracle for a caller who can write but not read secrets.
|
||||
/// `encrypt` (AES-256-CBC) always yields standard base64 decoding to a non-zero
|
||||
/// `encrypt` (AES-256-GCM, or legacy AES-256-CBC; see `windmill_common::workspace_crypt`)
|
||||
/// always yields standard base64 decoding to a non-zero
|
||||
/// multiple of the 16-byte block size; anything else cannot be our ciphertext.
|
||||
/// Values stored by an external backend ($vault:/$aws_sm:/$azure_kv: markers)
|
||||
/// are not workspace ciphertext and are passed through untouched.
|
||||
|
||||
+14
-14
@@ -1,21 +1,21 @@
|
||||
# windmill-chat
|
||||
# @windmill-labs/chat-sdk
|
||||
|
||||
Build a chat interface on a Windmill flow deployed in **chat mode**, from any frontend
|
||||
or from a Windmill raw app. The library is headless: it runs the flow, follows the
|
||||
answer as it streams, keeps the conversation history, and hands you state to render.
|
||||
|
||||
```
|
||||
npm install windmill-chat
|
||||
npm install @windmill-labs/chat-sdk
|
||||
```
|
||||
|
||||
No runtime dependencies. Optional peers: `react` for `windmill-chat/react`, `ai` for
|
||||
`windmill-chat/ai-sdk`, `@assistant-ui/react` for `windmill-chat/assistant-ui`.
|
||||
No runtime dependencies. Optional peers: `react` for `@windmill-labs/chat-sdk/react`, `ai` for
|
||||
`@windmill-labs/chat-sdk/ai-sdk`, `@assistant-ui/react` for `@windmill-labs/chat-sdk/assistant-ui`.
|
||||
|
||||
| You build the UI with | Import | You get |
|
||||
|---|---|---|
|
||||
| Vercel AI SDK `useChat`, AI Elements | `windmill-chat/ai-sdk` | a `ChatTransport`: `useChat({ transport })`, nothing else changes |
|
||||
| assistant-ui | `windmill-chat/assistant-ui` | a runtime for `AssistantRuntimeProvider`, threads included |
|
||||
| Your own components | `windmill-chat/react` or `windmill-chat` | a hook / a store with messages, status and actions |
|
||||
| Vercel AI SDK `useChat`, AI Elements | `@windmill-labs/chat-sdk/ai-sdk` | a `ChatTransport`: `useChat({ transport })`, nothing else changes |
|
||||
| assistant-ui | `@windmill-labs/chat-sdk/assistant-ui` | a runtime for `AssistantRuntimeProvider`, threads included |
|
||||
| Your own components | `@windmill-labs/chat-sdk/react` or `@windmill-labs/chat-sdk` | a hook / a store with messages, status and actions |
|
||||
|
||||
## The flow
|
||||
|
||||
@@ -31,7 +31,7 @@ so an AI agent step remembers earlier turns. The answer is:
|
||||
|
||||
```tsx
|
||||
import { useChat } from '@ai-sdk/react'
|
||||
import { createWindmillChatTransport } from 'windmill-chat/ai-sdk'
|
||||
import { createWindmillChatTransport } from '@windmill-labs/chat-sdk/ai-sdk'
|
||||
|
||||
const transport = createWindmillChatTransport({
|
||||
baseUrl: 'https://app.windmill.dev',
|
||||
@@ -65,7 +65,7 @@ explanatory error.
|
||||
|
||||
```tsx
|
||||
import { AssistantRuntimeProvider } from '@assistant-ui/react'
|
||||
import { useWindmillRuntime } from 'windmill-chat/assistant-ui'
|
||||
import { useWindmillRuntime } from '@windmill-labs/chat-sdk/assistant-ui'
|
||||
|
||||
export function Support() {
|
||||
const runtime = useWindmillRuntime({ baseUrl, workspace, flowPath, token })
|
||||
@@ -84,7 +84,7 @@ through `Reasoning`. It takes the same options as `useWindmillChat` below.
|
||||
## React
|
||||
|
||||
```tsx
|
||||
import { useWindmillChat } from 'windmill-chat/react'
|
||||
import { useWindmillChat } from '@windmill-labs/chat-sdk/react'
|
||||
|
||||
export function Support() {
|
||||
const chat = useWindmillChat({
|
||||
@@ -155,7 +155,7 @@ change, and returns the unsubscribe function. That is the Svelte store contract,
|
||||
`$chat` works as is; other frameworks wrap it in a few lines.
|
||||
|
||||
```ts
|
||||
import { createChat } from 'windmill-chat'
|
||||
import { createChat } from '@windmill-labs/chat-sdk'
|
||||
|
||||
const chat = createChat({ baseUrl, workspace, flowPath, token })
|
||||
chat.subscribe((state) => render(state))
|
||||
@@ -164,7 +164,7 @@ await chat.sendMessage('Hello')
|
||||
|
||||
```svelte
|
||||
<script>
|
||||
import { createChat } from 'windmill-chat'
|
||||
import { createChat } from '@windmill-labs/chat-sdk'
|
||||
const chat = createChat({ flowPath: 'f/support/assistant' })
|
||||
</script>
|
||||
|
||||
@@ -331,8 +331,8 @@ custom integrations.
|
||||
## For AI coding agents
|
||||
|
||||
When asked to add a chat over a Windmill flow: the flow must be deployed with chat mode
|
||||
on. Pick the entry point from the table at the top (`useChat` → `windmill-chat/ai-sdk`,
|
||||
assistant-ui → `windmill-chat/assistant-ui`, otherwise `windmill-chat/react`). Inside a
|
||||
on. Pick the entry point from the table at the top (`useChat` → `@windmill-labs/chat-sdk/ai-sdk`,
|
||||
assistant-ui → `@windmill-labs/chat-sdk/assistant-ui`, otherwise `@windmill-labs/chat-sdk/react`). Inside a
|
||||
Windmill raw app pass only `flowPath`. Elsewhere pass `baseUrl`, `workspace` and a
|
||||
`token`; for a public page use a token scoped to `jobs:run:flows:<flowPath>` and leave
|
||||
`history` at its default. Render `role`, `content`, `pending`, `success` and
|
||||
|
||||
Generated
+2
-2
@@ -1,11 +1,11 @@
|
||||
{
|
||||
"name": "windmill-chat",
|
||||
"name": "@windmill-labs/chat-sdk",
|
||||
"version": "1.830.0",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "windmill-chat",
|
||||
"name": "@windmill-labs/chat-sdk",
|
||||
"version": "1.830.0",
|
||||
"license": "Apache-2.0",
|
||||
"devDependencies": {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"name": "windmill-chat",
|
||||
"name": "@windmill-labs/chat-sdk",
|
||||
"description": "Build chat interfaces on Windmill flows deployed in chat mode, from any frontend or raw app",
|
||||
"version": "1.830.0",
|
||||
"author": "Ruben Fiszel",
|
||||
|
||||
@@ -107,7 +107,8 @@ export interface VariableFile {
|
||||
* plaintext value a user authored by hand.
|
||||
*
|
||||
* Mirrors the server guard (windmill-store/src/variables.rs): workspace
|
||||
* ciphertext (AES-256-CBC, base64) is standard base64 decoding to a non-zero
|
||||
* ciphertext (AES-256-GCM or legacy AES-256-CBC, base64) is standard base64
|
||||
* decoding to a non-zero
|
||||
* multiple of the 16-byte block size. External secret-backend markers
|
||||
* ($vault:/$aws_sm:/$azure_kv:) are stored verbatim too, so they count as
|
||||
* already-encrypted. This is a shape check only — it never decrypts.
|
||||
|
||||
Generated
+4
-4
@@ -6747,16 +6747,16 @@ Apply it to customer data, internal notes and anything else a viewer of the demo
|
||||
|
||||
### Chat UIs over a flow in chat mode
|
||||
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add \`windmill-chat\` to \`package.json\` and use it directly, it detects the app's Windmill and credential.
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add \`@windmill-labs/chat-sdk\` to \`package.json\` and use it directly, it detects the app's Windmill and credential.
|
||||
|
||||
\`\`\`tsx
|
||||
import { useWindmillChat } from 'windmill-chat/react'
|
||||
import { useWindmillChat } from '@windmill-labs/chat-sdk/react'
|
||||
|
||||
const chat = useWindmillChat({ flowPath: 'f/support/assistant' })
|
||||
// chat.messages ({ role, content, pending, success, tool? }), chat.status, chat.sendMessage(text), chat.stop()
|
||||
\`\`\`
|
||||
|
||||
\`windmill-chat/ai-sdk\` gives a \`ChatTransport\` for the Vercel AI SDK's \`useChat\`, \`windmill-chat/assistant-ui\` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs \`jobs:run\` in its frontend SDK scopes, plus \`flow_conversations:write\` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
\`@windmill-labs/chat-sdk/ai-sdk\` gives a \`ChatTransport\` for the Vercel AI SDK's \`useChat\`, \`@windmill-labs/chat-sdk/assistant-ui\` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs \`jobs:run\` in its frontend SDK scopes, plus \`flow_conversations:write\` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
|
||||
## Backend runnables
|
||||
|
||||
@@ -6917,7 +6917,7 @@ def main(user_id: str):
|
||||
7. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you.
|
||||
8. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable.
|
||||
9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying.
|
||||
10. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
10. **Use \`@windmill-labs/chat-sdk\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
`,
|
||||
"triggers": `---
|
||||
name: triggers
|
||||
|
||||
@@ -33,6 +33,10 @@
|
||||
;[norder[i], norder[i + 1]] = [norder[i + 1], norder[i]]
|
||||
}
|
||||
defaultScripts.update((s) => ({ ...s, order: norder }))
|
||||
await persist()
|
||||
}
|
||||
|
||||
async function persist() {
|
||||
await WorkspaceService.editDefaultScripts({
|
||||
workspace: $operatingWorkspace!,
|
||||
requestBody: $defaultScripts
|
||||
@@ -78,7 +82,7 @@
|
||||
size="xs"
|
||||
color="red"
|
||||
checked={$defaultScripts?.hidden?.includes(lang)}
|
||||
on:change={(e) => {
|
||||
on:change={async (e) => {
|
||||
let toggled = e.detail
|
||||
if (toggled) {
|
||||
defaultScripts.update((s) => ({
|
||||
@@ -91,6 +95,7 @@
|
||||
hidden: (s?.hidden ?? []).filter((h) => h != lang)
|
||||
}))
|
||||
}
|
||||
await persist()
|
||||
}}
|
||||
/>
|
||||
</div>
|
||||
|
||||
@@ -103,6 +103,7 @@ import {
|
||||
type RawAppRunsProvider,
|
||||
type RawAppScreenshotRequester
|
||||
} from '$lib/components/raw_apps/utils'
|
||||
import { resolveGlobalDraftStoragePathByKind } from '$lib/components/copilot/chat/global/userDraftAdapter'
|
||||
import type {
|
||||
RawAppDomQuery,
|
||||
RawAppDomRequester,
|
||||
@@ -1120,7 +1121,7 @@ function createRuntime(session: Session): SessionRuntime {
|
||||
},
|
||||
async requestRuntimeLogs(limit, appPath) {
|
||||
const path =
|
||||
appPath ??
|
||||
(appPath && draftStoragePath(session, 'raw_app', appPath)) ??
|
||||
activeDomAppPath ??
|
||||
(runtimeLogRequesters.size === 1 ? [...runtimeLogRequesters.keys()][0] : undefined)
|
||||
if (path === undefined) {
|
||||
@@ -1156,7 +1157,7 @@ function createRuntime(session: Session): SessionRuntime {
|
||||
// its element's app even when another tab is now visible), else the
|
||||
// active preview, else the only one open.
|
||||
const path =
|
||||
query.appPath ??
|
||||
(query.appPath && draftStoragePath(session, 'raw_app', query.appPath)) ??
|
||||
activeDomAppPath ??
|
||||
(domRequesters.size === 1 ? [...domRequesters.keys()][0] : undefined)
|
||||
if (path === undefined) return undefined
|
||||
@@ -1372,6 +1373,18 @@ export function removeSession(sessionId: string): void {
|
||||
deleteSessionState(sessionId)
|
||||
}
|
||||
|
||||
// The chat calls a new item by its chosen name while its draft is stored at
|
||||
// `draft_<uuid>`; resolve the name the way the chat's write tools do.
|
||||
function draftStoragePath(
|
||||
session: Session,
|
||||
kind: 'script' | 'flow' | 'raw_app',
|
||||
path: string
|
||||
): string {
|
||||
return session.workspace_id
|
||||
? resolveGlobalDraftStoragePathByKind(session.workspace_id, kind, path)
|
||||
: path
|
||||
}
|
||||
|
||||
// Register the global open_preview tool handler once at module load. It
|
||||
// dispatches to the *calling* session (sessionId threaded from the tool ctx),
|
||||
// falling back to the UI-active session only when none was passed — so a
|
||||
@@ -1387,7 +1400,10 @@ setOpenPreviewHandler(async ({ sessionId: callerSessionId, kind, path, mode }) =
|
||||
if (!session) {
|
||||
return 'Error: no active session to open the preview in.'
|
||||
}
|
||||
const target = previewTargetForSessionTarget(kind, path)
|
||||
const target = previewTargetForSessionTarget(
|
||||
kind,
|
||||
mode === 'view' || kind === 'pipeline' ? path : draftStoragePath(session, kind, path)
|
||||
)
|
||||
if (!target) {
|
||||
return `Error: ${kind} targets cannot be shown in the preview panel.`
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"gitSyncTest": "hub/28950/git-repo-test-read-write-windmill",
|
||||
"gitInitRepo": "hub/28993/git-sync-init-repository-windmill",
|
||||
"gitInitRepo": "hub/29150/git-sync-init-repository-windmill",
|
||||
"slackErrorHandler": "hub/28794/workspace-or-schedule-error-handler-slack",
|
||||
"emailErrorHandler": "hub/19795/workspace-or-error-handler-email",
|
||||
"slackRecoveryHandler": "hub/28791/slack/schedule-recovery-handler-slack",
|
||||
|
||||
@@ -1024,16 +1024,16 @@ Apply it to customer data, internal notes and anything else a viewer of the demo
|
||||
|
||||
### Chat UIs over a flow in chat mode
|
||||
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add \`windmill-chat\` to \`package.json\` and use it directly, it detects the app's Windmill and credential.
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add \`@windmill-labs/chat-sdk\` to \`package.json\` and use it directly, it detects the app's Windmill and credential.
|
||||
|
||||
\`\`\`tsx
|
||||
import { useWindmillChat } from 'windmill-chat/react'
|
||||
import { useWindmillChat } from '@windmill-labs/chat-sdk/react'
|
||||
|
||||
const chat = useWindmillChat({ flowPath: 'f/support/assistant' })
|
||||
// chat.messages ({ role, content, pending, success, tool? }), chat.status, chat.sendMessage(text), chat.stop()
|
||||
\`\`\`
|
||||
|
||||
\`windmill-chat/ai-sdk\` gives a \`ChatTransport\` for the Vercel AI SDK's \`useChat\`, \`windmill-chat/assistant-ui\` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs \`jobs:run\` in its frontend SDK scopes, plus \`flow_conversations:write\` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
\`@windmill-labs/chat-sdk/ai-sdk\` gives a \`ChatTransport\` for the Vercel AI SDK's \`useChat\`, \`@windmill-labs/chat-sdk/assistant-ui\` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs \`jobs:run\` in its frontend SDK scopes, plus \`flow_conversations:write\` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
|
||||
## Backend runnables
|
||||
|
||||
@@ -1183,7 +1183,7 @@ def main(user_id: str):
|
||||
7. **Mark sensitive UI with \`data-wm-no-record\`** — it is what keeps that data out of a recorded demo; passwords are handled for you.
|
||||
8. **Reach for \`backendAsync\` + \`waitJob\`** for long work — never a hand-written job-polling runnable.
|
||||
9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying.
|
||||
10. **Use \`windmill-chat\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
10. **Use \`@windmill-labs/chat-sdk\` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
`;
|
||||
|
||||
export const PIPELINE_BASE = `# Data pipeline authoring
|
||||
|
||||
@@ -378,16 +378,16 @@ Apply it to customer data, internal notes and anything else a viewer of the demo
|
||||
|
||||
### Chat UIs over a flow in chat mode
|
||||
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add `windmill-chat` to `package.json` and use it directly, it detects the app's Windmill and credential.
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add `@windmill-labs/chat-sdk` to `package.json` and use it directly, it detects the app's Windmill and credential.
|
||||
|
||||
```tsx
|
||||
import { useWindmillChat } from 'windmill-chat/react'
|
||||
import { useWindmillChat } from '@windmill-labs/chat-sdk/react'
|
||||
|
||||
const chat = useWindmillChat({ flowPath: 'f/support/assistant' })
|
||||
// chat.messages ({ role, content, pending, success, tool? }), chat.status, chat.sendMessage(text), chat.stop()
|
||||
```
|
||||
|
||||
`windmill-chat/ai-sdk` gives a `ChatTransport` for the Vercel AI SDK's `useChat`, `windmill-chat/assistant-ui` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs `jobs:run` in its frontend SDK scopes, plus `flow_conversations:write` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
`@windmill-labs/chat-sdk/ai-sdk` gives a `ChatTransport` for the Vercel AI SDK's `useChat`, `@windmill-labs/chat-sdk/assistant-ui` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs `jobs:run` in its frontend SDK scopes, plus `flow_conversations:write` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
|
||||
## Backend runnables
|
||||
|
||||
@@ -548,4 +548,4 @@ def main(user_id: str):
|
||||
7. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you.
|
||||
8. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable.
|
||||
9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying.
|
||||
10. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
10. **Use `@windmill-labs/chat-sdk` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
|
||||
@@ -139,16 +139,16 @@ Apply it to customer data, internal notes and anything else a viewer of the demo
|
||||
|
||||
### Chat UIs over a flow in chat mode
|
||||
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add `windmill-chat` to `package.json` and use it directly, it detects the app's Windmill and credential.
|
||||
A flow deployed with chat mode on is a chat backend (streaming answer, tool calls, memory, conversation history). Do not drive it through a runnable: add `@windmill-labs/chat-sdk` to `package.json` and use it directly, it detects the app's Windmill and credential.
|
||||
|
||||
```tsx
|
||||
import { useWindmillChat } from 'windmill-chat/react'
|
||||
import { useWindmillChat } from '@windmill-labs/chat-sdk/react'
|
||||
|
||||
const chat = useWindmillChat({ flowPath: 'f/support/assistant' })
|
||||
// chat.messages ({ role, content, pending, success, tool? }), chat.status, chat.sendMessage(text), chat.stop()
|
||||
```
|
||||
|
||||
`windmill-chat/ai-sdk` gives a `ChatTransport` for the Vercel AI SDK's `useChat`, `windmill-chat/assistant-ui` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs `jobs:run` in its frontend SDK scopes, plus `flow_conversations:write` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
`@windmill-labs/chat-sdk/ai-sdk` gives a `ChatTransport` for the Vercel AI SDK's `useChat`, `@windmill-labs/chat-sdk/assistant-ui` a runtime for assistant-ui. The flow must be deployed, not a draft. A sandboxed app needs `jobs:run` in its frontend SDK scopes, plus `flow_conversations:write` for the conversation sidebar; without them the chat keeps history in the browser.
|
||||
|
||||
## Backend runnables
|
||||
|
||||
@@ -320,4 +320,4 @@ def main(user_id: str):
|
||||
7. **Mark sensitive UI with `data-wm-no-record`** — it is what keeps that data out of a recorded demo; passwords are handled for you.
|
||||
8. **Reach for `backendAsync` + `waitJob`** for long work — never a hand-written job-polling runnable.
|
||||
9. **Deploy what a path runnable points at** — a path runnable aimed at a draft fails at runtime; tell the user what needs deploying.
|
||||
10. **Use `windmill-chat` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
10. **Use `@windmill-labs/chat-sdk` for a chat over a chat-mode flow** — never a runnable that runs the flow and polls its stream.
|
||||
|
||||
Reference in New Issue
Block a user