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

with cc

Browse Source
This commit is contained in:
Ruben Fiszel
2026-02-06 13:56:30 +00:00
parent 5aaecf0da3
commit 2b8af5ddd4
19 changed files with 2411 additions and 1477 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/2026-02-06-this-session-is-being-continued-from-a-previous-co.txt
+1998
View File
File diff suppressed because it is too large Load Diff
backend/windmill-api-auth/src/lib.rs
+2
View File
@@ -7,7 +7,9 @@
*/
mod auth;
pub mod permissions;
pub mod scopes;
pub mod tokens;
mod types;
pub use auth::*;
backend/windmill-api-auth/src/permissions.rs
+156
View File
@@ -0,0 +1,156 @@
/*
* Author: Windmill Labs, Inc
* Copyright: Windmill Labs, Inc 2024
* 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.
*/
use windmill_common::error::{Error, Result};
use windmill_common::DB;
use crate::ApiAuthed;
/// Check if the user is an owner of the given path.
pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
let splitted = path.split("/").collect::<Vec<&str>>();
if splitted[0] == "u" {
if splitted[1] == authed.username {
Ok(())
} else {
Err(Error::BadRequest(format!(
"only the owner {} is authorized to perform this operation",
splitted[1]
)))
}
} else if splitted[0] == "f" {
require_is_folder_owner(authed, splitted[1])
} else {
Err(Error::BadRequest(format!(
"Not recognized path kind: {}",
path
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be owner of an empty path"
)))
}
}
pub fn is_folder_owner(
ApiAuthed { is_admin, folders, .. }: &ApiAuthed,
name: &str,
) -> bool {
if *is_admin {
true
} else {
folders.into_iter().any(|x| x.0 == name && x.2)
}
}
pub fn require_is_folder_owner(authed: &ApiAuthed, name: &str) -> Result<()> {
if is_folder_owner(authed, name) {
Ok(())
} else {
Err(Error::NotAuthorized(format!(
"You are not owner of the folder {}",
name
)))
}
}
pub fn get_perm_in_extra_perms_for_authed(
v: serde_json::Value,
authed: &ApiAuthed,
) -> Option<bool> {
match v {
serde_json::Value::Object(obj) => {
let mut keys = vec![format!("u/{}", authed.username)];
for g in authed.groups.iter() {
keys.push(format!("g/{}", g));
}
let mut res = None;
for k in keys {
if let Some(v) = obj.get(&k) {
if let Some(v) = v.as_bool() {
if v {
return Some(true);
}
res = Some(v);
}
}
}
res
}
_ => None,
}
}
/// Generic require_is_writer with a configurable SQL query.
pub async fn require_is_writer(
authed: &ApiAuthed,
path: &str,
w_id: &str,
db: DB,
query: &str,
kind: &str,
) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
if require_owner_of_path(authed, path).is_ok() {
return Ok(());
}
if path.starts_with("f/") && path.split('/').count() >= 2 {
let folder = path.split('/').nth(1).unwrap();
let extra_perms = sqlx::query_scalar!(
"SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2",
folder,
w_id
)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let is_folder_writer =
get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false);
if is_folder_writer {
return Ok(());
}
}
}
let extra_perms = sqlx::query_scalar(query)
.bind(path)
.bind(w_id)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let perm = get_perm_in_extra_perms_for_authed(perms, authed);
match perm {
Some(true) => Ok(()),
Some(false) => Err(Error::BadRequest(format!(
"User {} is not a writer of {kind} path {path}",
authed.username
))),
None => Err(Error::BadRequest(format!(
"User {} has neither read or write permission on {kind} {path}",
authed.username
))),
}
} else {
Err(Error::BadRequest(format!(
"{path} does not exist yet and user {} is not an owner of the parent folder",
authed.username
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be writer of an empty path"
)))
}
}
backend/windmill-api-auth/src/scopes.rs
+31 -1
View File
@@ -9,7 +9,9 @@
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use windmill_common::error::{Error, Result};
use windmill_common::error::{self, Error, Result};
use windmill_common::jobs::check_tag_available_for_workspace_internal;
use windmill_common::DB;
use crate::ApiAuthed;
@@ -702,6 +704,34 @@ where
Ok(())
}
pub fn get_scope_tags(authed: &ApiAuthed) -> Option<Vec<&str>> {
authed.scopes.as_ref()?.iter().find_map(|s| {
if s.starts_with("if_jobs:filter_tags:") {
Some(
s.trim_start_matches("if_jobs:filter_tags:")
.split(",")
.collect::<Vec<_>>(),
)
} else {
None
}
})
}
pub async fn check_tag_available_for_workspace(
db: &DB,
w_id: &str,
tag: &Option<String>,
authed: &ApiAuthed,
) -> error::Result<()> {
if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) {
let tags = get_scope_tags(authed);
check_tag_available_for_workspace_internal(db, w_id, tag, &authed.email, tags).await
} else {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
backend/windmill-api-auth/src/tokens.rs
+97
View File
@@ -0,0 +1,97 @@
/*
* Author: Windmill Labs, Inc
* Copyright: Windmill Labs, Inc 2024
* 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.
*/
use sqlx::PgConnection;
use tracing::Instrument;
use windmill_audit::{audit_oss::audit_log, ActionKind};
use windmill_common::{
error::{Error, Result},
utils::rd_string,
worker::CLOUD_HOSTED,
DB,
};
use crate::ApiAuthed;
#[derive(serde::Deserialize)]
pub struct NewToken {
pub label: Option<String>,
pub expiration: Option<chrono::DateTime<chrono::Utc>>,
pub impersonate_email: Option<String>,
pub scopes: Option<Vec<String>>,
pub workspace_id: Option<String>,
}
impl NewToken {
pub fn new(
label: Option<String>,
expiration: Option<chrono::DateTime<chrono::Utc>>,
impersonate_email: Option<String>,
scopes: Option<Vec<String>>,
workspace_id: Option<String>,
) -> NewToken {
NewToken { label, expiration, impersonate_email, scopes, workspace_id }
}
}
pub async fn create_token_internal(
tx: &mut PgConnection,
db: &DB,
authed: &ApiAuthed,
token_config: NewToken,
) -> Result<String> {
let token = rd_string(32);
let is_super_admin = sqlx::query_scalar!(
"SELECT super_admin FROM password WHERE email = $1",
authed.email
)
.fetch_optional(&mut *tx)
.await?
.unwrap_or(false);
if *CLOUD_HOSTED {
let nb_tokens =
sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email)
.fetch_one(db)
.await?;
if nb_tokens.unwrap_or(0) >= 10000 {
return Err(Error::BadRequest(
"You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit"
.to_string(),
));
}
}
sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin, scopes, workspace_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
token,
authed.email,
token_config.label,
token_config.expiration,
is_super_admin,
token_config.scopes.as_ref().map(|x| x.as_slice()),
token_config.workspace_id,
)
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
authed,
"users.token.create",
ActionKind::Create,
&"global",
Some(&token[0..10]),
None,
)
.instrument(tracing::info_span!("token", email = &authed.email))
.await?;
Ok(token)
}
backend/windmill-api/src/args.rs
-678
View File
@@ -1,678 +0,0 @@
use std::collections::HashMap;
use axum::{
extract::{FromRequest, Multipart, Query, Request},
http::HeaderMap,
response::{IntoResponse, Response},
};
use bytes::Bytes;
use http::{header::CONTENT_TYPE, StatusCode};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::types::JsonRawValue;
use windmill_common::{
error::Error,
triggers::{RunnableFormat, RunnableFormatVersion, TriggerKind},
worker::to_raw_value,
DB,
};
use windmill_queue::PushArgsOwned;
use crate::{
db::ApiAuthed,
triggers::trigger_helpers::{get_runnable_format, RunnableId},
};
#[derive(Debug)]
pub enum RawBody {
Json(String),
CEJson(String),
Text(String),
Xml(String),
UrlEncoded(Bytes),
Multipart(Multipart),
Empty,
}
#[derive(Debug, Clone, Serialize)]
#[serde(untagged)]
pub enum Body {
HashMap(HashMap<String, Box<RawValue>>),
NoHashMap(Box<RawValue>),
}
#[derive(Debug, Clone, Default)]
pub struct WebhookArgsMetadata {
pub raw_string: Option<String>,
pub headers: HeaderMap,
pub query: Option<String>,
pub method: http::Method,
pub query_wrap_body: bool,
pub query_use_raw: bool,
pub query_include_header: Option<String>,
pub query_include_query: Option<String>,
}
pub struct RawWebhookArgs {
pub body: RawBody,
pub metadata: WebhookArgsMetadata,
}
#[derive(Debug, Clone)]
pub struct WebhookArgs {
pub body: Body,
pub metadata: WebhookArgsMetadata,
}
// capture
//
impl RawWebhookArgs {
#[cfg(not(feature = "parquet"))]
pub async fn process_multipart(
_multipart: Multipart,
_authed: &ApiAuthed,
_db: &DB,
_w_id: &str,
) -> Result<HashMap<String, Box<RawValue>>, Error> {
return Err(Error::BadRequest(format!(
"multipart/form-data requires the parquet feature"
)));
}
#[cfg(feature = "parquet")]
async fn process_multipart(
mut multipart: Multipart,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
) -> Result<HashMap<String, Box<RawValue>>, Error> {
use crate::job_helpers_oss::{
get_random_file_name, get_workspace_s3_resource, upload_file_internal,
};
use futures::TryStreamExt;
use object_store::{Attribute, Attributes};
use windmill_common::s3_helpers::build_object_store_client;
let (_, s3_resource) = get_workspace_s3_resource(authed, db, None, w_id, None).await?;
if let Some(s3_resource) = s3_resource {
let s3_client = build_object_store_client(&s3_resource).await?;
let mut body = HashMap::new();
let mut files = HashMap::new();
while let Some(field) = multipart.next_field().await.map_err(|e| {
Error::BadRequest(format!("Error reading multipart field: {}", e.body_text()))
})? {
if let Some(name) = field.name().map(|x| x.to_string()) {
if let Some(content_type) = field.content_type() {
let ext = field
.file_name()
.map(|x| x.split('.').last())
.flatten()
.map(|x| x.to_string());
let file_key = get_random_file_name(ext);
let options = Attributes::from_iter(vec![
(Attribute::ContentType, content_type.to_string()),
(
Attribute::ContentDisposition,
if let Some(filename) = field.file_name() {
format!("inline; filename=\"{}\"", filename)
} else {
"inline".to_string()
},
),
])
.into();
let bytes_stream = field
.into_stream()
.map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err));
upload_file_internal(s3_client.clone(), &file_key, bytes_stream, options)
.await?;
files.entry(name).or_insert(vec![]).push(serde_json::json!({
"s3": &file_key
}));
} else {
body.insert(name, to_raw_value(&field.text().await.unwrap_or_default()));
}
}
}
for (k, v) in files {
body.insert(k, to_raw_value(&v));
}
Ok(body)
} else {
Err(Error::BadRequest(format!(
"You need to connect your workspace to an S3 bucket to use multipart/form-data"
)))
}
}
pub async fn process_args(
self,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
force_use_raw: Option<bool>,
) -> Result<WebhookArgs, Error> {
let use_raw = force_use_raw.unwrap_or(self.metadata.query_use_raw);
match self.body {
RawBody::Multipart(multipart) => {
let body = Self::process_multipart(multipart, authed, db, w_id).await?;
Ok(WebhookArgs { body: Body::HashMap(body), metadata: self.metadata })
}
RawBody::Empty => {
let mut metadata = self.metadata;
if use_raw {
metadata.raw_string = Some("".to_string());
}
Ok(WebhookArgs { body: Body::HashMap(HashMap::new()), metadata })
}
RawBody::Text(s) | RawBody::Xml(s) => Ok(WebhookArgs {
body: Body::HashMap(HashMap::new()),
metadata: WebhookArgsMetadata { raw_string: Some(s), ..self.metadata },
}),
RawBody::UrlEncoded(bytes) => {
let mut metadata = self.metadata;
if use_raw {
let raw_string = String::from_utf8(bytes.to_vec())
.map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)))?;
metadata.raw_string = Some(raw_string);
}
let payload: HashMap<String, Option<String>> = serde_urlencoded::from_bytes(&bytes)
.map_err(|e| Error::BadRequest(format!("invalid urlencoded data: {}", e)))?;
let payload = payload
.into_iter()
.map(|(k, v)| (k, to_raw_value(&v)))
.collect::<HashMap<_, _>>();
Ok(WebhookArgs { body: Body::HashMap(payload), metadata })
}
RawBody::Json(s) => WebhookArgs::from_json(self.metadata, use_raw, s).await,
RawBody::CEJson(s) => WebhookArgs::from_ce_json(self.metadata, use_raw, s).await,
}
}
pub async fn to_main_args(
self,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
) -> Result<PushArgsOwned, Error> {
let args = self.process_args(authed, db, w_id, None).await?;
args.to_main_args()
}
pub async fn to_args_from_runnable(
self,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
runnable_id: RunnableId,
skip_preprocessor: Option<bool>,
) -> Result<PushArgsOwned, Error> {
let args = self.process_args(authed, db, w_id, None).await?;
args.to_args_from_runnable(db, w_id, runnable_id, skip_preprocessor)
.await
}
}
#[derive(Serialize)]
struct WebhookPreprocessorEvent {
kind: String,
body: Box<RawValue>,
raw_string: Option<String>,
headers: HashMap<String, Box<RawValue>>,
query: HashMap<String, Box<RawValue>>,
}
impl WebhookArgs {
pub fn to_main_args(self) -> Result<PushArgsOwned, Error> {
self.to_args_from_format(RunnableFormat {
has_preprocessor: false,
version: RunnableFormatVersion::V2,
})
}
pub async fn to_args_from_runnable(
self,
db: &DB,
w_id: &str,
runnable_id: RunnableId,
skip_preprocessor: Option<bool>,
) -> Result<PushArgsOwned, Error> {
if skip_preprocessor.unwrap_or(false) {
self.to_main_args()
} else {
let runnable_format =
get_runnable_format(runnable_id, w_id, db, &TriggerKind::Webhook).await?;
self.to_args_from_format(runnable_format)
}
}
pub fn to_args_from_format(
self,
runnable_format: RunnableFormat,
) -> Result<PushArgsOwned, Error> {
let headers = build_headers(
&self.metadata.headers,
self.metadata.query_include_header,
runnable_format.has_preprocessor,
);
let query = build_query(
self.metadata.query.as_deref(),
self.metadata.query_include_query,
runnable_format.has_preprocessor,
);
match runnable_format {
RunnableFormat { has_preprocessor: true, version: RunnableFormatVersion::V2 } => {
let mut args = HashMap::new();
args.insert(
"event".to_string(),
to_raw_value(&WebhookPreprocessorEvent {
kind: "webhook".to_string(),
body: to_raw_value(&self.body),
raw_string: self.metadata.raw_string,
headers,
query,
}),
);
Ok(PushArgsOwned { args, extra: None })
}
RunnableFormat { has_preprocessor, .. } => {
let mut extra = HashMap::new();
let WebhookArgsMetadata { query_wrap_body, raw_string, .. } = self.metadata;
for (k, v) in headers {
extra.insert(k, v);
}
for (k, v) in query {
extra.insert(k, v);
}
if let Some(raw_string) = raw_string {
extra.insert("raw_string".to_string(), to_raw_value(&raw_string));
}
if has_preprocessor {
// if has preprocessor, it has to be v1
extra.insert(
"wm_trigger".to_string(),
to_raw_value(&serde_json::json!({
"kind": "webhook",
})),
);
}
let extra = if extra.is_empty() { None } else { Some(extra) };
match self.body {
Body::HashMap(mut body) => {
if query_wrap_body {
body = HashMap::from([("body".to_string(), to_raw_value(&body))]);
}
Ok(PushArgsOwned { args: body, extra })
}
Body::NoHashMap(args) => {
let mut hm = HashMap::new();
hm.insert("body".to_string(), args);
Ok(PushArgsOwned { args: hm, extra })
}
}
}
}
}
}
#[derive(Deserialize)]
pub struct RequestQuery {
pub raw: Option<bool>,
pub wrap_body: Option<bool>,
pub include_header: Option<String>,
pub include_query: Option<String>,
}
async fn req_to_string<S: Send + Sync>(
req: Request<axum::body::Body>,
_state: &S,
) -> Result<String, Response> {
let bytes = Bytes::from_request(req, _state)
.await
.map_err(IntoResponse::into_response)?;
String::from_utf8(bytes.to_vec())
.map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())
}
pub async fn try_from_request_body<S>(
request: Request,
_state: &S,
is_http_trigger: bool,
) -> Result<RawWebhookArgs, Response>
where
S: Send + Sync,
{
let (content_type, metadata) = {
let headers_map = request.headers();
let content_type_header = headers_map.get(CONTENT_TYPE);
let content_type = content_type_header.and_then(|value| value.to_str().ok());
let uri = request.uri();
let request_query = Query::<RequestQuery>::try_from_uri(uri).unwrap().0;
let query = uri.query().map(|s| s.to_owned());
let raw = !is_http_trigger && request_query.raw.unwrap_or(false);
let wrap_body = !is_http_trigger && request_query.wrap_body.unwrap_or(false);
(
content_type,
WebhookArgsMetadata {
headers: headers_map.clone(),
query,
method: request.method().clone(),
raw_string: None,
query_wrap_body: wrap_body,
query_use_raw: raw,
query_include_header: request_query.include_header,
query_include_query: request_query.include_query,
},
)
};
let no_content_type = content_type.is_none();
if no_content_type || content_type.unwrap().starts_with("application/json") {
let bytes = Bytes::from_request(request, _state)
.await
.map_err(IntoResponse::into_response)?;
if no_content_type && bytes.is_empty() {
Ok(RawWebhookArgs { body: RawBody::Empty, metadata })
} else {
let str = String::from_utf8(bytes.to_vec())
.map_err(|e| Error::BadRequest(format!("invalid utf8: {}", e)).into_response())?;
Ok(RawWebhookArgs { body: RawBody::Json(str), metadata })
}
} else if content_type
.unwrap()
.starts_with("application/cloudevents+json")
{
let str = req_to_string(request, _state).await?;
Ok(RawWebhookArgs { body: RawBody::CEJson(str), metadata })
} else if content_type
.unwrap()
.starts_with("application/cloudevents-batch+json")
{
Err(
Error::BadRequest(format!("Cloud events batching is not supported yet"))
.into_response(),
)
} else if content_type.unwrap().starts_with("text/plain") {
let str = req_to_string(request, _state).await?;
Ok(RawWebhookArgs { body: RawBody::Text(str), metadata })
} else if content_type
.unwrap()
.starts_with("application/x-www-form-urlencoded")
{
let bytes = Bytes::from_request(request, _state)
.await
.map_err(IntoResponse::into_response)?;
Ok(RawWebhookArgs { body: RawBody::UrlEncoded(bytes), metadata })
} else if content_type.unwrap().starts_with("application/xml")
|| content_type.unwrap().starts_with("text/xml")
{
let str = req_to_string(request, _state).await?;
Ok(RawWebhookArgs { body: RawBody::Xml(str), metadata })
} else if content_type.unwrap().starts_with("multipart/form-data") {
let multipart = Multipart::from_request(request, _state)
.await
.map_err(IntoResponse::into_response)?;
Ok(RawWebhookArgs { body: RawBody::Multipart(multipart), metadata })
} else {
Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
}
}
#[axum::async_trait]
impl<S> FromRequest<S, axum::body::Body> for RawWebhookArgs
where
S: Send + Sync,
{
type Rejection = Response;
async fn from_request(request: Request, _state: &S) -> Result<Self, Self::Rejection> {
let args = try_from_request_body(request, _state, false).await?;
Ok(args)
}
}
lazy_static::lazy_static! {
static ref INCLUDE_HEADERS: Vec<String> = std::env::var("INCLUDE_HEADERS")
.ok().map(|x| x
.split(',')
.map(|s| s.to_string())
.collect()).unwrap_or_default();
}
pub fn build_headers(
headers: &HeaderMap,
include_header: Option<String>,
include_all_headers: bool,
) -> HashMap<String, Box<RawValue>> {
let mut selected_headers = HashMap::new();
if include_all_headers {
for (k, v) in headers.iter() {
selected_headers.insert(
k.to_string(),
to_raw_value(&v.to_str().unwrap_or("").to_string()),
);
}
} else {
let whitelist = include_header
.map(|s| s.split(",").map(|s| s.to_string()).collect::<Vec<_>>())
.unwrap_or_default();
whitelist
.iter()
.chain(INCLUDE_HEADERS.iter())
.for_each(|h| {
if let Some(v) = headers.get(h) {
selected_headers.insert(
h.to_string().to_lowercase().replace('-', "_"),
to_raw_value(&v.to_str().unwrap_or("").to_string()),
);
}
});
}
selected_headers
}
pub fn build_query(
query: Option<&str>,
include_query: Option<String>,
include_all_query: bool,
) -> HashMap<String, Box<RawValue>> {
let Some(query) = query else {
return HashMap::new();
};
if include_all_query {
let queries =
serde_urlencoded::from_str::<HashMap<String, String>>(&query).unwrap_or_default();
queries
.into_iter()
.map(|(k, v)| (k, to_raw_value(&v)))
.collect()
} else {
let parse_query_args = include_query
.map(|s| s.split(",").map(|p| p.to_string()).collect::<Vec<_>>())
.unwrap_or_default();
let mut args = HashMap::new();
if !parse_query_args.is_empty() {
let queries =
serde_urlencoded::from_str::<HashMap<String, String>>(&query).unwrap_or_default();
parse_query_args.iter().for_each(|h| {
if let Some(v) = queries.get(h) {
args.insert(h.to_string(), to_raw_value(v));
}
});
}
args
}
}
fn restructure_cloudevents_metadata(
mut p: HashMap<String, Box<RawValue>>,
) -> Result<HashMap<String, Box<RawValue>>, Error> {
let data = p
.remove("data")
.unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
let str = data.to_string();
let wrap_body = str.len() > 0 && str.chars().next().unwrap() != '{';
if wrap_body {
let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
.unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
let mut hm = HashMap::new();
hm.insert("body".to_string(), args);
hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
Ok(hm)
} else {
let mut hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
.unwrap_or_else(HashMap::new);
hm.insert("WEBHOOK__METADATA__".to_string(), to_raw_value(&p));
Ok(hm)
}
}
impl WebhookArgs {
async fn from_json(
mut metadata: WebhookArgsMetadata,
use_raw: bool,
str: String,
) -> Result<Self, Error> {
if use_raw {
metadata.raw_string = Some(str.clone());
}
let no_hashmap = str.len() > 0 && str.chars().next().unwrap() != '{';
if no_hashmap {
let args = serde_json::from_str::<Option<Box<RawValue>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
.unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
Ok(Self { body: Body::NoHashMap(args), metadata })
} else {
let hm = serde_json::from_str::<Option<HashMap<String, Box<JsonRawValue>>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid json: {}", e)))?
.unwrap_or_else(HashMap::new);
Ok(Self { body: Body::HashMap(hm), metadata })
}
}
async fn from_ce_json(
mut metadata: WebhookArgsMetadata,
use_raw: bool,
str: String,
) -> Result<Self, Error> {
if use_raw {
metadata.raw_string = Some(str.clone());
}
let hm = serde_json::from_str::<HashMap<String, Box<RawValue>>>(&str)
.map_err(|e| Error::BadRequest(format!("invalid cloudevents+json: {}", e)))?;
let hm = restructure_cloudevents_metadata(hm)?;
Ok(Self { body: Body::HashMap(hm), metadata })
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_cloudevents_json_payload() {
let r1 = r#"
{
"specversion" : "1.0",
"type" : "com.example.someevent",
"source" : "/mycontext",
"subject": null,
"id" : "C234-1234-1234",
"time" : "2018-04-05T17:31:00Z",
"comexampleextension1" : "value",
"comexampleothervalue" : 5,
"datacontenttype" : "application/json",
"data" : {
"appinfoA" : "abc",
"appinfoB" : 123,
"appinfoC" : true
}
}
"#;
let r2 = r#"
{
"specversion" : "1.0",
"type" : "com.example.someevent",
"source" : "/mycontext",
"subject": null,
"id" : "C234-1234-1234",
"time" : "2018-04-05T17:31:00Z",
"comexampleextension1" : "value",
"comexampleothervalue" : 5,
"datacontenttype" : "application/json",
"data" : 1.5
}
"#;
let metadata = WebhookArgsMetadata::default();
let a1 = WebhookArgs::from_ce_json(metadata.clone(), false, r1.to_string())
.await
.expect("Failed to parse the cloudevent");
let a2 = WebhookArgs::from_ce_json(metadata.clone(), false, r2.to_string())
.await
.expect("Failed to parse the cloudevent");
match a1.body {
Body::HashMap(body) => {
body.get("WEBHOOK__METADATA__").expect(
"CloudEvents should generate a neighboring `webhook-metadata` field in PushArgs",
);
}
_ => panic!("Expected a HashMap"),
}
match a2.body {
Body::HashMap(body) => {
assert_eq!(
body
.get("body")
.expect("Cloud events with a data field with no wrapping curly brackets should be inside of a `body` field in PushArgs")
.to_string(),
"1.5"
);
}
_ => panic!("Expected a HashMap"),
}
}
}
backend/windmill-api/src/capture.rs
+3 -70
View File
@@ -82,12 +82,11 @@ use windmill_common::{
error::{JsonResult, Result},
triggers::{RunnableFormat, RunnableFormatVersion, TriggerKind},
utils::{not_found_if_none, paginate, Pagination, RunnableKind, StripPath},
worker::{to_raw_value, CLOUD_HOSTED},
worker::to_raw_value,
};
use windmill_queue::{PushArgs, PushArgsOwned};
const KEEP_LAST: i64 = 20;
pub fn workspaced_service() -> Router {
Router::new()
@@ -802,73 +801,7 @@ async fn get_capture_trigger_config_and_owner<T: DeserializeOwned>(
))
}
async fn clear_captures_history(db: &DB, w_id: &str) -> Result<()> {
if *CLOUD_HOSTED {
/* Retain only KEEP_LAST most recent captures in this workspace. */
sqlx::query!(
r#"
DELETE FROM
capture
WHERE
workspace_id = $1
AND created_at <= (
SELECT
created_at
FROM
capture
WHERE
workspace_id = $1
ORDER BY
created_at DESC
OFFSET $2
LIMIT 1
)
"#,
&w_id,
KEEP_LAST,
)
.execute(db)
.await?;
}
Ok(())
}
pub async fn insert_capture_payload(
db: &DB,
w_id: &str,
path: &str,
is_flow: bool,
trigger_kind: &TriggerKind,
main_args: PushArgsOwned,
preprocessor_args: PushArgsOwned,
owner: &str,
) -> Result<()> {
sqlx::query!(
r#"
INSERT INTO
capture (
workspace_id, path, is_flow, trigger_kind, main_args, preprocessor_args, created_by
)
VALUES (
$1, $2, $3, $4, $5, $6, $7
)
"#,
&w_id,
path,
is_flow,
trigger_kind as &TriggerKind,
SqlxJson(PushArgs { args: &main_args.args, extra: main_args.extra }) as SqlxJson<PushArgs>,
SqlxJson(PushArgs { args: &preprocessor_args.args, extra: preprocessor_args.extra })
as SqlxJson<PushArgs>,
owner,
)
.execute(db)
.await?;
clear_captures_history(db, &w_id).await?;
Ok(())
}
pub use windmill_triggers::capture_ext::insert_capture_payload;
async fn webhook_payload(
Extension(db): Extension<DB>,
backend/windmill-api/src/folders.rs
+2 -18
View File
@@ -306,24 +306,8 @@ pub async fn is_owner_api(
Ok(Json(is_owner(&authed, &name)))
}
pub fn is_owner(ApiAuthed { is_admin, folders, .. }: &ApiAuthed, name: &str) -> bool {
if *is_admin {
true
} else {
folders.into_iter().any(|x| x.0 == name && x.2)
}
}
pub fn require_is_owner(authed: &ApiAuthed, name: &str) -> Result<()> {
if is_owner(authed, name) {
Ok(())
} else {
Err(windmill_common::error::Error::NotAuthorized(format!(
"You are not owner of the folder {}",
name
)))
}
}
pub use windmill_api_auth::permissions::is_folder_owner as is_owner;
pub use windmill_api_auth::permissions::require_is_folder_owner as require_is_owner;
async fn update_folder(
authed: ApiAuthed,
backend/windmill-api/src/job_ops_impl.rs
+4 -44
View File
@@ -23,8 +23,9 @@ use windmill_common::{
utils::StripPath,
DB,
};
use windmill_common::jobs::RunJobQuery;
use windmill_queue::PushArgsOwned;
use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream, RunJobQuery};
use windmill_triggers::jobs_ext::{JobOps, JobUpdateSSEStream};
pub struct JobOpsImpl;
@@ -46,29 +47,8 @@ impl JobOps for JobOpsImpl {
Option<bool>,
Option<sqlx::Transaction<'static, Postgres>>,
)> {
// Convert RunJobQuery from windmill-triggers to the windmill-api version
let api_run_query = crate::jobs::RunJobQuery {
scheduled_for: run_query.scheduled_for,
scheduled_in_secs: run_query.scheduled_in_secs,
parent_job: run_query.parent_job,
root_job: run_query.root_job,
invisible_to_owner: run_query.invisible_to_owner,
queue_limit: run_query.queue_limit,
payload: run_query.payload,
job_id: run_query.job_id,
tag: run_query.tag,
timeout: run_query.timeout,
cache_ttl: run_query.cache_ttl,
cache_ignore_s3_path: run_query.cache_ignore_s3_path,
skip_preprocessor: run_query.skip_preprocessor,
poll_delay_ms: run_query.poll_delay_ms,
memory_id: run_query.memory_id,
trigger_external_id: run_query.trigger_external_id,
service_name: run_query.service_name,
suspended_mode: run_query.suspended_mode,
};
crate::jobs::push_script_job_by_path_into_queue(
authed, db, tx_o, user_db, w_id, script_path, api_run_query, args, trigger,
authed, db, tx_o, user_db, w_id, script_path, run_query, args, trigger,
)
.await
}
@@ -89,28 +69,8 @@ impl JobOps for JobOpsImpl {
Option<String>,
Option<sqlx::Transaction<'static, Postgres>>,
)> {
let api_run_query = crate::jobs::RunJobQuery {
scheduled_for: run_query.scheduled_for,
scheduled_in_secs: run_query.scheduled_in_secs,
parent_job: run_query.parent_job,
root_job: run_query.root_job,
invisible_to_owner: run_query.invisible_to_owner,
queue_limit: run_query.queue_limit,
payload: run_query.payload,
job_id: run_query.job_id,
tag: run_query.tag,
timeout: run_query.timeout,
cache_ttl: run_query.cache_ttl,
cache_ignore_s3_path: run_query.cache_ignore_s3_path,
skip_preprocessor: run_query.skip_preprocessor,
poll_delay_ms: run_query.poll_delay_ms,
memory_id: run_query.memory_id,
trigger_external_id: run_query.trigger_external_id,
service_name: run_query.service_name,
suspended_mode: run_query.suspended_mode,
};
crate::jobs::push_flow_job_by_path_into_queue(
authed, db, tx_o, user_db, w_id, flow_path, api_run_query, args, trigger,
authed, db, tx_o, user_db, w_id, flow_path, run_query, args, trigger,
)
.await
}
backend/windmill-api/src/jobs.rs
+46 -111
View File
@@ -35,8 +35,8 @@ use windmill_common::error::JsonResult;
use windmill_common::flow_conversations::add_message_to_conversation_tx;
use windmill_common::flow_status::{JobResult, RestartedFrom};
use windmill_common::jobs::{
check_tag_available_for_workspace_internal, format_completed_job_result, format_result,
DynamicInput, JobTriggerKind, RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE,
format_completed_job_result, format_result, DynamicInput, JobTriggerKind,
RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE,
};
use windmill_common::runnable_settings::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings,
@@ -1775,60 +1775,38 @@ pub struct ListableCompletedJob {
pub args: Option<serde_json::Value>,
}
#[derive(Debug, Deserialize, Clone, Default)]
pub struct RunJobQuery {
pub scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
pub scheduled_in_secs: Option<i64>,
pub parent_job: Option<Uuid>,
pub root_job: Option<Uuid>,
pub invisible_to_owner: Option<bool>,
pub queue_limit: Option<i64>,
pub payload: Option<String>,
pub job_id: Option<Uuid>,
pub tag: Option<String>,
pub timeout: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub skip_preprocessor: Option<bool>,
pub poll_delay_ms: Option<u64>,
pub memory_id: Option<Uuid>,
pub trigger_external_id: Option<String>,
pub service_name: Option<String>,
pub suspended_mode: Option<bool>,
pub use windmill_common::jobs::RunJobQuery;
async fn run_query_get_scheduled_for(
run_query: &RunJobQuery,
db: &DB,
) -> error::Result<Option<chrono::DateTime<chrono::Utc>>> {
if let Some(scheduled_for) = run_query.scheduled_for {
Ok(Some(scheduled_for))
} else if let Some(scheduled_in_secs) = run_query.scheduled_in_secs {
let now = now_from_db(db).await?;
Ok(Some(
now + chrono::Duration::try_seconds(scheduled_in_secs).unwrap_or_default(),
))
} else {
Ok(None)
}
}
impl RunJobQuery {
async fn get_scheduled_for<'c>(
&self,
db: &DB,
) -> error::Result<Option<chrono::DateTime<chrono::Utc>>> {
if let Some(scheduled_for) = self.scheduled_for {
Ok(Some(scheduled_for))
} else if let Some(scheduled_in_secs) = self.scheduled_in_secs {
let now = now_from_db(db).await?;
Ok(Some(
now + chrono::Duration::try_seconds(scheduled_in_secs).unwrap_or_default(),
))
} else {
Ok(None)
}
}
fn run_query_payload_as_args(run_query: &RunJobQuery) -> error::Result<HashMap<String, Box<RawValue>>> {
let payload_r = run_query.payload.clone().map(decode_payload).map(|x| {
x.map_err(|e| {
error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}"))
})
});
fn payload_as_args(&self) -> error::Result<HashMap<String, Box<RawValue>>> {
let payload_r = self.payload.clone().map(decode_payload).map(|x| {
x.map_err(|e| {
error::Error::internal_err(format!("Impossible to decode query payload: {e:#?}"))
})
});
let payload_as_args = if let Some(payload) = payload_r {
payload?
} else {
HashMap::new()
};
let payload_as_args = if let Some(payload) = payload_r {
payload?
} else {
HashMap::new()
};
Ok(payload_as_args)
}
Ok(payload_as_args)
}
#[derive(Deserialize, Clone)]
@@ -3869,33 +3847,10 @@ pub fn add_raw_string(
return args;
}
pub async fn check_tag_available_for_workspace(
db: &DB,
w_id: &str,
tag: &Option<String>,
authed: &ApiAuthed,
) -> error::Result<()> {
if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) {
let tags = get_scope_tags(authed);
check_tag_available_for_workspace_internal(&db, w_id, tag, &authed.email, tags).await
} else {
Ok(())
}
}
pub use windmill_triggers::jobs_ext::check_tag_available_for_workspace;
#[cfg(feature = "enterprise")]
pub async fn check_license_key_valid() -> error::Result<()> {
use windmill_common::ee_oss::LICENSE_KEY_VALID;
let valid = *LICENSE_KEY_VALID.read().await;
if !valid {
return Err(Error::BadRequest(
"License key is not valid. Go to your superadmin settings to update your license key."
.to_string(),
));
}
Ok(())
}
pub use windmill_common::ee_oss::check_license_key_valid;
use windmill_common::flows::InputTransform;
@@ -4323,7 +4278,7 @@ pub async fn run_flow<'c>(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let return_tx = tx_o.is_some();
@@ -4627,7 +4582,7 @@ pub async fn restart_flow(
.map(|json| PushArgs { args: &json.0, extra: None })
.unwrap_or_else(|| PushArgs::from(&ehm));
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into());
@@ -4775,7 +4730,7 @@ pub async fn push_script_job_by_path_into_queue<'c>(
run_query.skip_preprocessor,
)
.await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -4953,7 +4908,7 @@ pub async fn run_workflow_as_code(
extra.insert(ENTRYPOINT_OVERRIDE.to_string(), to_raw_value(&entrypoint));
let args = PushArgs { args: &task.args.unwrap_or_else(HashMap::new), extra: Some(extra) };
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(tag).or(Some(job.tag));
@@ -5324,27 +5279,7 @@ pub async fn run_wait_result(
result_to_response(result, success)
}
pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> {
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1",
job_uuid,
)
.execute(db)
.await?;
Ok(())
}
pub use windmill_common::jobs::delete_job_metadata_after_use;
pub async fn check_queue_too_long(db: &DB, queue_limit: Option<i64>) -> error::Result<()> {
if let Some(limit) = queue_limit {
@@ -5475,7 +5410,7 @@ pub async fn run_wait_result_job_by_path_get(
return Ok(Json(serde_json::json!("")).into_response());
}
let payload_as_args = run_query.payload_as_args()?;
let payload_as_args = run_query_payload_as_args(&run_query)?;
let mut args = args.process_args(&authed, &db, &w_id, None).await?;
args.body = args::Body::HashMap(payload_as_args);
@@ -5580,7 +5515,7 @@ pub async fn run_wait_result_flow_by_path_get(
if method == http::Method::HEAD {
return Ok(Json(serde_json::json!("")).into_response());
}
let payload_as_args = run_query.payload_as_args()?;
let payload_as_args = run_query_payload_as_args(&run_query)?;
let mut args = args.process_args(&authed, &db, &w_id, None).await?;
args.body = args::Body::HashMap(payload_as_args);
@@ -5958,7 +5893,7 @@ pub async fn stream_job(
is_get: bool,
) -> error::Result<Response> {
let args = if is_get {
let payload_as_args = run_query.payload_as_args()?;
let payload_as_args = run_query_payload_as_args(&run_query)?;
let mut args = args.process_args(&authed, &db, &w_id, None).await?;
args.body = args::Body::HashMap(payload_as_args);
@@ -6140,7 +6075,7 @@ pub async fn run_wait_result_flow_by_version_get(
return Ok(Json(serde_json::json!("")).into_response());
}
let payload_as_args = run_query.payload_as_args()?;
let payload_as_args = run_query_payload_as_args(&run_query)?;
let mut args = args.process_args(&authed, &db, &w_id, None).await?;
args.body = args::Body::HashMap(payload_as_args);
@@ -6243,7 +6178,7 @@ async fn run_preview_script(
));
}
require_path_read_access_for_preview(&authed, &preview.path)?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(preview.tag.clone());
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into());
@@ -6456,7 +6391,7 @@ async fn run_bundle_preview_script(
.and_then(|s| BundleFormat::from_string(&s))
.unwrap_or(BundleFormat::Cjs);
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(preview.tag.clone());
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
let ltx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into());
@@ -7063,7 +6998,7 @@ async fn run_preview_flow_job(
));
}
require_path_read_access_for_preview(&authed, &raw_flow.path)?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(raw_flow.tag.clone());
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into());
@@ -7263,7 +7198,7 @@ async fn run_dynamic_select(
}
}
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into());
let (uuid, tx) = push(
@@ -7387,7 +7322,7 @@ pub async fn run_job_by_hash_inner(
cache_ttl = Some(run_query_cache_ttl);
cache_ignore_s3_path = run_query.cache_ignore_s3_path;
}
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let scheduled_for = run_query_get_scheduled_for(&run_query, &db).await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
backend/windmill-api/src/lib.rs
+1 -1
View File
@@ -74,7 +74,7 @@ pub mod agent_workers_ee;
mod agent_workers_oss;
mod ai;
mod apps;
pub mod args;
pub use windmill_triggers::args_ext as args;
mod assets;
mod audit;
pub mod auth;
backend/windmill-api/src/users.rs
+6 -209
View File
@@ -8,7 +8,7 @@
#![allow(non_snake_case)]
use sqlx::{PgConnection, Postgres, Transaction};
use sqlx::{Postgres, Transaction};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
@@ -161,19 +161,7 @@ pub async fn maybe_refresh_folders(
}
}
pub fn get_scope_tags(authed: &ApiAuthed) -> Option<Vec<&str>> {
authed.scopes.as_ref()?.iter().find_map(|s| {
if s.starts_with("if_jobs:filter_tags:") {
Some(
s.trim_start_matches("if_jobs:filter_tags:")
.split(",")
.collect::<Vec<_>>(),
)
} else {
None
}
})
}
pub use windmill_api_auth::scopes::get_scope_tags;
#[derive(Clone, Debug)]
pub struct OptAuthed(pub Option<ApiAuthed>);
@@ -319,27 +307,7 @@ pub struct TruncatedToken {
pub scopes: Option<Vec<String>>,
}
#[derive(Deserialize)]
pub struct NewToken {
pub label: Option<String>,
pub expiration: Option<chrono::DateTime<chrono::Utc>>,
pub impersonate_email: Option<String>,
pub scopes: Option<Vec<String>>,
pub workspace_id: Option<String>,
}
#[cfg(feature = "native_trigger")]
impl NewToken {
pub fn new(
label: Option<String>,
expiration: Option<chrono::DateTime<chrono::Utc>>,
impersonate_email: Option<String>,
scopes: Option<Vec<String>>,
workspace_id: Option<String>,
) -> NewToken {
NewToken { label, expiration, impersonate_email, scopes, workspace_id }
}
}
pub use windmill_api_auth::tokens::NewToken;
#[derive(Deserialize)]
pub struct Login {
@@ -852,35 +820,7 @@ pub async fn is_owner_of_path(
}
}
pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
let splitted = path.split("/").collect::<Vec<&str>>();
if splitted[0] == "u" {
if splitted[1] == authed.username {
Ok(())
} else {
Err(Error::BadRequest(format!(
"only the owner {} is authorized to perform this operation",
splitted[1]
)))
}
} else if splitted[0] == "f" {
crate::folders::require_is_owner(authed, splitted[1])
} else {
Err(Error::BadRequest(format!(
"Not recognized path kind: {}",
path
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be owner of an empty path"
)))
}
}
pub use windmill_api_auth::permissions::require_owner_of_path;
/// Checks that a user has at least read access to the path for preview jobs.
/// This prevents privilege escalation where a user could run preview code
@@ -939,95 +879,7 @@ pub fn require_path_read_access_for_preview(
}
}
pub fn get_perm_in_extra_perms_for_authed(
v: serde_json::Value,
authed: &ApiAuthed,
) -> Option<bool> {
match v {
serde_json::Value::Object(obj) => {
let mut keys = vec![format!("u/{}", authed.username)];
for g in authed.groups.iter() {
keys.push(format!("g/{}", g));
}
let mut res = None;
for k in keys {
if let Some(v) = obj.get(&k) {
if let Some(v) = v.as_bool() {
if v {
return Some(true);
}
res = Some(v);
}
}
}
res
}
_ => None,
}
}
pub async fn require_is_writer(
authed: &ApiAuthed,
path: &str,
w_id: &str,
db: DB,
query: &str,
kind: &str,
) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
if require_owner_of_path(authed, path).is_ok() {
return Ok(());
}
if path.starts_with("f/") && path.split('/').count() >= 2 {
let folder = path.split('/').nth(1).unwrap();
let extra_perms = sqlx::query_scalar!(
"SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2",
folder,
w_id
)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let is_folder_writer =
get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false);
if is_folder_writer {
return Ok(());
}
}
}
let extra_perms = sqlx::query_scalar(query)
.bind(path)
.bind(w_id)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let perm = get_perm_in_extra_perms_for_authed(perms, authed);
match perm {
Some(true) => Ok(()),
Some(false) => Err(Error::BadRequest(format!(
"User {} is not a writer of {kind} path {path}",
authed.username
))),
None => Err(Error::BadRequest(format!(
"User {} has neither read or write permission on {kind} {path}",
authed.username
))),
}
} else {
Err(Error::BadRequest(format!(
"{path} does not exist yet and user {} is not an owner of the parent folder",
authed.username
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be writer of an empty path"
)))
}
}
pub use windmill_api_auth::permissions::require_is_writer;
async fn whois(
Extension(db): Extension<DB>,
Path((w_id, username)): Path<(String, String)>,
@@ -2107,62 +1959,7 @@ pub async fn create_session_token<'c>(
Ok(token)
}
pub async fn create_token_internal(
tx: &mut PgConnection,
db: &DB,
authed: &ApiAuthed,
token_config: NewToken,
) -> Result<String> {
let token = rd_string(32);
let is_super_admin = sqlx::query_scalar!(
"SELECT super_admin FROM password WHERE email = $1",
authed.email
)
.fetch_optional(&mut *tx)
.await?
.unwrap_or(false);
if *CLOUD_HOSTED {
let nb_tokens =
sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email)
.fetch_one(db)
.await?;
if nb_tokens.unwrap_or(0) >= 10000 {
return Err(Error::BadRequest(
"You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit"
.to_string(),
));
}
}
sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin, scopes, workspace_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
token,
authed.email,
token_config.label,
token_config.expiration,
is_super_admin,
token_config.scopes.as_ref().map(|x| x.as_slice()),
token_config.workspace_id,
)
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
authed,
"users.token.create",
ActionKind::Create,
&"global",
Some(&token[0..10]),
None,
)
.instrument(tracing::info_span!("token", email = &authed.email))
.await?;
Ok(token)
}
pub use windmill_api_auth::tokens::create_token_internal;
async fn create_token(
Extension(db): Extension<DB>,
backend/windmill-common/src/ee_oss.rs
+12
View File
@@ -122,3 +122,15 @@ pub async fn low_disk_alerts(
) {
// Implementation is not open source
}
#[cfg(all(feature = "enterprise", not(feature = "private")))]
pub async fn check_license_key_valid() -> error::Result<()> {
let valid = *LICENSE_KEY_VALID.read().await;
if !valid {
return Err(error::Error::BadRequest(
"License key is not valid. Go to your superadmin settings to update your license key."
.to_string(),
));
}
Ok(())
}
backend/windmill-common/src/jobs.rs
+44
View File
@@ -917,3 +917,47 @@ pub struct WorkerInternalServerInlineUtils {
// The server cannot call the worker functions directly because they are independent crates
pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell<WorkerInternalServerInlineUtils> =
OnceCell::new();
#[derive(Debug, Deserialize, Clone, Default)]
pub struct RunJobQuery {
pub scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
pub scheduled_in_secs: Option<i64>,
pub parent_job: Option<Uuid>,
pub root_job: Option<Uuid>,
pub invisible_to_owner: Option<bool>,
pub queue_limit: Option<i64>,
pub payload: Option<String>,
pub job_id: Option<Uuid>,
pub tag: Option<String>,
pub timeout: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub skip_preprocessor: Option<bool>,
pub poll_delay_ms: Option<u64>,
pub memory_id: Option<Uuid>,
pub trigger_external_id: Option<String>,
pub service_name: Option<String>,
pub suspended_mode: Option<bool>,
}
pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> {
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1",
job_uuid,
)
.execute(db)
.await?;
Ok(())
}
backend/windmill-triggers/src/flow_ext.rs
+1 -3
View File
@@ -9,10 +9,8 @@
use windmill_api_auth::ApiAuthed;
use windmill_common::{error::Result, DB};
use crate::require_is_writer_internal;
pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> {
require_is_writer_internal(
windmill_api_auth::permissions::require_is_writer(
authed,
path,
w_id,
backend/windmill-triggers/src/jobs_ext.rs
+6 -100
View File
@@ -13,7 +13,6 @@
//! implementation at startup via `set_ops()`.
use axum::response::Response;
use serde::Deserialize;
use serde_json::value::RawValue;
use sqlx::Postgres;
use std::sync::Arc;
@@ -21,99 +20,21 @@ use uuid::Uuid;
use windmill_api_auth::{ApiAuthed, OptTokened};
use windmill_common::{
db::UserDB,
error::{self, Error},
jobs::check_tag_available_for_workspace_internal,
error,
triggers::TriggerMetadata,
utils::StripPath,
DB,
};
use windmill_queue::PushArgsOwned;
#[derive(Debug, Deserialize, Clone, Default)]
pub struct RunJobQuery {
pub scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
pub scheduled_in_secs: Option<i64>,
pub parent_job: Option<Uuid>,
pub root_job: Option<Uuid>,
pub invisible_to_owner: Option<bool>,
pub queue_limit: Option<i64>,
pub payload: Option<String>,
pub job_id: Option<Uuid>,
pub tag: Option<String>,
pub timeout: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub skip_preprocessor: Option<bool>,
pub poll_delay_ms: Option<u64>,
pub memory_id: Option<Uuid>,
pub trigger_external_id: Option<String>,
pub service_name: Option<String>,
pub suspended_mode: Option<bool>,
}
pub fn get_scope_tags(authed: &ApiAuthed) -> Option<Vec<&str>> {
authed.scopes.as_ref()?.iter().find_map(|s| {
if s.starts_with("if_jobs:filter_tags:") {
Some(
s.trim_start_matches("if_jobs:filter_tags:")
.split(",")
.collect::<Vec<_>>(),
)
} else {
None
}
})
}
pub async fn check_tag_available_for_workspace(
db: &DB,
w_id: &str,
tag: &Option<String>,
authed: &ApiAuthed,
) -> error::Result<()> {
if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) {
let tags = get_scope_tags(authed);
check_tag_available_for_workspace_internal(db, w_id, tag, &authed.email, tags).await
} else {
Ok(())
}
}
pub async fn delete_job_metadata_after_use(db: &DB, job_uuid: Uuid) -> Result<(), Error> {
sqlx::query!(
"UPDATE v2_job SET args = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE v2_job_completed SET result = '{}'::jsonb WHERE id = $1",
job_uuid,
)
.execute(db)
.await?;
sqlx::query!(
"UPDATE job_logs SET logs = '##DELETED##' WHERE job_id = $1",
job_uuid,
)
.execute(db)
.await?;
Ok(())
}
// Re-export shared types/functions from windmill-common
pub use windmill_common::jobs::{delete_job_metadata_after_use, RunJobQuery};
#[cfg(feature = "enterprise")]
pub async fn check_license_key_valid() -> error::Result<()> {
use windmill_common::ee_oss::LICENSE_KEY_VALID;
pub use windmill_common::ee_oss::check_license_key_valid;
let valid = *LICENSE_KEY_VALID.read().await;
if !valid {
return Err(Error::BadRequest(
"License key is not valid. Go to your superadmin settings to update your license key."
.to_string(),
));
}
Ok(())
}
// Re-export scope helpers from windmill-api-auth
pub use windmill_api_auth::scopes::{check_tag_available_for_workspace, get_scope_tags};
/// Trait for complex job operations that stay in windmill-api.
/// windmill-api provides the implementation at startup via `set_ops()`.
@@ -236,21 +157,6 @@ pub trait JobOps: Send + Sync + 'static {
) -> error::Result<(Option<bool>, Option<windmill_common::s3_helpers::ObjectStoreResource>)>;
}
// Re-export types used by SSE streaming
#[derive(Debug, Clone, serde::Serialize)]
pub struct JobUpdate {
#[serde(skip_serializing_if = "Option::is_none")]
pub running: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub completed: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub new_logs: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub log_offset: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub mem_peak: Option<i32>,
}
#[derive(Debug, serde::Serialize)]
pub enum JobUpdateSSEStream {
Update(serde_json::Value),
backend/windmill-triggers/src/lib.rs
-151
View File
@@ -17,154 +17,3 @@ pub mod jobs_ext;
pub mod resource_ext;
pub mod script_ext;
pub mod user_ext;
// Shared permission helpers used by flow_ext, script_ext, and native_triggers
use windmill_api_auth::ApiAuthed;
use windmill_common::error::{Error, Result};
use windmill_common::DB;
/// Check if the user is an owner of the given path.
pub fn require_owner_of_path(authed: &ApiAuthed, path: &str) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
let splitted = path.split("/").collect::<Vec<&str>>();
if splitted[0] == "u" {
if splitted[1] == authed.username {
Ok(())
} else {
Err(Error::BadRequest(format!(
"only the owner {} is authorized to perform this operation",
splitted[1]
)))
}
} else if splitted[0] == "f" {
require_is_folder_owner(authed, splitted[1])
} else {
Err(Error::BadRequest(format!(
"Not recognized path kind: {}",
path
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be owner of an empty path"
)))
}
}
fn is_folder_owner(
ApiAuthed { is_admin, folders, .. }: &ApiAuthed,
name: &str,
) -> bool {
if *is_admin {
true
} else {
folders.into_iter().any(|x| x.0 == name && x.2)
}
}
fn require_is_folder_owner(authed: &ApiAuthed, name: &str) -> Result<()> {
if is_folder_owner(authed, name) {
Ok(())
} else {
Err(Error::NotAuthorized(format!(
"You are not owner of the folder {}",
name
)))
}
}
fn get_perm_in_extra_perms_for_authed(
v: serde_json::Value,
authed: &ApiAuthed,
) -> Option<bool> {
match v {
serde_json::Value::Object(obj) => {
let mut keys = vec![format!("u/{}", authed.username)];
for g in authed.groups.iter() {
keys.push(format!("g/{}", g));
}
let mut res = None;
for k in keys {
if let Some(v) = obj.get(&k) {
if let Some(v) = v.as_bool() {
if v {
return Some(true);
}
res = Some(v);
}
}
}
res
}
_ => None,
}
}
/// Generic require_is_writer with a configurable SQL query.
/// Used by flow_ext and script_ext.
pub async fn require_is_writer_internal(
authed: &ApiAuthed,
path: &str,
w_id: &str,
db: DB,
query: &str,
kind: &str,
) -> Result<()> {
if authed.is_admin {
return Ok(());
}
if !path.is_empty() {
if require_owner_of_path(authed, path).is_ok() {
return Ok(());
}
if path.starts_with("f/") && path.split('/').count() >= 2 {
let folder = path.split('/').nth(1).unwrap();
let extra_perms = sqlx::query_scalar!(
"SELECT extra_perms FROM folder WHERE name = $1 AND workspace_id = $2",
folder,
w_id
)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let is_folder_writer =
get_perm_in_extra_perms_for_authed(perms, authed).unwrap_or(false);
if is_folder_writer {
return Ok(());
}
}
}
let extra_perms = sqlx::query_scalar(query)
.bind(path)
.bind(w_id)
.fetch_optional(&db)
.await?;
if let Some(perms) = extra_perms {
let perm = get_perm_in_extra_perms_for_authed(perms, authed);
match perm {
Some(true) => Ok(()),
Some(false) => Err(Error::BadRequest(format!(
"User {} is not a writer of {kind} path {path}",
authed.username
))),
None => Err(Error::BadRequest(format!(
"User {} has neither read or write permission on {kind} {path}",
authed.username
))),
}
} else {
Err(Error::BadRequest(format!(
"{path} does not exist yet and user {} is not an owner of the parent folder",
authed.username
)))
}
} else {
Err(Error::BadRequest(format!(
"Cannot be writer of an empty path"
)))
}
}
backend/windmill-triggers/src/script_ext.rs
+1 -3
View File
@@ -9,10 +9,8 @@
use windmill_api_auth::ApiAuthed;
use windmill_common::{error::Result, DB};
use crate::require_is_writer_internal;
pub async fn require_is_writer(authed: &ApiAuthed, path: &str, w_id: &str, db: DB) -> Result<()> {
require_is_writer_internal(
windmill_api_auth::permissions::require_is_writer(
authed,
path,
w_id,
backend/windmill-triggers/src/user_ext.rs
+1 -88
View File
@@ -6,91 +6,4 @@
* LICENSE-AGPL for a copy of the license.
*/
use sqlx::PgConnection;
use tracing::Instrument;
use windmill_api_auth::ApiAuthed;
use windmill_audit::{audit_oss::audit_log, ActionKind};
use windmill_common::{
error::{Error, Result},
utils::rd_string,
worker::CLOUD_HOSTED,
DB,
};
#[derive(serde::Deserialize)]
pub struct NewToken {
pub label: Option<String>,
pub expiration: Option<chrono::DateTime<chrono::Utc>>,
pub impersonate_email: Option<String>,
pub scopes: Option<Vec<String>>,
pub workspace_id: Option<String>,
}
impl NewToken {
pub fn new(
label: Option<String>,
expiration: Option<chrono::DateTime<chrono::Utc>>,
impersonate_email: Option<String>,
scopes: Option<Vec<String>>,
workspace_id: Option<String>,
) -> NewToken {
NewToken { label, expiration, impersonate_email, scopes, workspace_id }
}
}
pub async fn create_token_internal(
tx: &mut PgConnection,
db: &DB,
authed: &ApiAuthed,
token_config: NewToken,
) -> Result<String> {
let token = rd_string(32);
let is_super_admin = sqlx::query_scalar!(
"SELECT super_admin FROM password WHERE email = $1",
authed.email
)
.fetch_optional(&mut *tx)
.await?
.unwrap_or(false);
if *CLOUD_HOSTED {
let nb_tokens =
sqlx::query_scalar!("SELECT COUNT(*) FROM token WHERE email = $1", &authed.email)
.fetch_one(db)
.await?;
if nb_tokens.unwrap_or(0) >= 10000 {
return Err(Error::BadRequest(
"You have reached the maximum number of tokens (10000) on cloud. Contact support@windmill.dev to increase the limit"
.to_string(),
));
}
}
sqlx::query!(
"INSERT INTO token
(token, email, label, expiration, super_admin, scopes, workspace_id)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
token,
authed.email,
token_config.label,
token_config.expiration,
is_super_admin,
token_config.scopes.as_ref().map(|x| x.as_slice()),
token_config.workspace_id,
)
.execute(&mut *tx)
.await?;
audit_log(
&mut *tx,
authed,
"users.token.create",
ActionKind::Create,
&"global",
Some(&token[0..10]),
None,
)
.instrument(tracing::info_span!("token", email = &authed.email))
.await?;
Ok(token)
}
pub use windmill_api_auth::tokens::{create_token_internal, NewToken};