mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-12 16:02:56 +00:00
fix: key Slack conversations with the workspace secret and require signed Slack requests
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
9620255636
commit
31d6fec188
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "WITH a AS (DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day')\n DELETE FROM slack_thread_agent WHERE created_at < now() - interval '90 days'",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "127d1d6a9cec24f34ff91718cf62befb986da3c1081b08a3a650435c53a88df5"
|
||||
}
|
||||
+3
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent\n WHERE workspace_id = $1 ORDER BY channel_name",
|
||||
"query": "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c\n JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path\n WHERE c.workspace_id = $1 AND c.agent_path = $2 ORDER BY c.channel_name",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -21,6 +21,7 @@
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
@@ -30,5 +31,5 @@
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8"
|
||||
"hash": "35b5261c148af3bf68de2772847125dcfc4f0f8a22b5eeb927f3484d7534826c"
|
||||
}
|
||||
+2
-3
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent\n WHERE workspace_id = $1 AND agent_path = $2 ORDER BY channel_name",
|
||||
"query": "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c\n JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path\n WHERE c.workspace_id = $1 ORDER BY c.channel_name",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -21,7 +21,6 @@
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
@@ -31,5 +30,5 @@
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120"
|
||||
"hash": "538edc1ca25281f2b2b52f55ef787f4b71f41914694d090a5f10ff6478627bb4"
|
||||
}
|
||||
-12
@@ -1,12 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day'",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "cad02f11ce7127c1e74eface0cb01939a9effee10fb7664d08cf39da5d2e13d7"
|
||||
}
|
||||
@@ -28,6 +28,7 @@ CREATE TABLE slack_thread_agent (
|
||||
);
|
||||
|
||||
CREATE INDEX index_slack_thread_agent_workspace ON slack_thread_agent (workspace_id, agent_path);
|
||||
CREATE INDEX index_slack_thread_agent_created_at ON slack_thread_agent (created_at);
|
||||
|
||||
-- The Slack messages an agent has taken up. Slack resends an event it did not see acknowledged,
|
||||
-- and posts a direct message mentioning the bot as two events; only the first is answered.
|
||||
|
||||
@@ -42,9 +42,6 @@ pub(crate) async fn run_agent(
|
||||
Query(run_query): Query<RunJobQuery>,
|
||||
Json(args): Json<HashMap<String, Box<RawValue>>>,
|
||||
) -> Result<(StatusCode, String)> {
|
||||
#[cfg(feature = "enterprise")]
|
||||
crate::jobs::check_license_key_valid().await?;
|
||||
|
||||
let path = path.to_path();
|
||||
check_scopes(&authed, || format!("jobs:run:agents:{path}"))?;
|
||||
|
||||
@@ -62,6 +59,9 @@ pub(crate) async fn run_agent_as(
|
||||
run_query: &RunJobQuery,
|
||||
args: &HashMap<String, Box<RawValue>>,
|
||||
) -> Result<uuid::Uuid> {
|
||||
#[cfg(feature = "enterprise")]
|
||||
crate::jobs::check_license_key_valid().await?;
|
||||
|
||||
// Read through the caller's own permissions: reading the agent is what allows running it.
|
||||
let mut tx = user_db.clone().begin(authed).await?;
|
||||
let value = sqlx::query_scalar!(
|
||||
|
||||
@@ -22,8 +22,8 @@ use windmill_common::{
|
||||
db::UserDB,
|
||||
error::{Error, JsonResult, Result},
|
||||
oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH,
|
||||
utils::{require_admin, StripPath},
|
||||
variables::get_secret_value_as_admin,
|
||||
utils::{calculate_hash, require_admin, StripPath},
|
||||
variables::{get_secret_value_as_admin, get_workspace_key},
|
||||
BASE_URL,
|
||||
};
|
||||
|
||||
@@ -58,19 +58,23 @@ struct ChannelAgent {
|
||||
agent_path: String,
|
||||
}
|
||||
|
||||
/// Listed through the caller's permissions: an agent they cannot read stays unnamed.
|
||||
async fn list_channel_agents(
|
||||
_authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
authed: ApiAuthed,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path(w_id): Path<String>,
|
||||
) -> JsonResult<Vec<ChannelAgent>> {
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
let rows = sqlx::query_as!(
|
||||
ChannelAgent,
|
||||
"SELECT channel_id, channel_name, agent_path FROM slack_channel_agent
|
||||
WHERE workspace_id = $1 ORDER BY channel_name",
|
||||
"SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c
|
||||
JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path
|
||||
WHERE c.workspace_id = $1 ORDER BY c.channel_name",
|
||||
w_id
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(Json(rows))
|
||||
}
|
||||
|
||||
@@ -82,8 +86,9 @@ struct AgentSlack {
|
||||
}
|
||||
|
||||
async fn list_agent_channels(
|
||||
_authed: ApiAuthed,
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path((w_id, path)): Path<(String, StripPath)>,
|
||||
) -> JsonResult<AgentSlack> {
|
||||
let slack_team_name = sqlx::query_scalar!(
|
||||
@@ -94,15 +99,18 @@ async fn list_agent_channels(
|
||||
.fetch_optional(&db)
|
||||
.await?
|
||||
.flatten();
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
let channels = sqlx::query_as!(
|
||||
ChannelAgent,
|
||||
"SELECT channel_id, channel_name, agent_path FROM slack_channel_agent
|
||||
WHERE workspace_id = $1 AND agent_path = $2 ORDER BY channel_name",
|
||||
"SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c
|
||||
JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path
|
||||
WHERE c.workspace_id = $1 AND c.agent_path = $2 ORDER BY c.channel_name",
|
||||
w_id,
|
||||
path.to_path()
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(Json(AgentSlack { slack_team_name, channels }))
|
||||
}
|
||||
|
||||
@@ -304,6 +312,12 @@ enum Route {
|
||||
/// Whether an agent answers this message; if so, the answer is under way when this returns.
|
||||
/// `false` leaves the message to the workspace's command script.
|
||||
pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result<bool> {
|
||||
// Without it the Slack endpoints take unsigned requests, and the sender's `user_id` would
|
||||
// pick whose permissions the agent runs with.
|
||||
#[cfg(feature = "oauth2")]
|
||||
if crate::SLACK_SIGNING_SECRET.is_none() {
|
||||
return Ok(false);
|
||||
}
|
||||
let text = decode_entities(&msg.text);
|
||||
let (name, question) = split_agent_name(&text);
|
||||
let thread_root = msg.thread_ts.clone().or_else(|| msg.ts.clone());
|
||||
@@ -370,11 +384,15 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result<bool> {
|
||||
if !first {
|
||||
return Ok(true);
|
||||
}
|
||||
sqlx::query!(
|
||||
"DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day'"
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
}
|
||||
// Recorded before answering, so a follow-up sent while the first answer is on its way
|
||||
// already finds its agent. A named agent is recorded once the sender's name resolves.
|
||||
if let (Route::Agent(path), Some(root)) = (&route, &thread_root) {
|
||||
if let Err(e) =
|
||||
record_thread_agent(db, &msg.team_id, &msg.channel_id, root, &w_id, path).await
|
||||
{
|
||||
tracing::warn!("Could not record the Slack thread's agent: {e:#}");
|
||||
}
|
||||
}
|
||||
|
||||
if !matches!(route, Route::Help) {
|
||||
@@ -385,6 +403,7 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result<bool> {
|
||||
// ponytail: the reply waits in this API process, so a restart mid-run loses the reply (not
|
||||
// the run, which still shows in Windmill); a job-completion hook would make it durable.
|
||||
tokio::spawn(async move {
|
||||
prune(&db).await;
|
||||
if let Err(e) = answer(&db, &w_id, route, &question, msg).await {
|
||||
tracing::error!("Slack agent answer failed: {e:#}");
|
||||
}
|
||||
@@ -392,6 +411,41 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result<bool> {
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
async fn record_thread_agent(
|
||||
db: &DB,
|
||||
team_id: &str,
|
||||
channel_id: &str,
|
||||
thread_ts: &str,
|
||||
w_id: &str,
|
||||
agent_path: &str,
|
||||
) -> Result<()> {
|
||||
sqlx::query!(
|
||||
"INSERT INTO slack_thread_agent (slack_team_id, channel_id, thread_ts, workspace_id, agent_path)
|
||||
VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING",
|
||||
team_id,
|
||||
channel_id,
|
||||
thread_ts,
|
||||
w_id,
|
||||
agent_path
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Best effort: a message already marked answered must still be answered if this fails.
|
||||
async fn prune(db: &DB) {
|
||||
let res = sqlx::query!(
|
||||
"WITH a AS (DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day')
|
||||
DELETE FROM slack_thread_agent WHERE created_at < now() - interval '90 days'"
|
||||
)
|
||||
.execute(db)
|
||||
.await;
|
||||
if let Err(e) = res {
|
||||
tracing::warn!("Could not prune Slack agent routing: {e:#}");
|
||||
}
|
||||
}
|
||||
|
||||
/// `+name rest` or `~name rest` → (name, rest).
|
||||
fn split_agent_name(text: &str) -> (Option<&str>, &str) {
|
||||
let text = text.trim();
|
||||
@@ -605,23 +659,16 @@ async fn answer(
|
||||
.await;
|
||||
}
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO slack_thread_agent (slack_team_id, channel_id, thread_ts, workspace_id, agent_path)
|
||||
VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING",
|
||||
msg.team_id,
|
||||
channel,
|
||||
thread_root,
|
||||
w_id,
|
||||
agent_path
|
||||
)
|
||||
.execute(db)
|
||||
.await?;
|
||||
record_thread_agent(db, &msg.team_id, &channel, &thread_root, w_id, &agent_path).await?;
|
||||
|
||||
// Keyed with the workspace's secret: any member could rebuild the plain thread and user ids
|
||||
// and load this person's conversation, tool results included, through the run API.
|
||||
let workspace_key = get_workspace_key(w_id, db).await?;
|
||||
let run_query = RunJobQuery {
|
||||
memory_id: Some(format!(
|
||||
"slack:{}:{}:{}:{}",
|
||||
memory_id: Some(calculate_hash(&format!(
|
||||
"{workspace_key}:slack:{}:{}:{}:{}",
|
||||
msg.team_id, channel, thread_root, msg.user_id
|
||||
)),
|
||||
))),
|
||||
..Default::default()
|
||||
};
|
||||
let args: HashMap<String, Box<RawValue>> = [(
|
||||
|
||||
Reference in New Issue
Block a user