From 244ab5737db7eef99b7cc95f39b594075cef938b Mon Sep 17 00:00:00 2001 From: Simon Kassing Date: Wed, 22 Jul 2026 14:47:49 +0200 Subject: [PATCH] pipeline-manager: outline OIDC changes Draft of the SQL migration and the database operations that would need to be implemented. Signed-off-by: Simon Kassing --- .../migrations/V35__oidc_tenant.sql | 58 +++ crates/pipeline-manager/src/db/error.rs | 57 +++ crates/pipeline-manager/src/db/operations.rs | 1 + .../src/db/operations/oidc.rs | 413 ++++++++++++++++++ crates/pipeline-manager/src/db/types.rs | 1 + crates/pipeline-manager/src/db/types/oidc.rs | 62 +++ 6 files changed, 592 insertions(+) create mode 100644 crates/pipeline-manager/migrations/V35__oidc_tenant.sql create mode 100644 crates/pipeline-manager/src/db/operations/oidc.rs create mode 100644 crates/pipeline-manager/src/db/types/oidc.rs diff --git a/crates/pipeline-manager/migrations/V35__oidc_tenant.sql b/crates/pipeline-manager/migrations/V35__oidc_tenant.sql new file mode 100644 index 00000000000..5c6dc8e082f --- /dev/null +++ b/crates/pipeline-manager/migrations/V35__oidc_tenant.sql @@ -0,0 +1,58 @@ +-- Allowed OIDC provider. +CREATE TABLE oidc_provider ( + issuer VARCHAR PRIMARY KEY NOT NULL, -- Issuer URL for the provider (uniquely identifies it) + subject_filter VARCHAR NOT NULL, -- Filter which subjects of the provider are allowed to authenticate + client_id VARCHAR NOT NULL, -- Client identifier Feldera uses when contacting the provider + CONSTRAINT unique_issuer UNIQUE (issuer) +); + +-- A user is created when they are first authenticated through the OIDC provider. +-- An OIDC provider cannot be deleted until all users are deleted. +CREATE TABLE oidc_user ( + id UUID PRIMARY KEY NOT NULL, -- Unique identifier for the user + issuer VARCHAR NOT NULL, -- OIDC issuer + subject VARCHAR NOT NULL, -- OIDC subject + CONSTRAINT unique_issuer_subject UNIQUE (issuer, subject), + FOREIGN KEY (issuer) REFERENCES oidc_provider(issuer) ON DELETE CASCADE +); + +-- A tenant is the conceptual entity which owns zero or more pipelines. +-- +-- If the tenant is the user-dedicated one, it will have the format: `user-dedicated-`. +-- It is not possible to create a tenant with this name unless it's during user creation. +-- +-- The table already exists, as such we need to do alterations to get it to the following: +-- CREATE TABLE tenant ( +-- id UUID PRIMARY KEY NOT NULL, -- Unique identifier for the tenant +-- name VARCHAR NOT NULL, +-- CONSTRAINT unique_name UNIQUE (name) +-- ); +ALTER TABLE tenant ADD COLUMN name VARCHAR NULL; +UPDATE tenant SET name = CONCAT(tenant, '-', provider); +ALTER TABLE tenant DROP CONSTRAINT tenant_tenant_provider_key; +ALTER TABLE tenant DROP COLUMN provider; +ALTER TABLE tenant DROP COLUMN tenant; +ALTER TABLE tenant ALTER COLUMN name SET NOT NULL; +ALTER TABLE tenant ADD CONSTRAINT unique_name UNIQUE (name); +ALTER TABLE tenant ALTER COLUMN id SET NOT NULL; + +-- A user belongs to zero or more tenants. +-- A tenant has zero or more users. +CREATE TABLE user_tenant ( + user_id UUID NOT NULL, -- User identifier + tenant_id UUID NOT NULL, -- Tenant identifier + role VARCHAR NOT NULL, -- "read", "write", "owner" + PRIMARY KEY (user_id, tenant_id), + FOREIGN KEY (user_id) REFERENCES oidc_user(id) ON DELETE CASCADE, + FOREIGN KEY (tenant_id) REFERENCES tenant(id) ON DELETE CASCADE +); + +-- The api_key table only needs to have its `scopes` removed, and replaced with `role`. +ALTER TABLE api_key DROP COLUMN scopes; +ALTER TABLE api_key ADD COLUMN role VARCHAR NULL; +UPDATE api_key SET role = 'write'; +ALTER TABLE api_key ALTER COLUMN id SET NOT NULL; + +-- All pipelines belonging to a tenant need to be removed before a tenant can be deleted. +ALTER TABLE pipeline DROP CONSTRAINT pipeline_tenant_id_fkey; +ALTER TABLE pipeline ADD CONSTRAINT pipeline_tenant_id_fkey FOREIGN KEY (tenant_id) REFERENCES tenant(id) ON DELETE RESTRICT; diff --git a/crates/pipeline-manager/src/db/error.rs b/crates/pipeline-manager/src/db/error.rs index d38e7222567..c81395db35b 100644 --- a/crates/pipeline-manager/src/db/error.rs +++ b/crates/pipeline-manager/src/db/error.rs @@ -1,5 +1,6 @@ use crate::cluster_monitor::{MONITOR_RETENTION_HOURS, MONITOR_RETENTION_NUM}; use crate::db::types::monitor::{ClusterMonitorEventId, PipelineMonitorEventId}; +use crate::db::types::oidc::UserId; use crate::db::types::pipeline::PipelineId; use crate::db::types::program::ProgramStatus; use crate::db::types::resources_status::{ResourcesDesiredStatus, ResourcesStatus}; @@ -258,6 +259,27 @@ pub enum DBError { NoPipelineMonitorEventsAvailable, LockTookTooLong, DeadlockDetected, + UnknownProvider { + issuer: String, + }, + InvalidIssuer { + error: String, + }, + InvalidSubjectFilter { + error: String, + }, + InvalidClientId { + error: String, + }, + UnknownUser { + id: UserId, + }, + InvalidSubject { + error: String, + }, + InvalidTenantName { + error: String, + }, } impl DBError { @@ -827,6 +849,27 @@ impl Display for DBError { DBError::DeadlockDetected => { write!(f, "A deadlock was detected while performing the operation. Try this operation again later. Please also file a bug report, as this error should not happen.") } + DBError::UnknownProvider { issuer } => { + write!(f, "Unknown OIDC provider with issuer: {issuer}") + } + DBError::InvalidIssuer { error } => { + write!(f, "Invalid OIDC provider issuer: {error}") + } + DBError::InvalidSubjectFilter { error } => { + write!(f, "Invalid OIDC provider subject filter: {error}") + } + DBError::InvalidClientId { error } => { + write!(f, "Invalid OIDC provider client identifier: {error}") + } + DBError::UnknownUser { id } => { + write!(f, "Unknown OIDC user with identifier: {id}") + } + DBError::InvalidSubject { error } => { + write!(f, "Invalid OIDC subject: {error}") + } + DBError::InvalidTenantName { error } => { + write!(f, "Invalid tenant name: {error}") + } } } } @@ -948,6 +991,13 @@ impl DetailedError for DBError { Self::NoPipelineMonitorEventsAvailable => Cow::from("NoPipelineMonitorEventsAvailable"), Self::LockTookTooLong => Cow::from("LockTookTooLong"), Self::DeadlockDetected => Cow::from("DeadlockDetected"), + Self::UnknownProvider { .. } => Cow::from("UnknownProvider"), + Self::InvalidIssuer { .. } => Cow::from("InvalidIssuer"), + Self::InvalidSubjectFilter { .. } => Cow::from("InvalidSubjectFilter"), + Self::InvalidClientId { .. } => Cow::from("InvalidClientId"), + Self::UnknownUser { .. } => Cow::from("UnknownUser"), + Self::InvalidSubject { .. } => Cow::from("InvalidSubject"), + Self::InvalidTenantName { .. } => Cow::from("InvalidTenantName"), } } } @@ -1033,6 +1083,13 @@ impl ResponseError for DBError { Self::NoPipelineMonitorEventsAvailable => StatusCode::NOT_FOUND, Self::LockTookTooLong => StatusCode::SERVICE_UNAVAILABLE, Self::DeadlockDetected => StatusCode::SERVICE_UNAVAILABLE, + Self::UnknownProvider { .. } => StatusCode::NOT_FOUND, + Self::InvalidIssuer { .. } => StatusCode::BAD_REQUEST, + Self::InvalidSubjectFilter { .. } => StatusCode::BAD_REQUEST, + Self::InvalidClientId { .. } => StatusCode::BAD_REQUEST, + Self::UnknownUser { .. } => StatusCode::NOT_FOUND, + Self::InvalidSubject { .. } => StatusCode::BAD_REQUEST, + Self::InvalidTenantName { .. } => StatusCode::BAD_REQUEST, } } diff --git a/crates/pipeline-manager/src/db/operations.rs b/crates/pipeline-manager/src/db/operations.rs index acab3b47409..e96c2cec65c 100644 --- a/crates/pipeline-manager/src/db/operations.rs +++ b/crates/pipeline-manager/src/db/operations.rs @@ -14,6 +14,7 @@ pub mod api_key; pub mod cluster_monitor; pub mod connectivity; +pub mod oidc; pub mod pipeline; pub mod pipeline_monitor; mod pipeline_parsing; diff --git a/crates/pipeline-manager/src/db/operations/oidc.rs b/crates/pipeline-manager/src/db/operations/oidc.rs new file mode 100644 index 00000000000..bf2cbb36706 --- /dev/null +++ b/crates/pipeline-manager/src/db/operations/oidc.rs @@ -0,0 +1,413 @@ +use crate::db::error::DBError; +use crate::db::operations::utils::maybe_unique_violation; +use crate::db::types::oidc::{Provider, Tenant, User, UserId}; +use crate::db::types::tenant::TenantId; +use deadpool_postgres::Transaction; +use tokio_postgres::Row; +use uuid::Uuid; + +/// Validates that the provided issuer follows basic format rules. +/// Specifically, it cannot be empty. +// TODO: additional validation, e.g. that it must start with `https://`, though that would +// interfere with local testing +fn validate_issuer(issuer: &str) -> Result<(), DBError> { + if issuer.is_empty() { + return Err(DBError::InvalidIssuer { + error: "cannot be empty".to_string(), + }); + } + Ok(()) +} + +/// Validates that the provided subject follows basic format rules (cannot be empty) and passes +/// the provided filter. +// TODO: more precise subject filter formatting +fn validate_subject(subject: &str, subject_filter: &str) -> Result<(), DBError> { + if subject.is_empty() { + return Err(DBError::InvalidSubject { + error: "cannot be empty".to_string(), + }); + } + // Only two filters are supported: + // - '*': any subject matches + // - Else, it does exact matching + if subject_filter == "*" || subject == subject_filter { + Ok(()) + } else { + Err(DBError::InvalidSubject { + error: format!("subject '{subject}' does not match filter '{subject_filter}'"), + }) + } +} + +/// Validates that the provided subject filter follows basic format rules. +/// - It cannot be empty +// TODO: additional filter rules, like the present of wildcard asterisks +fn validate_subject_filter(subject_filter: &str) -> Result<(), DBError> { + // TODO: check presence of wildcard asterisks + if subject_filter.is_empty() { + return Err(DBError::InvalidSubjectFilter { + error: "cannot be empty".to_string(), + }); + } + Ok(()) +} + +/// Validates that the provided client identifier follows basic format rules. +/// Specifically, that it cannot be empty. +fn validate_client_id(client_id: &str) -> Result<(), DBError> { + if client_id.is_empty() { + return Err(DBError::InvalidClientId { + error: "cannot be empty".to_string(), + }); + } + Ok(()) +} + +/// Validates that the provided tenant name follows basic format rules. +/// Specifically, that it cannot be empty. +// TODO: length restriction? +// TODO: if it is user dedicated, the name must match `user-dedicated-`. +// if it is not, it must not match that. +fn validate_tenant_name(name: &str, is_user_dedicated: bool) -> Result<(), DBError> { + if name.is_empty() { + return Err(DBError::InvalidTenantName { + error: "cannot be empty".to_string(), + }); + } + Ok(()) +} + +/// Parses the provided row from the `oidc_provider` table into a [`Provider`]. +pub fn parse_provider_row(row: &Row) -> Result { + Ok(Provider { + issuer: row.get("issuer"), + subject_filter: row.get("subject_filter"), + client_id: row.get("client_id"), + }) +} + +/// Lists all OIDC providers. +pub(crate) async fn list_providers(txn: &Transaction<'_>) -> Result, DBError> { + let stmt = txn + .prepare_cached( + "SELECT issuer, subject_filter, client_id + FROM oidc_provider + ORDER BY issuer", + ) + .await?; + let rows: Vec = txn.query(&stmt, &[]).await?; + let mut result = Vec::with_capacity(rows.len()); + for row in rows { + result.push(parse_provider_row(&row)?); + } + Ok(result) +} + +/// Creates a new OIDC provider. +pub(crate) async fn new_provider( + txn: &Transaction<'_>, + issuer: &str, + subject_filter: &str, + client_id: &str, +) -> Result<(), DBError> { + // Field validation + validate_issuer(issuer)?; + validate_subject_filter(subject_filter)?; + validate_client_id(client_id)?; + + // Query + let stmt = txn + .prepare_cached( + "INSERT INTO oidc_provider (issuer, subject_filter, client_id) + VALUES ($1, $2, $3)", + ) + .await?; + txn.execute( + &stmt, + &[ + &issuer, // $1: issuer + &subject_filter, // $2: subject_filter + &client_id, // $3: client_id + ], + ) + .await + .map_err(maybe_unique_violation)?; // TODO: additional constraint violations + Ok(()) +} + +/// Updates an existing OIDC provider. +pub(crate) async fn update_provider( + txn: &Transaction<'_>, + issuer: &str, + new_subject_filter: Option, + new_client_id: Option, +) -> Result<(), DBError> { + // Field validation + validate_issuer(issuer)?; + if let Some(new_subject_filter) = new_subject_filter.as_ref() { + validate_subject_filter(new_subject_filter)?; + } + if let Some(new_client_id) = new_client_id.as_ref() { + validate_client_id(new_client_id)?; + } + + // Query + let stmt = txn + .prepare_cached( + "UPDATE oidc_provider + SET subject_filter = COALESCE($1, subject_filter), + client_id = COALESCE($2, client_id), + WHERE issuer = $3", + ) + .await?; + txn.execute( + &stmt, + &[ + &new_subject_filter, // $1: subject_filter + &new_client_id, // $2: client_id + &issuer, // $3: issuer + ], + ) + .await + .map_err(maybe_unique_violation)?; // TODO: additional constraint violations + Ok(()) +} + +/// Deletes an existing OIDC provider. +pub(crate) async fn delete_provider(txn: &Transaction<'_>, issuer: &str) -> Result<(), DBError> { + let stmt = txn + .prepare_cached("DELETE FROM oidc_provider WHERE issuer = $1") + .await?; + let res = txn.execute(&stmt, &[&issuer]).await?; + if res > 0 { + Ok(()) + } else { + Err(DBError::UnknownProvider { + issuer: issuer.to_string(), + }) + } +} + +/// Parses the provided row from the `oidc_user` table into a [`User`]. +pub fn parse_user_row(row: &Row) -> Result { + Ok(User { + id: UserId(row.get("id")), + issuer: row.get("issuer"), + subject: row.get("subject"), + }) +} + +/// Lists all OIDC users, ordered by (issuer, subject). +pub(crate) async fn list_users(txn: &Transaction<'_>) -> Result, DBError> { + let stmt = txn + .prepare_cached( + "SELECT id, issuer, subject + FROM oidc_user + ORDER BY (issuer, subject)", + ) + .await?; + let rows: Vec = txn.query(&stmt, &[]).await?; + let mut result = Vec::with_capacity(rows.len()); + for row in rows { + result.push(parse_user_row(&row)?); + } + Ok(result) +} + +/// Creates a new OIDC user if it does not yet exist. This is called when it has passed +/// authentication and is in search for the user identifier that can be used to retrieve +/// its tenants. +pub(crate) async fn new_user_if_not_exists( + txn: &Transaction<'_>, + issuer: &str, + subject_filter: &str, // TODO: will this be provided, or do we need to separately fetch from `oidc_provider` table? + subject: &str, + new_user_id: Uuid, + new_tenant_id: Uuid, +) -> Result { + // Field validation + validate_issuer(issuer)?; + validate_subject(subject, subject_filter)?; + + // Check if user already exists + let stmt = txn + .prepare_cached( + "SELECT id, issuer, subject + FROM oidc_user + WHERE issuer = $1 AND subject = $2", + ) + .await?; + if let Some(row) = txn.query_opt(&stmt, &[&issuer, &subject]).await? { + return parse_user_row(&row); + } + + // If it does not, insert it + let stmt = txn + .prepare_cached( + "INSERT INTO oidc_user (id, issuer, subject) + VALUES ($1, $2, $3)", + ) + .await?; + txn.execute( + &stmt, + &[ + &new_user_id, // $1: id + &issuer, // $2: issuer + &subject, // $3: subject + ], + ) + .await + .map_err(maybe_unique_violation)?; // TODO: additional constraint violations + + // Create a dedicated tenant for the user + new_tenant( + txn, + &format!("user-dedicated-{new_user_id}"), + new_tenant_id, + true, + ) + .await?; + + // Map the dedicated tenant to the user + let stmt = txn + .prepare_cached( + "INSERT INTO user_tenant (user_id, tenant_id, role) + VALUES ($1, $2, 'owner')", + ) + .await?; + txn.execute( + &stmt, + &[ + &new_user_id, // $1: user_id + &new_tenant_id, // $2: tenant_id + ], + ) + .await?; + + Ok(User { + id: UserId(new_user_id), + issuer: issuer.to_string(), + subject: subject.to_string(), + }) +} + +/// Deletes an existing OIDC user. +pub(crate) async fn delete_user(txn: &Transaction<'_>, user_id: UserId) -> Result<(), DBError> { + let stmt = txn + .prepare_cached("DELETE FROM oidc_user WHERE user_id = $1") + .await?; + let res = txn.execute(&stmt, &[&user_id.0]).await?; + if res > 0 { + Ok(()) + } else { + Err(DBError::UnknownUser { id: user_id }) + } +} + +// TODO: fn grant_user_tenant_access() +// TODO: fn revoke_user_tenant_access() + +/// Parses the provided row from the `tenant` table into a [`Tenant`]. +pub fn parse_tenant_row(row: &Row) -> Result { + Ok(Tenant { + id: TenantId(row.get("id")), + name: row.get("name"), + }) +} + +/// Lists all tenants. +pub(crate) async fn list_tenants(txn: &Transaction<'_>) -> Result, DBError> { + let stmt = txn + .prepare_cached( + "SELECT id, name + FROM tenant + ORDER BY name", + ) + .await?; + let rows: Vec = txn.query(&stmt, &[]).await?; + let mut result = Vec::with_capacity(rows.len()); + for row in rows { + result.push(parse_tenant_row(&row)?); + } + Ok(result) +} + +/// Creates a new tenant. +pub(crate) async fn new_tenant( + txn: &Transaction<'_>, + name: &str, + new_id: Uuid, + is_user_dedicated: bool, +) -> Result<(), DBError> { + // Field validation + validate_tenant_name(name, is_user_dedicated)?; + + // Query + let stmt = txn + .prepare_cached( + "INSERT INTO tenant (id, name) + VALUES ($1, $2)", + ) + .await?; + txn.execute( + &stmt, + &[ + &new_id, // $1: id + &name, // $2: name + ], + ) + .await?; + Ok(()) +} + +/// Updates an existing tenant. +pub(crate) async fn update_tenant( + txn: &Transaction<'_>, + tenant_id: TenantId, + new_name: Option, +) -> Result<(), DBError> { + // Field validation + // TODO: deny name update if it is a user-dedicated tenant + + // Query + let stmt = txn + .prepare_cached( + "UPDATE tenant + SET name = COALESCE($1, name) + WHERE id = $2", + ) + .await?; + txn.execute( + &stmt, + &[ + &new_name, // $1: name + &tenant_id.0, // $2: tenant_id + ], + ) + .await + .map_err(maybe_unique_violation)?; // TODO: additional constraint violations + Ok(()) +} + +/// Deletes an existing tenant. +pub(crate) async fn delete_tenant( + txn: &Transaction<'_>, + tenant_id: TenantId, +) -> Result<(), DBError> { + // TODO: check that the tenant is not dedicated to a user + let stmt = txn + .prepare_cached("DELETE FROM tenant WHERE id = $1") + .await?; + let res = txn.execute(&stmt, &[&tenant_id.0]).await?; + if res > 0 { + Ok(()) + } else { + Err(DBError::UnknownTenant { tenant_id }) + } +} + +#[cfg(test)] +mod test { + // TODO: unit test that validation works +} diff --git a/crates/pipeline-manager/src/db/types.rs b/crates/pipeline-manager/src/db/types.rs index 2306ee4ab3d..5d20f87b66a 100644 --- a/crates/pipeline-manager/src/db/types.rs +++ b/crates/pipeline-manager/src/db/types.rs @@ -4,6 +4,7 @@ pub mod api_key; pub mod combined_status; pub mod monitor; +pub mod oidc; pub mod pipeline; pub mod program; pub mod resources_status; diff --git a/crates/pipeline-manager/src/db/types/oidc.rs b/crates/pipeline-manager/src/db/types/oidc.rs new file mode 100644 index 00000000000..4c3da614a83 --- /dev/null +++ b/crates/pipeline-manager/src/db/types/oidc.rs @@ -0,0 +1,62 @@ +use crate::db::types::tenant::TenantId; +use serde::{Deserialize, Serialize}; +use std::fmt; +use std::fmt::Display; +use utoipa::ToSchema; +use uuid::Uuid; + +/// Allowed OIDC provider. In addition to the ones retrieved from the database, the root OIDC +/// provider is supplied through command-line arguments. +#[derive(Clone, ToSchema)] +pub struct Provider { + /// OIDC issuer URL. + pub issuer: String, + + /// Only subjects that satisfy this filter are allowed. + /// - "*": all subjects are allowed + /// - "example": exactly one subject is allowed: "example" + /// - "example*: all subjects that have "example" as prefix are allowed + pub subject_filter: String, + + /// OIDC client identifier for the Feldera instance. + pub client_id: String, +} + +/// User identifier. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Ord, PartialOrd, Serialize, Deserialize, ToSchema)] +#[cfg_attr(test, derive(proptest_derive::Arbitrary))] +#[repr(transparent)] +#[serde(transparent)] +pub struct UserId( + #[cfg_attr(test, proptest(strategy = "crate::db::test::limited_uuid()"))] pub Uuid, +); +impl Display for UserId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +/// User authenticated through its OIDC provider. +pub struct User { + /// Unique identifier for the user. + pub id: UserId, + + /// OIDC issuer. + pub issuer: String, + + /// OIDC subject. + pub subject: String, +} + +/// Tenant. +pub struct Tenant { + /// Unique identifier for the tenant. + pub id: TenantId, + + /// Tenant name, which is system-generated if it is user-dedicated, and provided by the user + /// otherwise. + /// + /// If the tenant is the user-dedicated one, it will have the format: `user-dedicated-`. + /// It is not possible to create a tenant with this name format unless it's during user creation. + pub name: String, +}