feat: add tag filtering to external JWT authentication (#4425)

* feat: tag filtering jwt ext auth

* move tags to scopes

* fix symlink

* update ee ref
This commit is contained in:
HugoCasa
2024-09-25 11:21:19 +02:00
committed by GitHub
parent 118ffc57fb
commit 1feccd493e
5 changed files with 32 additions and 11 deletions
+1 -1
View File
@@ -1 +1 @@
e093d51a219ce4ff0562a02db24ec402554a1f05
3d37b6c31155265d8d026ae9d6ced0b433078f87
@@ -144,7 +144,7 @@ async fn get_concurrent_intervals(
Query(iq): Query<ExtendedJobsParams>,
Query(lq): Query<ListCompletedQuery>,
) -> JsonResult<ExtendedJobs> {
check_scopes(&authed, || format!("listjobs"))?;
check_scopes(&authed, || format!("jobs:listjobs"))?;
if lq.success.is_some() && lq.running.is_some_and(|x| x) {
return Err(error::Error::BadRequest(
+28 -9
View File
@@ -33,6 +33,7 @@ use crate::add_webhook_allowed_origin;
use crate::concurrency_groups::join_concurrency_key;
use crate::db::ApiAuthed;
use crate::users::get_scope_tags;
use crate::utils::content_plain;
use crate::{
db::DB,
@@ -1248,13 +1249,18 @@ pub fn list_queue_jobs_query(
lq: &ListQueueQuery,
fields: &[&str],
join_outstanding_wait_times: bool,
tags: Option<Vec<&str>>,
) -> SqlBuilder {
let sqlb = SqlBuilder::select_from("queue")
let mut sqlb = SqlBuilder::select_from("queue")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.limit(1000)
.clone();
if let Some(tags) = tags {
sqlb.and_where_in("tag", &tags.iter().map(|x| quote(x)).collect::<Vec<_>>());
}
filter_list_queue_query(sqlb, lq, w_id, join_outstanding_wait_times)
}
@@ -1310,6 +1316,7 @@ async fn list_queue_jobs(
"workspace_id",
],
false,
get_scope_tags(&authed),
)
.sql()?;
let mut tx = user_db.begin(&authed).await?;
@@ -1498,6 +1505,10 @@ async fn list_filtered_uuids(
sqlb.and_where_is_null("schedule_path");
if let Some(tags) = get_scope_tags(&authed) {
sqlb.and_where_in("tag", &tags.iter().map(|x| quote(x)).collect::<Vec<_>>());
}
sqlb = filter_list_queue_query(sqlb, &lq, w_id.as_str(), false);
let sql = sqlb.query()?;
@@ -1557,7 +1568,7 @@ async fn list_jobs(
Query(lq): Query<ListCompletedQuery>,
Extension(_api_list_jobs_query_duration): Extension<Option<Histo>>,
) -> error::JsonResult<Vec<Job>> {
check_scopes(&authed, || format!("listjobs"))?;
check_scopes(&authed, || format!("jobs:listjobs"))?;
let (per_page, offset) = paginate(pagination);
let lqc = lq.clone();
@@ -1575,6 +1586,7 @@ async fn list_jobs(
&ListCompletedQuery { order_desc: Some(true), ..lqc },
UnifiedJob::completed_job_fields(),
true,
get_scope_tags(&authed),
))
} else {
None
@@ -1590,6 +1602,7 @@ async fn list_jobs(
&ListQueueQuery { order_desc: Some(true), ..lq.into() },
UnifiedJob::queued_job_fields(),
true,
get_scope_tags(&authed),
);
if let Some(sqlc) = sqlc {
@@ -1640,7 +1653,7 @@ pub async fn resume_suspended_flow_as_owner(
Path((_w_id, flow_id)): Path<(String, Uuid)>,
QueryOrBody(value): QueryOrBody<serde_json::Value>,
) -> error::Result<StatusCode> {
check_scopes(&authed, || format!("resumeflow"))?;
check_scopes(&authed, || format!("jobs:resumeflow"))?;
let value = value.unwrap_or(serde_json::Value::Null);
let mut tx = db.begin().await?;
@@ -3847,7 +3860,7 @@ async fn run_preview_script(
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
check_scopes(&authed, || format!("runscript"))?;
check_scopes(&authed, || format!("jobs:runscript"))?;
if authed.is_operator {
return Err(error::Error::NotAuthorized(
"Operators cannot run preview jobs for security reasons".to_string(),
@@ -3917,7 +3930,7 @@ async fn run_bundle_preview_script(
check_license_key_valid().await?;
check_scopes(&authed, || format!("runscript"))?;
check_scopes(&authed, || format!("jobs:runscript"))?;
if authed.is_operator {
return Err(error::Error::NotAuthorized(
"Operators cannot run preview jobs for security reasons".to_string(),
@@ -4403,7 +4416,7 @@ async fn run_preview_flow_job(
Query(run_query): Query<RunJobQuery>,
Json(raw_flow): Json<PreviewFlow>,
) -> error::Result<(StatusCode, String)> {
check_scopes(&authed, || format!("runflow"))?;
check_scopes(&authed, || format!("jobs:runflow"))?;
if authed.is_operator {
return Err(error::Error::NotAuthorized(
"Operators cannot run preview jobs for security reasons".to_string(),
@@ -4843,14 +4856,19 @@ pub fn list_completed_jobs_query(
lq: &ListCompletedQuery,
fields: &[&str],
join_outstanding_wait_times: bool,
tags: Option<Vec<&str>>,
) -> SqlBuilder {
let sqlb = SqlBuilder::select_from("completed_job")
let mut sqlb = SqlBuilder::select_from("completed_job")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.offset(offset)
.limit(per_page)
.clone();
if let Some(tags) = tags {
sqlb.and_where_in("tag", &tags.iter().map(|x| quote(x)).collect::<Vec<_>>());
}
filter_list_completed_query(sqlb, lq, w_id, join_outstanding_wait_times)
}
#[derive(Deserialize, Clone)]
@@ -4895,7 +4913,7 @@ async fn list_completed_jobs(
Query(pagination): Query<Pagination>,
Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<ListableCompletedJob>> {
check_scopes(&authed, || format!("listjobs"))?;
check_scopes(&authed, || format!("jobs:listjobs"))?;
let (per_page, offset) = paginate(pagination);
@@ -4937,6 +4955,7 @@ async fn list_completed_jobs(
"'CompletedJob' as type",
],
false,
get_scope_tags(&authed),
)
.sql()?;
let mut tx = user_db.begin(&authed).await?;
@@ -5161,7 +5180,7 @@ async fn delete_completed_job<'a>(
Extension(user_db): Extension<UserDB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
check_scopes(&authed, || format!("deletejob"))?;
check_scopes(&authed, || format!("jobs:deletejob"))?;
let mut tx = user_db.begin(&authed).await?;
+1
View File
@@ -25,6 +25,7 @@ pub struct JWTAuthClaims {
pub workspace_id: String,
pub exp: usize,
pub job_id: Option<String>,
pub scopes: Option<Vec<String>>,
}
#[derive(Deserialize)]
+1
View File
@@ -194,6 +194,7 @@ pub async fn create_token_for_owner(
exp: (chrono::Utc::now() + chrono::Duration::seconds(expires_in as i64)).timestamp()
as usize,
job_id: Some(job_id.to_string()),
scopes: None,
};
let token = jsonwebtoken::encode(