From 5bc40f7d28d509c88b290e4ff45432ecd94a5e26 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 12 Nov 2022 20:17:08 +0100 Subject: [PATCH] fix(backend): apps backend v0 (#888) * progress * post merge * progress * fix * fix * fix * fix * v1 * fix openapi --- backend/Cargo.lock | 1 + .../migrations/20221024225533_apps.down.sql | 4 + backend/migrations/20221024225533_apps.up.sql | 25 + backend/windmill-api/Cargo.toml | 1 + backend/windmill-api/openapi.yaml | 263 +++++++++ backend/windmill-api/src/apps.rs | 535 ++++++++++++++++++ backend/windmill-api/src/lib.rs | 10 +- docker-compose.yml | 2 + 8 files changed, 839 insertions(+), 2 deletions(-) create mode 100644 backend/migrations/20221024225533_apps.down.sql create mode 100644 backend/migrations/20221024225533_apps.up.sql create mode 100644 backend/windmill-api/src/apps.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 24e5f1edda..17dbd7dec0 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4071,6 +4071,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", + "sha2 0.10.6", "sql-builder", "sqlx", "tempfile", diff --git a/backend/migrations/20221024225533_apps.down.sql b/backend/migrations/20221024225533_apps.down.sql new file mode 100644 index 0000000000..bb187a9f8f --- /dev/null +++ b/backend/migrations/20221024225533_apps.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here +DROP TABLE app; +DROP TABLE app_version; +DROP TYPE EXECUTION_MODE; \ No newline at end of file diff --git a/backend/migrations/20221024225533_apps.up.sql b/backend/migrations/20221024225533_apps.up.sql new file mode 100644 index 0000000000..3006fd7150 --- /dev/null +++ b/backend/migrations/20221024225533_apps.up.sql @@ -0,0 +1,25 @@ +-- Add up migration script here +CREATE TABLE app ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id), + path varchar(255) NOT NULL, + summary VARCHAR(1000) NOT NULL DEFAULT '', + policy JSONB NOT NULL, + versions BIGINT[] NOT NULL, + extra_perms JSONB NOT NULL DEFAULT '{}' +); + +CREATE TABLE app_version( + id BIGSERIAL PRIMARY KEY, + flow_id BIGINT NOT NULL, + value JSONB NOT NULL, + created_by VARCHAR(50) NOT NULL, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), + FOREIGN KEY (flow_id) REFERENCES app(id) ON DELETE CASCADE +); + +CREATE POLICY see_own ON app FOR ALL +USING (SPLIT_PART(app.path, '/', 1) = 'u' AND SPLIT_PART(app.path, '/', 2) = current_setting('session.user')); + +CREATE POLICY see_member ON app FOR ALL +USING (SPLIT_PART(app.path, '/', 1) = 'g' AND SPLIT_PART(app.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[])); diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 2e4c7aceaa..339ee0e77e 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -64,3 +64,4 @@ tokio-util.workspace = true tokio-tar.workspace = true hmac.workspace = true cookie.workspace = true +sha2.workspace = true \ No newline at end of file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 86455d0796..2e186b8d77 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2301,6 +2301,203 @@ paths: schema: type: string + /w/{workspace}/apps/list: + get: + summary: list all available apps + operationId: listApps + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Page" + - $ref: "#/components/parameters/PerPage" + - $ref: "#/components/parameters/OrderDesc" + - $ref: "#/components/parameters/CreatedBy" + - name: path_start + description: mask to filter matching starting path + in: query + schema: + type: string + - name: path_exact + description: mask to filter exact matching path + in: query + schema: + type: string + responses: + "200": + description: All available apps + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/ListableApp" + + /w/{workspace}/apps/get/p/{path}: + get: + summary: get app by path + operationId: getAppByPath + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/ScriptPath" + responses: + "200": + description: app details + content: + application/json: + schema: + $ref: "#/components/schemas/AppWithLastVersion" + + /w/{workspace}/apps/get/v/{id}: + get: + summary: get app by version + operationId: getAppByVersion + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/PathId" + responses: + "200": + description: app details + content: + application/json: + schema: + $ref: "#/components/schemas/AppWithLastVersion" + + /w/{workspace}/apps/create: + post: + summary: create app + operationId: createApp + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: new app + required: true + content: + application/json: + schema: + type: object + properties: + path: + type: string + value: {} + summary: + type: string + policy: + $ref: "#/components/schemas/Policy" + required: + - path + - value + - summary + - policy + responses: + "201": + description: app created + content: + text/plain: + schema: + type: string + + /w/{workspace}/apps/delete/{path}: + delete: + summary: delete app + operationId: deleteApp + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + responses: + "200": + description: app deleted + content: + text/plain: + schema: + type: string + + /w/{workspace}/apps/update/{path}: + post: + summary: update app + operationId: updateApp + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/ScriptPath" + requestBody: + description: update app + required: true + content: + application/json: + schema: + type: object + properties: + path: + type: string + summary: + type: string + value: {} + policy: + $ref: "#/components/schemas/Policy" + responses: + "200": + description: app updated + content: + text/plain: + schema: + type: string + + /w/{workspace}/apps/execute_component/{path}: + post: + summary: executeComponent + operationId: executeComponent + tags: + - app + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/ScriptPath" + requestBody: + description: update app + required: true + content: + application/json: + schema: + type: object + properties: + #script: script/ + #flow: flow/ + path: + type: string + args: {} + raw_code: + type: object + properties: + content: + type: string + language: + type: string + path: + type: string + required: + - content + - language + required: + - args + + responses: + "200": + description: job uuid + content: + text/plain: + schema: + type: string + + /w/{workspace}/jobs/run/f/{path}: post: summary: run flow by path @@ -4496,6 +4693,72 @@ components: - content - args + Policy: + type: object + properties: + triggerables: + type: object + additionalProperties: + type: object + execution_mode: + type: string + enum: [viewer, publisher, anonymous] + on_behalf_of: + type: string + + + ListableApp: + type: object + properties: + id: + type: integer + workspace_id: + type: string + path: + type: string + summary: + type: string + version: + type: integer + extra_perms: + type: object + additionalProperties: + type: boolean + execution_mode: + type: string + enum: [viewer, publisher, anonymous] + + AppWithLastVersion: + type: object + properties: + id: + type: integer + workspace_id: + type: string + path: + type: string + summary: + type: string + versions: + type: array + items: + type: integer + created_by: + type: string + created_at: + type: string + format: date-time + value: {} + policy: + $ref: "#/components/schemas/Policy" + execution_mode: + type: string + enum: [viewer, publisher, anonymous] + extra_perms: + type: object + additionalProperties: + type: boolean + SlackToken: type: object properties: diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs new file mode 100644 index 0000000000..fe3ff65daf --- /dev/null +++ b/backend/windmill-api/src/apps.rs @@ -0,0 +1,535 @@ +use std::collections::HashMap; + +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * 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 crate::{ + db::{UserDB, DB}, + jobs::script_path_to_payload, + users::{Authed, OptAuthed}, +}; +use axum::{ + extract::{Extension, Path, Query}, + routing::{delete, get, post}, + Json, Router, +}; +use hyper::StatusCode; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Map, Value}; +use sha2::{Digest, Sha256}; +use sql_builder::{bind::Bind, SqlBuilder}; +use sqlx::{types::Uuid, FromRow}; +use windmill_audit::{audit_log, ActionKind}; +use windmill_common::{ + error::{to_anyhow, Error, JsonResult, Result}, + users::owner_to_token_owner, + utils::{not_found_if_none, paginate, Pagination, StripPath}, +}; +use windmill_queue::{push, JobPayload, RawCode}; + +pub fn workspaced_service() -> Router { + Router::new() + .route("/list", get(list_apps)) + .route("/get/p/*path", get(get_app)) + .route("/get/v/*id", get(get_app_by_id)) + .route("/update/*path", post(update_app)) + .route("/delete/*path", delete(delete_app)) + .route("/create", post(create_app)) +} + +pub fn unauthed_service() -> Router { + Router::new().route("/execute_component/*path", post(execute_component)) +} + +#[derive(FromRow, Deserialize, Serialize)] +pub struct ListableApp { + pub id: i64, + pub workspace_id: String, + pub path: String, + pub summary: String, + pub version: i64, + pub extra_perms: serde_json::Value, + pub execution_mode: String, +} + +#[derive(FromRow, Serialize, Deserialize)] +pub struct AppVersion { + pub id: i64, + pub flow_id: Uuid, + pub value: serde_json::Value, + pub created_by: String, + pub created_at: chrono::DateTime, +} + +#[derive(Serialize, Deserialize)] +pub struct AppWithLastVersion { + pub id: i64, + pub path: String, + pub summary: String, + pub policy: serde_json::Value, + pub versions: Vec, + pub value: serde_json::Value, + pub created_by: String, + pub created_at: chrono::DateTime, + pub extra_perms: serde_json::Value, +} + +pub type StaticFields = Map; + +#[derive(Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ExecutionMode { + Anonymous, + Publisher, + Viewer, +} + +#[derive(Serialize, Deserialize)] +pub struct Policy { + pub on_behalf_of: Option, + //paths: + // - script/ + // - flow/ + // - rawscript/ + pub triggerables: HashMap, + pub execution_mode: ExecutionMode, +} + +#[derive(Deserialize)] +pub struct CreateApp { + pub path: String, + pub summary: String, + pub value: serde_json::Value, + pub policy: Policy, +} + +#[derive(Deserialize)] +pub struct EditApp { + pub path: Option, + pub summary: Option, + pub value: Option, + pub policy: Option, +} + +async fn list_apps( + authed: Authed, + Query(pagination): Query, + Extension(user_db): Extension, + Path(w_id): Path, +) -> JsonResult> { + let (per_page, offset) = paginate(pagination); + + let sqlb = SqlBuilder::select_from("app") + .fields(&[ + "id", + "workspace_id", + "path", + "summary", + "versions[array_upper(versions, 1)] as version", + "policy->>execution_mode as execution_mode", + "extra_perms", + ]) + .order_by("path", true) + .and_where("workspace_id = ?".bind(&w_id)) + .offset(offset) + .limit(per_page) + .clone(); + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let mut tx = user_db.begin(&authed).await?; + let rows = sqlx::query_as::<_, ListableApp>(&sql) + .fetch_all(&mut tx) + .await?; + + tx.commit().await?; + + Ok(Json(rows)) +} + +async fn get_app( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> JsonResult { + let path = path.to_path(); + let mut tx = user_db.begin(&authed).await?; + + let app_o = sqlx::query_as!( + AppWithLastVersion, + "SELECT app.id, app.path, app.summary, app.versions, app.policy, + app.extra_perms, app_version.value, + app_version.created_at, app_version.created_by from app, app_version + WHERE app.path = $1 AND app.workspace_id = $2 AND app_version.id = app.versions[array_upper(app.versions, 1)]", + path.to_owned(), + &w_id + ) + .fetch_optional(&mut tx) + .await?; + tx.commit().await?; + + let app = not_found_if_none(app_o, "App", path)?; + Ok(Json(app)) +} + +async fn get_app_by_id( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, id)): Path<(String, i64)>, +) -> JsonResult { + let mut tx = user_db.begin(&authed).await?; + + let app_o = sqlx::query_as!( + AppWithLastVersion, + "SELECT app.id, app.path, app.summary, app.versions, app.policy, + app.extra_perms, app_version.value, + app_version.created_at, app_version.created_by from app, app_version + WHERE app_version.id = $1 AND app.id = app_version.flow_id AND app.workspace_id = $2", + id, + &w_id + ) + .fetch_optional(&mut tx) + .await?; + tx.commit().await?; + + let app = not_found_if_none(app_o, "App", id.to_string())?; + Ok(Json(app)) +} + +async fn create_app( + authed: Authed, + Extension(user_db): Extension, + Path(w_id): Path, + Json(app): Json, +) -> Result<(StatusCode, String)> { + let mut tx = user_db.begin(&authed).await?; + + let id = sqlx::query_scalar!( + "INSERT INTO app + (workspace_id, path, summary, policy) + VALUES ($1, $2, $3, $4) RETURNING id", + w_id, + app.path, + app.summary, + json!(app.policy), + ) + .fetch_one(&mut tx) + .await?; + + let v_id = sqlx::query_scalar!( + "INSERT INTO app_version + (flow_id, value, created_by) + VALUES ($1, $2, $3) RETURNING id", + id, + app.value, + authed.username, + ) + .fetch_one(&mut tx) + .await?; + + sqlx::query!( + "UPDATE app SET versions = array_append(versions, $1) WHERE id = $2", + v_id, + id + ) + .execute(&mut tx) + .await?; + + audit_log( + &mut tx, + &authed.username, + "apps.create", + ActionKind::Create, + &w_id, + Some(&app.path), + None, + ) + .await?; + tx.commit().await?; + + Ok((StatusCode::CREATED, format!("app {} created", app.path))) +} + +async fn delete_app( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> Result { + let path = path.to_path(); + let mut tx = user_db.begin(&authed).await?; + + sqlx::query!( + "DELETE FROM app WHERE path = $1 AND workspace_id = $2", + path, + w_id + ) + .execute(&mut tx) + .await?; + audit_log( + &mut tx, + &authed.username, + "apps.delete", + ActionKind::Delete, + &w_id, + Some(path), + None, + ) + .await?; + tx.commit().await?; + + Ok(format!("app {} deleted", path)) +} + +async fn update_app( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, + Json(ns): Json, +) -> Result { + use sql_builder::prelude::*; + + let path = path.to_path(); + + let mut tx = user_db.begin(&authed).await?; + + let npath = if ns.policy.is_some() || ns.path.is_some() || ns.summary.is_some() { + let mut sqlb = SqlBuilder::update_table("app"); + sqlb.and_where_eq("path", "?".bind(&path)); + sqlb.and_where_eq("workspace_id", "?".bind(&w_id)); + + if let Some(npath) = &ns.path { + sqlb.set_str("path", npath); + } + + if let Some(nsummary) = &ns.summary { + sqlb.set_str("summary", nsummary); + } + + if let Some(npolicy) = ns.policy { + sqlb.set( + "policy", + &format!( + "'{}'", + serde_json::to_string(&json!(npolicy)).map_err(|e| { + Error::BadRequest(format!("failed to serialize policy: {}", e)) + })? + ), + ); + } + + sqlb.returning("path"); + + let sql = sqlb.sql().map_err(|e| Error::InternalErr(e.to_string()))?; + let npath_o: Option = sqlx::query_scalar(&sql).fetch_optional(&mut tx).await?; + not_found_if_none(npath_o, "App", path)? + } else { + "".to_string() + }; + if let Some(nvalue) = &ns.value { + let flow_id = sqlx::query_scalar!( + "SELECT id FROM app WHERE path = $1 AND workspace_id = $2", + path, + w_id + ) + .fetch_one(&mut tx) + .await?; + + let v_id = sqlx::query_scalar!( + "INSERT INTO app_version + (flow_id, value, created_by) + VALUES ($1, $2, $3) RETURNING id", + flow_id, + nvalue, + authed.username, + ) + .fetch_one(&mut tx) + .await?; + + sqlx::query!( + "UPDATE app SET versions = array_append(versions, $1) WHERE path = $2 AND workspace_id = $3", + v_id, + path, + w_id + ) + .execute(&mut tx) + .await?; + } + audit_log( + &mut tx, + &authed.username, + "apps.update", + ActionKind::Update, + &w_id, + Some(&npath), + None, + ) + .await?; + tx.commit().await?; + + Ok(format!("app {} updated (npath: {:?})", path, npath)) +} + +#[derive(Debug, Deserialize)] +pub struct ExecuteApp { + pub args: Map, + // - script: script/ + // - flow: flow/ + pub path: Option, + pub raw_code: Option, +} + +fn digest(code: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(code); + let result = hasher.finalize(); + format!("rawscript/{:x}", result) +} + +async fn execute_component( + OptAuthed(opt_authed): OptAuthed, + Extension(db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, + Json(payload): Json, +) -> Result { + match (payload.path.is_some(), payload.raw_code.is_some()) { + (true, true) => { + return Err(Error::BadRequest( + "path or raw_code is required".to_string(), + )) + } + (false, false) => { + return Err(Error::BadRequest( + "path and raw_code cannot be set at the same time".to_string(), + )) + } + _ => {} + }; + + let path = path.to_path(); + let mut tx = db.begin().await?; + + let policy_o = sqlx::query_scalar!( + "SELECT policy from app WHERE path = $1 AND workspace_id = $2", + path, + &w_id + ) + .fetch_optional(&mut tx) + .await?; + + let policy = not_found_if_none(policy_o, "App", path)?; + + let policy = serde_json::from_value::(policy).map_err(to_anyhow)?; + + let (username, permissioned_as) = match policy.execution_mode { + ExecutionMode::Anonymous => { + let username = opt_authed + .map(|a| a.username) + .unwrap_or_else(|| "anonymous".to_string()); + let permissioned_as = policy + .on_behalf_of + .as_ref() + .ok_or_else(|| { + Error::BadRequest( + "on_behalf_of is missing in the app policy and is required for anonymous execution" + .to_string(), + ) + })? + .to_string(); + (username, permissioned_as) + } + ExecutionMode::Publisher => { + let username = opt_authed.map(|a| a.username).ok_or_else(|| { + Error::BadRequest("publisher execution mode requires authentication".to_string()) + })?; + let permissioned_as = policy + .on_behalf_of + .as_ref() + .ok_or_else(|| { + Error::BadRequest( + "on_behalf_of is missing in the app policy and is required for publisher execution" + .to_string(), + ) + })? + .to_string(); + (username, permissioned_as) + } + ExecutionMode::Viewer => { + let username = opt_authed + .map(|a| a.username) + .ok_or_else(|| Error::BadRequest("".to_string()))?; + (username.clone(), owner_to_token_owner(&username, false)) + } + }; + + let (job_payload, args) = match &payload { + ExecuteApp { args, raw_code: Some(raw_code), path: None } => { + let content = &raw_code.content; + let payload = JobPayload::Code(raw_code.clone()); + let path = digest(content); + let args = build_args(policy, path, args)?; + (payload, args) + } + ExecuteApp { args, raw_code: None, path: Some(path) } => { + let payload = if path.starts_with("script/") { + script_path_to_payload(path.strip_prefix("script/").unwrap(), &mut tx, &w_id) + .await? + } else if path.starts_with("flow/") { + JobPayload::Flow(path.strip_prefix("flow/").unwrap().to_string()) + } else { + return Err(Error::BadRequest(format!( + "path must start with script/ or flow/ (got {})", + path + ))); + }; + let args = build_args(policy, path.to_string(), args)?; + (payload, args) + } + _ => unreachable!(), + }; + + let (uuid, tx) = push( + tx, + &w_id, + job_payload, + Some(args), + &username, + permissioned_as, + None, + None, + None, + false, + false, + ) + .await?; + + tx.commit().await?; + Ok(uuid.to_string()) +} + +fn build_args( + policy: Policy, + path: String, + args: &Map, +) -> Result> { + let static_args = policy + .triggerables + .get(&path) + .map(|x| x.clone()) + .or_else(|| { + if matches!(policy.execution_mode, ExecutionMode::Viewer) { + Some(Map::new()) + } else { + None + } + }) + .ok_or_else(|| { + Error::BadRequest(format!("path {} is not allowed in the app policy", path)) + })?; + let mut args = args.clone(); + for (k, v) in static_args { + args.insert(k.to_string(), v.to_owned()); + } + Ok(args) +} diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 13ad53ea3e..fde76ffbb7 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -20,9 +20,10 @@ use crate::{ db::UserDB, oauth2::{build_oauth_clients, SlackVerifier}, tracing_init::{MyMakeSpan, MyOnResponse}, - users::Authed, + users::{Authed, OptAuthed}, }; +mod apps; mod audit; mod capture; mod db; @@ -117,7 +118,8 @@ pub async fn run_server( .nest("/acls", granular_acls::workspaced_service()) .nest("/workspaces", workspaces::workspaced_service()) .nest("/flows", flows::workspaced_service()) - .nest("/capture", capture::workspaced_service()), + .nest("/capture", capture::workspaced_service()) + .nest("/apps", apps::workspaced_service()), ) .nest("/workspaces", workspaces::global_service()) .nest( @@ -130,6 +132,10 @@ pub async fn run_server( .nest("/schedules", schedule::global_service()) .route_layer(from_extractor::()) .route_layer(from_extractor::()) + .nest( + "/w/:workspace_id/apps", + apps::unauthed_service().layer(from_extractor::()), + ) .nest("/w/:workspace_id/jobs", jobs::global_service()) .nest("/w/:workspace_id/capture", capture::global_service()) .nest( diff --git a/docker-compose.yml b/docker-compose.yml index 775e025695..4fea5d7c3a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -19,6 +19,7 @@ services: interval: 10s timeout: 5s retries: 5 + windmill: image: ghcr.io/windmill-labs/windmill:main privileged: true @@ -51,6 +52,7 @@ services: restart: unless-stopped ports: - 3001:3001 + caddy: image: caddy:2.5.2-alpine restart: unless-stopped