diff --git a/api/src/routes/v2/deployment_permissions.rs b/api/src/routes/v2/deployment_permissions.rs index b9864e1..a9fd91b 100644 --- a/api/src/routes/v2/deployment_permissions.rs +++ b/api/src/routes/v2/deployment_permissions.rs @@ -3,6 +3,7 @@ use crate::result::ApiResult; use actix_web::{HttpResponse, delete, get, post, web}; use platz_auth::ApiIdentity; use platz_db::{ + AccessScope, diesel_pagination::{Paginated, PaginationParams}, schema::deployment_permission::{ DeploymentPermission, DeploymentPermissionFilters, NewDeploymentPermission, @@ -28,12 +29,14 @@ use uuid::Uuid; )] #[get("/deployment-permissions")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok().json( - DeploymentPermission::all_filtered(filters.into_inner(), pagination.into_inner()).await?, + DeploymentPermission::all_filtered(filters.into_inner(), pagination.into_inner(), &scope) + .await?, )) } @@ -53,8 +56,9 @@ async fn get_all( ), )] #[get("/deployment-permissions/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(DeploymentPermission::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(DeploymentPermission::find_scoped(id.into_inner(), &scope).await?)) } #[utoipa::path( diff --git a/api/src/routes/v2/deployment_resources.rs b/api/src/routes/v2/deployment_resources.rs index 70f7413..168e2d6 100644 --- a/api/src/routes/v2/deployment_resources.rs +++ b/api/src/routes/v2/deployment_resources.rs @@ -3,12 +3,14 @@ use actix_web::{HttpResponse, delete, get, post, put, web}; use futures::future::try_join_all; use platz_auth::ApiIdentity; use platz_db::{ + AccessScope, diesel_pagination::{Paginated, PaginationParams}, schema::{ deployment::Deployment, deployment_resource::{ - DeploymentResource, DeploymentResourceFilters, DeploymentResourceSyncStatus, - NewDeploymentResource, UpdateDeploymentResource, UpdateDeploymentResourceSyncStatus, + DeploymentResource, DeploymentResourceExtraFilters, DeploymentResourceFilters, + DeploymentResourceSyncStatus, NewDeploymentResource, UpdateDeploymentResource, + UpdateDeploymentResourceSyncStatus, }, deployment_resource_type::DeploymentResourceType, }, @@ -34,12 +36,19 @@ use uuid::Uuid; )] #[get("/deployment-resources")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, + extra_filters: web::Query, pagination: web::Query, ) -> ApiResult { - let mut result = - DeploymentResource::all_filtered(filters.into_inner(), pagination.into_inner()).await?; + let scope = AccessScope::for_identity(identity.inner()).await?; + let mut result = DeploymentResource::all_filtered( + filters.into_inner(), + extra_filters.into_inner(), + pagination.into_inner(), + &scope, + ) + .await?; result.items = try_join_all( result .items @@ -66,9 +75,10 @@ async fn get_all( ), )] #[get("/deployment-resources/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok().json( - DeploymentResource::find(id.into_inner()) + DeploymentResource::find_scoped(id.into_inner(), &scope) .await? .without_sensitive_props() .await?, diff --git a/api/src/routes/v2/deployment_tasks.rs b/api/src/routes/v2/deployment_tasks.rs index a612a74..c833e5a 100644 --- a/api/src/routes/v2/deployment_tasks.rs +++ b/api/src/routes/v2/deployment_tasks.rs @@ -5,7 +5,7 @@ use actix_web::{HttpResponse, delete, get, post, web}; use chrono::prelude::*; use platz_auth::ApiIdentity; use platz_db::{ - DbError, DbTableOrDeploymentResource, Json, + AccessScope, DbError, DbTableOrDeploymentResource, Json, diesel_pagination::{Paginated, PaginationParams}, schema::{ deployment::Deployment, @@ -41,16 +41,18 @@ use uuid::Uuid; )] #[get("/deployment-tasks")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, extra_filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok().json( DeploymentTask::all_filtered( filters.into_inner(), extra_filters.into_inner(), pagination.into_inner(), + &scope, ) .await?, )) @@ -72,8 +74,9 @@ async fn get_all( ), )] #[get("/deployment-tasks/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(DeploymentTask::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(DeploymentTask::find_scoped(id.into_inner(), &scope).await?)) } #[derive(Debug, Deserialize, ToSchema)] @@ -105,7 +108,8 @@ async fn cancel_one( body: web::Json, ) -> ApiResult { let body = body.into_inner(); - let task = DeploymentTask::find(id.into_inner()).await?; + let scope = AccessScope::for_identity(identity.inner()).await?; + let task = DeploymentTask::find_scoped(id.into_inner(), &scope).await?; let canceled_by_user_id = identity.inner().user_id(); let canceled_by_deployment_id = identity.inner().deployment_id(); diff --git a/api/src/routes/v2/deployments.rs b/api/src/routes/v2/deployments.rs index b10a5e1..4613ea0 100644 --- a/api/src/routes/v2/deployments.rs +++ b/api/src/routes/v2/deployments.rs @@ -6,7 +6,7 @@ use actix_web::{HttpResponse, delete, get, post, put, web}; use platz_auth::ApiIdentity; use platz_chart_ext::ChartExtCardinality; use platz_db::{ - DbTable, DbTableOrDeploymentResource, + AccessScope, DbTable, DbTableOrDeploymentResource, diesel_pagination::{Paginated, PaginationParams}, schema::{ deployment::{ @@ -38,16 +38,18 @@ use uuid::Uuid; )] #[get("/deployments")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, extra_filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok().json( Deployment::all_filtered( filters.into_inner(), extra_filters.into_inner(), pagination.into_inner(), + &scope, ) .await?, )) @@ -69,8 +71,9 @@ async fn get_all( ), )] #[get("/deployments/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(Deployment::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(Deployment::find_scoped(id.into_inner(), &scope).await?)) } #[utoipa::path( diff --git a/api/src/routes/v2/env_user_permissions.rs b/api/src/routes/v2/env_user_permissions.rs index 603dfdd..2e53ef3 100644 --- a/api/src/routes/v2/env_user_permissions.rs +++ b/api/src/routes/v2/env_user_permissions.rs @@ -2,6 +2,7 @@ use crate::{permissions::verify_env_admin, result::ApiResult}; use actix_web::{HttpResponse, delete, get, post, web}; use platz_auth::ApiIdentity; use platz_db::{ + AccessScope, diesel_pagination::{Paginated, PaginationParams}, schema::env_user_permission::{ EnvUserPermission, EnvUserPermissionFilters, NewEnvUserPermission, @@ -28,12 +29,14 @@ use uuid::Uuid; )] #[get("/env-user-permissions")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok().json( - EnvUserPermission::all_filtered(filters.into_inner(), pagination.into_inner()).await?, + EnvUserPermission::all_filtered(filters.into_inner(), pagination.into_inner(), &scope) + .await?, )) } @@ -53,8 +56,9 @@ async fn get_all( ), )] #[get("/env-user-permissions/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(EnvUserPermission::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(EnvUserPermission::find_scoped(id.into_inner(), &scope).await?)) } #[utoipa::path( diff --git a/api/src/routes/v2/envs.rs b/api/src/routes/v2/envs.rs index 147f185..b3c8e13 100644 --- a/api/src/routes/v2/envs.rs +++ b/api/src/routes/v2/envs.rs @@ -4,6 +4,7 @@ use actix_web::{HttpResponse, delete, get, post, put, web}; use itertools::Itertools; use platz_auth::ApiIdentity; use platz_db::{ + AccessScope, diesel_pagination::{Paginated, PaginationParams}, schema::{ deployment::Deployment, @@ -32,12 +33,13 @@ use uuid::Uuid; )] #[get("/envs")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok() - .json(Env::all_filtered(filters.into_inner(), pagination.into_inner()).await?)) + .json(Env::all_filtered(filters.into_inner(), pagination.into_inner(), &scope).await?)) } #[utoipa::path( @@ -56,8 +58,9 @@ async fn get_all( ), )] #[get("/envs/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(Env::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(Env::find_scoped(id.into_inner(), &scope).await?)) } #[utoipa::path( diff --git a/api/src/routes/v2/secrets.rs b/api/src/routes/v2/secrets.rs index 6f3d40a..695fd36 100644 --- a/api/src/routes/v2/secrets.rs +++ b/api/src/routes/v2/secrets.rs @@ -3,7 +3,7 @@ use crate::{permissions::verify_env_admin, result::ApiResult}; use actix_web::{HttpResponse, delete, get, post, put, web}; use platz_auth::ApiIdentity; use platz_db::{ - DbTable, DbTableOrDeploymentResource, + AccessScope, DbTable, DbTableOrDeploymentResource, diesel_pagination::{Paginated, PaginationParams}, schema::{ deployment::Deployment, @@ -31,12 +31,13 @@ use uuid::Uuid; )] #[get("/secrets")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; Ok(HttpResponse::Ok() - .json(Secret::all_filtered(filters.into_inner(), pagination.into_inner()).await?)) + .json(Secret::all_filtered(filters.into_inner(), pagination.into_inner(), &scope).await?)) } #[utoipa::path( @@ -55,8 +56,9 @@ async fn get_all( ), )] #[get("/secrets/{id}")] -async fn get_one(_identity: ApiIdentity, id: web::Path) -> ApiResult { - Ok(HttpResponse::Ok().json(Secret::find(id.into_inner()).await?)) +async fn get_one(identity: ApiIdentity, id: web::Path) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; + Ok(HttpResponse::Ok().json(Secret::find_scoped(id.into_inner(), &scope).await?)) } #[utoipa::path( diff --git a/api/src/routes/v2/ws.rs b/api/src/routes/v2/ws.rs index 1216620..6f02483 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -1,13 +1,53 @@ use actix::prelude::*; use actix_web::{Error, HttpRequest, HttpResponse, web}; use actix_web_actors::ws; -use platz_db::{DbEvent, DbEventData, DbEventOperation, db}; +use platz_auth::ApiIdentity; +use platz_db::{AccessScope, DbEvent, DbEventData, DbEventOperation, DbTable, db}; +use serde::Deserialize; +use std::collections::HashSet; use std::time::Duration; use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}; -use tracing::error; +use tracing::{error, warn}; +use uuid::Uuid; -#[derive(Default)] -struct DbEventsWs {} +/// Subprotocol used to carry the access token. Browsers cannot set an +/// `Authorization` header on a WebSocket, so the client authenticates by +/// connecting with `new WebSocket(url, [WS_AUTH_PROTOCOL, ])`, +/// which the browser sends as the `Sec-WebSocket-Protocol` request header. +const WS_AUTH_PROTOCOL: &str = "platz-auth-bearer"; + +/// A message sent by the client to control which events it receives. The client +/// subscribes to the (collection, environment) pairs the current view needs and +/// unsubscribes when navigating away, so the server only forwards events that +/// are both permitted and currently relevant. +#[derive(Debug, Deserialize, utoipa::ToSchema)] +#[serde(tag = "type", rename_all = "lowercase")] +pub enum ClientMessage { + Subscribe { + table: DbTable, + /// Environment to scope the subscription to. Omit for global + /// (non-environment-scoped) collections. + #[serde(default)] + #[schema(required)] + env_id: Option, + }, + Unsubscribe { + table: DbTable, + #[serde(default)] + #[schema(required)] + env_id: Option, + }, +} + +/// A websocket connection that streams database change events to a single +/// authenticated client, filtered to the environments the client may access and +/// to the (collection, environment) pairs it has subscribed to. +struct DbEventsWs { + scope: AccessScope, + /// Active subscriptions as (table, env) pairs. `None` env means a global + /// collection. An event is forwarded only when its (table, env_id) is here. + subscriptions: HashSet<(DbTable, Option)>, +} impl Actor for DbEventsWs { type Context = ws::WebsocketContext; @@ -33,12 +73,25 @@ impl DbEventsWs { } } +impl DbEventsWs { + fn handle_client_message(&mut self, text: &str) { + match serde_json::from_str::(text) { + Ok(ClientMessage::Subscribe { table, env_id }) => { + self.subscriptions.insert((table, env_id)); + } + Ok(ClientMessage::Unsubscribe { table, env_id }) => { + self.subscriptions.remove(&(table, env_id)); + } + Err(err) => warn!("Ignoring invalid websocket client message: {err}"), + } + } +} + impl StreamHandler> for DbEventsWs { fn handle(&mut self, msg: Result, ctx: &mut Self::Context) { match msg { Ok(ws::Message::Ping(msg)) => ctx.pong(&msg), - Ok(ws::Message::Text(text)) => ctx.text(text), - Ok(ws::Message::Binary(bin)) => ctx.binary(bin), + Ok(ws::Message::Text(text)) => self.handle_client_message(&text), _ => (), } } @@ -51,13 +104,24 @@ impl StreamHandler> for DbEventsWs { ctx: &mut Self::Context, ) { match event { - Ok(event) => match serde_json::to_string(&event) { - Ok(payload) => ctx.text(payload), - Err(err) => { - error!("Error serializing DB event for websocket: {err}"); - ctx.stop(); + Ok(event) => { + // Only forward events the connected identity is allowed to see. + // The event carries its environment, so this is a cheap check. + if !self.scope.can_receive_event(&event) { + return; + } + // ...and only those the client is currently subscribed to. + if !self.subscriptions.contains(&(event.table, event.env_id)) { + return; } - }, + match serde_json::to_string(&event) { + Ok(payload) => ctx.text(payload), + Err(err) => { + error!("Error serializing DB event for websocket: {err}"); + ctx.stop(); + } + } + } Err(err) => { error!("Error in websocket stream handler: {:?}", err); ctx.stop(); @@ -66,8 +130,40 @@ impl StreamHandler> for DbEventsWs { } } +/// Extract the access token from the `Sec-WebSocket-Protocol` header. The header +/// carries the auth subprotocol name followed by the token itself. +fn extract_token(req: &HttpRequest) -> Option { + let header = req.headers().get("Sec-WebSocket-Protocol")?.to_str().ok()?; + let mut parts = header.split(',').map(str::trim); + if parts.next()? != WS_AUTH_PROTOCOL { + return None; + } + parts + .next() + .filter(|token| !token.is_empty()) + .map(String::from) +} + async fn connect_ws(req: HttpRequest, stream: web::Payload) -> Result { - ws::start(DbEventsWs::default(), &req, stream) + // Authenticate the connection. Unlike the firehose this replaced, an + // unauthenticated client can no longer connect and receive events. + let token = extract_token(&req) + .ok_or_else(|| actix_web::error::ErrorUnauthorized("Missing websocket access token"))?; + let identity = ApiIdentity::from_access_token(&token).await?; + let scope = AccessScope::for_identity(identity.inner()) + .await + .map_err(|err| actix_web::error::ErrorServiceUnavailable(err.to_string()))?; + + // Echo the auth subprotocol back so the browser's WebSocket handshake + // succeeds. The connection starts with no subscriptions; the client + // subscribes to what each view needs. + let actor = DbEventsWs { + scope, + subscriptions: HashSet::new(), + }; + ws::WsResponseBuilder::new(actor, &req, stream) + .protocols(&[WS_AUTH_PROTOCOL]) + .start() } pub fn config(cfg: &mut web::ServiceConfig) { @@ -80,6 +176,6 @@ pub fn config(cfg: &mut web::ServiceConfig) { name = "Events", description = "Events sent through the Websocket.", )), - components(schemas(DbEvent, DbEventOperation, DbEventData)), + components(schemas(DbEvent, DbEventOperation, DbEventData, ClientMessage)), )] pub(super) struct OpenApi; diff --git a/auth/src/access_token.rs b/auth/src/access_token.rs index 524e20b..28a4f40 100644 --- a/auth/src/access_token.rs +++ b/auth/src/access_token.rs @@ -2,7 +2,7 @@ use crate::error::AuthError; use base64::prelude::*; use chrono::Duration; use chrono::prelude::*; -use jsonwebtoken::{EncodingKey, Header, encode}; +use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation, decode, encode}; use platz_db::{ Identity, schema::{deployment::Deployment, setting::Setting, user::User}, @@ -48,6 +48,23 @@ impl AccessToken { .map_err(AuthError::JwtEncodeError) } + /// Decode and validate a signed access token (JWT) string into its claims. + /// Shared by the `Authorization: Bearer` extractor and the websocket + /// subprotocol authentication path, which cannot use request headers. + pub(crate) async fn decode(token: &str) -> Result { + let mut validation = Validation::new(Algorithm::HS256); + validation.set_required_spec_claims(&["exp", "nbf"]); + validation.validate_exp = true; + validation.validate_nbf = true; + validation.leeway = 5; + let jwt_secret = get_jwt_secret().await?; + Ok( + decode::(token, &DecodingKey::from_secret(&jwt_secret), &validation) + .map_err(AuthError::JwtDecodeError)? + .claims, + ) + } + pub fn expires_at(&self) -> Result, AuthError> { DateTime::from_timestamp(self.exp as i64, 0) .ok_or_else(|| AuthError::NaiveDateTimeConvertOverflow(self.exp)) diff --git a/auth/src/actix_traits.rs b/auth/src/actix_traits.rs index 26271b3..82820f8 100644 --- a/auth/src/actix_traits.rs +++ b/auth/src/actix_traits.rs @@ -8,10 +8,7 @@ use actix_web::{FromRequest, HttpRequest, dev::Payload, http::header::HeaderName use actix_web_httpauth::extractors::bearer::BearerAuth; use futures::future::{BoxFuture, FutureExt, TryFutureExt, ok, ready}; use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, decode}; -use platz_db::{ - Identity, - schema::{bot::Bot, deployment::Deployment, user::User}, -}; +use platz_db::Identity; async fn validate_token(bearer: BearerAuth) -> Result, AuthError> { let mut validation = Validation::new(Algorithm::HS256); @@ -41,27 +38,6 @@ impl FromRequest for AccessToken { } } -impl super::ApiIdentity { - async fn validate(self) -> Result { - match self.inner() { - Identity::User(user_id) => User::find_only_active(user_id.to_owned()) - .await? - .map(|_| self) - .ok_or(AuthError::UserNotFound), - Identity::Bot(bot_id) => Bot::find(bot_id.to_owned()) - .await? - .map(|_| self) - .ok_or(AuthError::BotNotFound), - Identity::Deployment(deployment_id) => { - Deployment::find_optional(deployment_id.to_owned()) - .await? - .map(|_| self) - .ok_or(AuthError::DeploymentNotFound) - } - } - } -} - impl FromRequest for super::ApiIdentity { type Error = AuthError; type Future = BoxFuture<'static, Result>; diff --git a/auth/src/identity.rs b/auth/src/identity.rs index 27f927b..78e77e6 100644 --- a/auth/src/identity.rs +++ b/auth/src/identity.rs @@ -1,4 +1,8 @@ -use platz_db::Identity; +use crate::{access_token::AccessToken, error::AuthError}; +use platz_db::{ + Identity, + schema::{bot::Bot, deployment::Deployment, user::User}, +}; use serde::Serialize; use std::borrow::Borrow; @@ -13,6 +17,34 @@ impl ApiIdentity { pub fn into_inner(self) -> Identity { self.0 } + + /// Build and validate an identity from a raw access-token (JWT) string. + /// Used by the websocket authentication path, where the token arrives via + /// the `Sec-WebSocket-Protocol` header rather than `Authorization`. + pub async fn from_access_token(token: &str) -> Result { + let claims = AccessToken::decode(token).await?; + Self::from(Identity::from(claims)).validate().await + } + + /// Verify the identity still refers to an existing, active subject. + pub(crate) async fn validate(self) -> Result { + match self.inner() { + Identity::User(user_id) => User::find_only_active(user_id.to_owned()) + .await? + .map(|_| self) + .ok_or(AuthError::UserNotFound), + Identity::Bot(bot_id) => Bot::find(bot_id.to_owned()) + .await? + .map(|_| self) + .ok_or(AuthError::BotNotFound), + Identity::Deployment(deployment_id) => { + Deployment::find_optional(deployment_id.to_owned()) + .await? + .map(|_| self) + .ok_or(AuthError::DeploymentNotFound) + } + } + } } impl From for ApiIdentity { diff --git a/db/migrations/2026-06-13-184500_db-events-env-id/down.sql b/db/migrations/2026-06-13-184500_db-events-env-id/down.sql new file mode 100644 index 0000000..e5cdd9a --- /dev/null +++ b/db/migrations/2026-06-13-184500_db-events-env-id/down.sql @@ -0,0 +1,44 @@ +-- Restore the original notification function that emits only the row id, +-- without the resolved env_id. + +CREATE OR REPLACE FUNCTION notify_specific_trigger_name() RETURNS trigger AS $trigger$ +DECLARE + rec RECORD; + payload TEXT; + column_name TEXT; + column_value TEXT; + payload_items TEXT[]; +BEGIN + -- Set record row depending on operation + CASE TG_OP + WHEN 'INSERT', 'UPDATE' THEN + rec := NEW; + WHEN 'DELETE' THEN + rec := OLD; + ELSE + RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP; + END CASE; + + -- Get required fields + FOREACH column_name IN ARRAY TG_ARGV LOOP + EXECUTE format('SELECT $1.%I::TEXT', column_name) + INTO column_value + USING rec; + payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"'); + END LOOP; + + -- Build the payload + payload := '' + || '{' + || '"timestamp":"' || CURRENT_TIMESTAMP || '",' + || '"operation":"' || TG_OP || '",' + || '"schema":"' || TG_TABLE_SCHEMA || '",' + || '"table":"' || TG_TABLE_NAME || '",' + || '"data":{' || array_to_string(payload_items, ',') || '}' + || '}'; + + -- Notify the channel + PERFORM pg_notify(format('db_%I_notifications',TG_TABLE_NAME), payload); + RETURN rec; +END; +$trigger$ LANGUAGE plpgsql; diff --git a/db/migrations/2026-06-13-184500_db-events-env-id/up.sql b/db/migrations/2026-06-13-184500_db-events-env-id/up.sql new file mode 100644 index 0000000..9b5d337 --- /dev/null +++ b/db/migrations/2026-06-13-184500_db-events-env-id/up.sql @@ -0,0 +1,79 @@ +-- Enrich database change notifications with the environment of the changed +-- row. The API uses this to forward each websocket event only to clients that +-- are permitted to see that environment, without a per-event lookup. Because +-- the environment is resolved inside the trigger (from OLD on DELETE), it is +-- available even for deletions, where the row no longer exists afterwards. +-- +-- The environment is not a direct column on the notified tables, so it is +-- resolved through the cluster: +-- deployments -> k8s_clusters.env_id (via cluster_id) +-- deployment_tasks -> k8s_clusters.env_id (via cluster_id) +-- deployment_resources -> deployments -> k8s_clusters.env_id +-- Global tables (e.g. helm_tag_formats) carry a null env_id. + +CREATE OR REPLACE FUNCTION notify_specific_trigger_name() RETURNS trigger AS $trigger$ +DECLARE + rec RECORD; + payload TEXT; + column_name TEXT; + column_value TEXT; + payload_items TEXT[]; + v_env_id UUID; + env_id_json TEXT; +BEGIN + -- Set record row depending on operation + CASE TG_OP + WHEN 'INSERT', 'UPDATE' THEN + rec := NEW; + WHEN 'DELETE' THEN + rec := OLD; + ELSE + RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP; + END CASE; + + -- Resolve the environment of the changed row, where applicable. + v_env_id := NULL; + CASE TG_TABLE_NAME + WHEN 'deployments' THEN + SELECT k.env_id INTO v_env_id FROM k8s_clusters k WHERE k.id = rec.cluster_id; + WHEN 'deployment_tasks' THEN + SELECT k.env_id INTO v_env_id FROM k8s_clusters k WHERE k.id = rec.cluster_id; + WHEN 'deployment_resources' THEN + SELECT k.env_id INTO v_env_id + FROM deployments d + JOIN k8s_clusters k ON k.id = d.cluster_id + WHERE d.id = rec.deployment_id; + ELSE + v_env_id := NULL; + END CASE; + + IF v_env_id IS NULL THEN + env_id_json := 'null'; + ELSE + env_id_json := '"' || v_env_id::TEXT || '"'; + END IF; + + -- Get required fields + FOREACH column_name IN ARRAY TG_ARGV LOOP + EXECUTE format('SELECT $1.%I::TEXT', column_name) + INTO column_value + USING rec; + payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"'); + END LOOP; + + -- Build the payload + payload := '' + || '{' + || '"timestamp":"' || CURRENT_TIMESTAMP || '",' + || '"operation":"' || TG_OP || '",' + || '"schema":"' || TG_TABLE_SCHEMA || '",' + || '"table":"' || TG_TABLE_NAME || '",' + || '"env_id":' || env_id_json || ',' + || '"data":{' || array_to_string(payload_items, ',') || '}' + || '}'; + + -- Notify the channel + PERFORM pg_notify(format('db_%I_notifications',TG_TABLE_NAME), payload); + RETURN rec; +END; +$trigger$ LANGUAGE plpgsql; diff --git a/db/src/access.rs b/db/src/access.rs new file mode 100644 index 0000000..3c2139f --- /dev/null +++ b/db/src/access.rs @@ -0,0 +1,102 @@ +//! Authorization scoping shared by the REST API and the websocket event feed. +//! +//! Both the REST list/detail endpoints and the websocket notification feed must +//! make the *same* decision about which environments an identity is allowed to +//! see. [`AccessScope`] is that single source of truth: it is resolved once per +//! request (or once per websocket connection) and then applied as a SQL filter +//! so the database does the filtering in a single paginated query rather than +//! the application filtering rows one by one. + +use crate::{ + DbEvent, DbResult, DbTable, Identity, db_conn, + schema::{env_user_permission::env_user_permissions, user::users}, +}; +use diesel::prelude::*; +use diesel_async::RunQueryDsl; +use std::ops::DerefMut; +use uuid::Uuid; + +/// The set of environments an identity is allowed to access. +#[derive(Debug, Clone)] +pub enum AccessScope { + /// Unrestricted access. Site admins and the service identities (bots and + /// in-cluster deployments) see everything. + All, + /// A regular user, restricted to the environments they have any permission + /// in. May be empty, in which case the identity sees nothing. + Envs(Vec), +} + +impl AccessScope { + /// Resolve the scope for an identity with a single small, indexed query. + pub async fn for_identity(identity: &Identity) -> DbResult { + match identity { + Identity::User(user_id) => { + let is_admin = users::table + .find(user_id) + .select(users::is_admin) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()? + .unwrap_or(false); + if is_admin { + return Ok(Self::All); + } + let env_ids = env_user_permissions::table + .filter(env_user_permissions::user_id.eq(user_id)) + .select(env_user_permissions::env_id) + .get_results::(db_conn().await?.deref_mut()) + .await?; + Ok(Self::Envs(env_ids)) + } + // Service identities are trusted: bots and in-cluster deployments + // authenticate with their own tokens and operate across envs. + Identity::Bot(_) | Identity::Deployment(_) => Ok(Self::All), + } + } + + /// Whether this scope is unrestricted. + pub fn is_all(&self) -> bool { + matches!(self, Self::All) + } + + /// Whether the identity may access the given (possibly absent) environment. + /// An object with no environment is never visible to a restricted user. + pub fn allows_env(&self, env_id: Option) -> bool { + match self { + Self::All => true, + Self::Envs(env_ids) => env_id.is_some_and(|env_id| env_ids.contains(&env_id)), + } + } + + /// Decide whether the identity behind this scope may receive a websocket + /// [`DbEvent`]. + /// + /// The event's environment is carried on the event itself (resolved by the + /// database trigger), so this is a cheap in-memory check with no query — + /// and it works for `DELETE`s, where the row is already gone. Global tables + /// that are not environment-scoped are visible to any authenticated + /// identity; env-scoped events with no resolved environment fail closed. + pub fn can_receive_event(&self, event: &DbEvent) -> bool { + match self { + Self::All => true, + Self::Envs(env_ids) => match event.table { + // Global catalog / infrastructure tables are not env-scoped and + // are visible to every authenticated identity. + DbTable::HelmTagFormats + | DbTable::HelmCharts + | DbTable::HelmRegistries + | DbTable::DeploymentKinds + | DbTable::DeploymentResourceTypes + | DbTable::K8sClusters + | DbTable::K8sResources + | DbTable::Users + | DbTable::Bots + | DbTable::Settings => true, + // Everything else is env-scoped: forward only when the event's + // resolved environment is one the identity may access. + _ => event.env_id.is_some_and(|env_id| env_ids.contains(&env_id)), + }, + } + } +} diff --git a/db/src/db_table.rs b/db/src/db_table.rs index 699d1c9..a3b3ad0 100644 --- a/db/src/db_table.rs +++ b/db/src/db_table.rs @@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize}; use utoipa::ToSchema; use uuid::Uuid; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "snake_case")] pub enum DbTable { Bots, diff --git a/db/src/events.rs b/db/src/events.rs index 9ed86cd..ddbc72a 100644 --- a/db/src/events.rs +++ b/db/src/events.rs @@ -33,6 +33,13 @@ pub struct DbEventData { pub struct DbEvent { pub operation: DbEventOperation, pub table: DbTable, + /// Environment the changed row belongs to, resolved by the database trigger. + /// `None` for rows that are not environment-scoped (global tables) or whose + /// environment could not be resolved. Used to forward each event only to + /// clients permitted to see that environment. + #[serde(default)] + #[schema(required)] + pub env_id: Option, pub data: DbEventData, } diff --git a/db/src/identity.rs b/db/src/identity.rs index 16bfec0..a42ed71 100644 --- a/db/src/identity.rs +++ b/db/src/identity.rs @@ -2,7 +2,7 @@ use crate::schema::{deployment::Deployment, user::User}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -#[derive(Debug, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize)] pub enum Identity { User(Uuid), Bot(Uuid), diff --git a/db/src/lib.rs b/db/src/lib.rs index ffbd96c..924a6a3 100644 --- a/db/src/lib.rs +++ b/db/src/lib.rs @@ -1,3 +1,4 @@ +mod access; mod config; mod db_table; mod errors; @@ -10,6 +11,7 @@ pub mod tls; mod ui_collection; use crate::config::{DbPoolOptions, database_url, db_pool_options}; +pub use access::AccessScope; pub use config::{SslMode, SslSettings}; pub use db_table::*; use diesel_async::{ diff --git a/db/src/schema/deployment.rs b/db/src/schema/deployment.rs index c6fcaf3..b1f5ab0 100644 --- a/db/src/schema/deployment.rs +++ b/db/src/schema/deployment.rs @@ -5,7 +5,7 @@ use super::{ helm_chart::HelmChart, k8s_cluster::K8sCluster, }; -use crate::{DbError, DbResult, DbTableOrDeploymentResource, Identity, db_conn}; +use crate::{AccessScope, DbError, DbResult, DbTableOrDeploymentResource, Identity, db_conn}; use chrono::prelude::*; use diesel::{QueryDsl, prelude::*}; use diesel_async::RunQueryDsl; @@ -124,6 +124,7 @@ impl Deployment { filters: DeploymentFilters, extra_filters: DeploymentExtraFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { let allowed_cluster_ids: Option> = if let Some(env_id) = extra_filters.env_id { Some( @@ -141,6 +142,12 @@ impl Deployment { if let Some(cluster_ids) = allowed_cluster_ids { filtered = filtered.filter(deployments::cluster_id.eq_any(cluster_ids)) } + // Restrict to clusters in the environments the identity may access. The + // database does the filtering as part of the single paginated query. + if let AccessScope::Envs(env_ids) = scope { + let cluster_ids = K8sCluster::ids_in_envs(env_ids).await?; + filtered = filtered.filter(deployments::cluster_id.eq_any(cluster_ids)); + } Ok(filtered .order_by(deployments::created_at.asc()) @@ -156,6 +163,35 @@ impl Deployment { .await?) } + /// Like [`Self::find`] but only returns the deployment if it is within the + /// identity's [`AccessScope`]. A deployment outside the scope is + /// indistinguishable from a missing one (both yield `NotFound`), so the + /// endpoint returns the same `404` either way and does not leak existence. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => { + let cluster_ids = K8sCluster::ids_in_envs(env_ids).await?; + Ok(deployments::table + .find(id) + .filter(deployments::cluster_id.eq_any(cluster_ids)) + .get_result(db_conn().await?.deref_mut()) + .await?) + } + } + } + + /// IDs of all deployments in any of the given environments. Used to scope + /// env-scoped child collections (such as deployment resources). + pub async fn ids_in_envs(env_ids: &[Uuid]) -> DbResult> { + let cluster_ids = K8sCluster::ids_in_envs(env_ids).await?; + Ok(deployments::table + .filter(deployments::cluster_id.eq_any(cluster_ids)) + .select(deployments::id) + .get_results(db_conn().await?.deref_mut()) + .await?) + } + pub async fn find_optional(id: Uuid) -> DbResult> { Ok(deployments::table .find(id) diff --git a/db/src/schema/deployment_permission.rs b/db/src/schema/deployment_permission.rs index 18618c0..79aa6cd 100644 --- a/db/src/schema/deployment_permission.rs +++ b/db/src/schema/deployment_permission.rs @@ -1,4 +1,4 @@ -use crate::{DbResult, db_conn}; +use crate::{AccessScope, DbResult, db_conn}; use chrono::prelude::*; use diesel::prelude::*; use diesel_async::RunQueryDsl; @@ -65,8 +65,13 @@ impl DeploymentPermission { pub async fn all_filtered( filters: DeploymentPermissionFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + if let AccessScope::Envs(env_ids) = scope { + filtered = filtered.filter(deployment_permissions::env_id.eq_any(env_ids.clone())); + } + Ok(filtered .paginate(pagination) .load_and_count(db_conn().await?.deref_mut()) .await?) @@ -79,6 +84,19 @@ impl DeploymentPermission { .await?) } + /// Like [`Self::find`] but only returns the permission if its environment is + /// within the identity's [`AccessScope`]. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => Ok(deployment_permissions::table + .find(id) + .filter(deployment_permissions::env_id.eq_any(env_ids.clone())) + .get_result(db_conn().await?.deref_mut()) + .await?), + } + } + pub async fn find_user_role( env_id: Uuid, user_id: Uuid, diff --git a/db/src/schema/deployment_resource.rs b/db/src/schema/deployment_resource.rs index a4149b7..ac4bc32 100644 --- a/db/src/schema/deployment_resource.rs +++ b/db/src/schema/deployment_resource.rs @@ -1,5 +1,5 @@ use super::{deployment::Deployment, deployment_resource_type::DeploymentResourceType}; -use crate::{DbError, DbResult, db_conn}; +use crate::{AccessScope, DbError, DbResult, db_conn}; use chrono::prelude::*; use diesel::prelude::*; use diesel_async::RunQueryDsl; @@ -66,6 +66,12 @@ pub struct DeploymentResource { pub sync_reason: Option, } +#[derive(Debug, Default, Deserialize, ToSchema)] +pub struct DeploymentResourceExtraFilters { + #[schema(required)] + env_id: Option, +} + impl DeploymentResource { pub async fn all() -> DbResult> { Ok(deployment_resources::table @@ -75,9 +81,25 @@ impl DeploymentResource { pub async fn all_filtered( filters: DeploymentResourceFilters, + extra_filters: DeploymentResourceExtraFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + // Narrow to a single environment when requested, so a view can load + // just the resources of the environment it shows. + if let Some(env_id) = extra_filters.env_id { + let deployment_ids = Deployment::ids_in_envs(&[env_id]).await?; + filtered = filtered.filter(deployment_resources::deployment_id.eq_any(deployment_ids)); + } + // A resource is visible if its owning deployment is in an accessible + // environment. Resources with no deployment are not env-scoped and are + // hidden from restricted users. + if let AccessScope::Envs(env_ids) = scope { + let deployment_ids = Deployment::ids_in_envs(env_ids).await?; + filtered = filtered.filter(deployment_resources::deployment_id.eq_any(deployment_ids)); + } + Ok(filtered .paginate(pagination) .load_and_count(db_conn().await?.deref_mut()) .await?) @@ -90,6 +112,22 @@ impl DeploymentResource { .await?) } + /// Like [`Self::find`] but only returns the resource if its owning + /// deployment is within the identity's [`AccessScope`]. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => { + let deployment_ids = Deployment::ids_in_envs(env_ids).await?; + Ok(deployment_resources::table + .find(id) + .filter(deployment_resources::deployment_id.eq_any(deployment_ids)) + .get_result(db_conn().await?.deref_mut()) + .await?) + } + } + } + pub async fn find_by_type(type_id: Uuid) -> DbResult> { Ok(deployment_resources::table .filter(deployment_resources::type_id.eq(type_id)) diff --git a/db/src/schema/deployment_task.rs b/db/src/schema/deployment_task.rs index 39ec296..7d1092b 100644 --- a/db/src/schema/deployment_task.rs +++ b/db/src/schema/deployment_task.rs @@ -3,7 +3,7 @@ use super::{ helm_chart::HelmChart, k8s_cluster::K8sCluster, }; use crate::{ - DbError, DbResult, Identity, db_conn, + AccessScope, DbError, DbResult, Identity, db_conn, json_diff::{JsonDiff, json_diff}, }; use chrono::prelude::*; @@ -125,6 +125,7 @@ impl DeploymentTask { filters: DeploymentTaskFilters, extra_filters: DeploymentTaskExtraFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { let allowed_cluster_ids: Option> = if let Some(env_id) = extra_filters.env_id { Some( @@ -138,6 +139,10 @@ impl DeploymentTask { None }; let mut filtered = Self::filter(filters); + if let AccessScope::Envs(env_ids) = scope { + let cluster_ids = K8sCluster::ids_in_envs(env_ids).await?; + filtered = filtered.filter(deployment_tasks::cluster_id.eq_any(cluster_ids)); + } if extra_filters.active_only.unwrap_or(false) { filtered = filtered.filter( deployment_tasks::status @@ -176,6 +181,23 @@ impl DeploymentTask { .await?) } + /// Like [`Self::find`] but only returns the task if it is within the + /// identity's [`AccessScope`]. Out-of-scope and missing both yield + /// `NotFound` so the endpoint cannot be used to probe existence. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => { + let cluster_ids = K8sCluster::ids_in_envs(env_ids).await?; + Ok(deployment_tasks::table + .find(id) + .filter(deployment_tasks::cluster_id.eq_any(cluster_ids)) + .get_result(db_conn().await?.deref_mut()) + .await?) + } + } + } + pub async fn next_pending(cluster_ids: &Vec) -> DbResult> { Ok(deployment_tasks::table .filter(deployment_tasks::status.eq(DeploymentTaskStatus::Pending)) diff --git a/db/src/schema/env.rs b/db/src/schema/env.rs index 36a7128..95ae55a 100644 --- a/db/src/schema/env.rs +++ b/db/src/schema/env.rs @@ -1,5 +1,5 @@ use super::k8s_cluster::K8sCluster; -use crate::{DbResult, db_conn}; +use crate::{AccessScope, DbResult, db_conn}; use chrono::prelude::*; use diesel::prelude::*; use diesel_async::RunQueryDsl; @@ -44,8 +44,13 @@ impl Env { pub async fn all_filtered( filters: EnvFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + if let AccessScope::Envs(env_ids) = scope { + filtered = filtered.filter(envs::id.eq_any(env_ids.clone())); + } + Ok(filtered .paginate(pagination) .load_and_count(db_conn().await?.deref_mut()) .await?) @@ -58,6 +63,16 @@ impl Env { .await?) } + /// Like [`Self::find`] but only returns the env if it is within the + /// identity's [`AccessScope`]. Out-of-scope and missing both yield + /// `NotFound`. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + if !scope.allows_env(Some(id)) { + return Err(crate::DbError::NotFound); + } + Self::find(id).await + } + pub async fn delete(&self) -> DbResult<()> { K8sCluster::detach_from_env(self.id).await?; diesel::delete(envs::table.find(self.id)) diff --git a/db/src/schema/env_user_permission.rs b/db/src/schema/env_user_permission.rs index c6fa254..b2e85a0 100644 --- a/db/src/schema/env_user_permission.rs +++ b/db/src/schema/env_user_permission.rs @@ -1,4 +1,4 @@ -use crate::{DbResult, db_conn}; +use crate::{AccessScope, DbResult, db_conn}; use chrono::prelude::*; use diesel::prelude::*; use diesel_async::RunQueryDsl; @@ -60,8 +60,13 @@ impl EnvUserPermission { pub async fn all_filtered( filters: EnvUserPermissionFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + if let AccessScope::Envs(env_ids) = scope { + filtered = filtered.filter(env_user_permissions::env_id.eq_any(env_ids.clone())); + } + Ok(filtered .paginate(pagination) .load_and_count(db_conn().await?.deref_mut()) .await?) @@ -74,6 +79,19 @@ impl EnvUserPermission { .await?) } + /// Like [`Self::find`] but only returns the permission if its environment is + /// within the identity's [`AccessScope`]. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => Ok(env_user_permissions::table + .find(id) + .filter(env_user_permissions::env_id.eq_any(env_ids.clone())) + .get_result(db_conn().await?.deref_mut()) + .await?), + } + } + pub async fn find_user_role_in_env( env_id: Uuid, user_id: Uuid, diff --git a/db/src/schema/k8s_cluster.rs b/db/src/schema/k8s_cluster.rs index 7d8265a..2e8b473 100644 --- a/db/src/schema/k8s_cluster.rs +++ b/db/src/schema/k8s_cluster.rs @@ -89,6 +89,17 @@ impl K8sCluster { .await?) } + /// IDs of all clusters attached to any of the given environments. Used to + /// translate an [`AccessScope`](crate::AccessScope) of environments into the + /// cluster IDs that env-scoped queries (deployments, tasks) filter on. + pub async fn ids_in_envs(env_ids: &[Uuid]) -> DbResult> { + Ok(k8s_clusters::table + .filter(k8s_clusters::env_id.eq_any(env_ids.to_vec())) + .select(k8s_clusters::id) + .get_results(db_conn().await?.deref_mut()) + .await?) + } + pub async fn find_by_provider_id(value: String) -> DbResult> { Ok(k8s_clusters::table .filter(k8s_clusters::provider_id.eq(value)) diff --git a/db/src/schema/secret.rs b/db/src/schema/secret.rs index 2269666..b31e6fe 100644 --- a/db/src/schema/secret.rs +++ b/db/src/schema/secret.rs @@ -1,4 +1,4 @@ -use crate::{DbResult, db_conn}; +use crate::{AccessScope, DbResult, db_conn}; use chrono::prelude::*; use diesel::{QueryDsl, prelude::*}; use diesel_async::RunQueryDsl; @@ -47,8 +47,13 @@ impl Secret { pub async fn all_filtered( filters: SecretFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + if let AccessScope::Envs(env_ids) = scope { + filtered = filtered.filter(secrets::env_id.eq_any(env_ids.clone())); + } + Ok(filtered .paginate(pagination) .load_and_count(db_conn().await?.deref_mut()) .await?) @@ -61,6 +66,20 @@ impl Secret { .await?) } + /// Like [`Self::find`] but only returns the secret if it is within the + /// identity's [`AccessScope`]. Out-of-scope and missing both yield + /// `NotFound`. + pub async fn find_scoped(id: Uuid, scope: &AccessScope) -> DbResult { + match scope { + AccessScope::All => Self::find(id).await, + AccessScope::Envs(env_ids) => Ok(secrets::table + .find(id) + .filter(secrets::env_id.eq_any(env_ids.clone())) + .get_result(db_conn().await?.deref_mut()) + .await?), + } + } + pub async fn delete(&self) -> DbResult<()> { diesel::delete(secrets::table.find(self.id)) .execute(db_conn().await?.deref_mut())