From feecbdb61ee8dbef2ebff785c1f68d083f1ac928 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 13 Jun 2026 18:05:49 +0000 Subject: [PATCH 1/4] Scope REST and websocket events to the caller's environments Phase 1 (security) of the websocket scalability work. The websocket previously broadcast every database change to every connected client with no authentication, and the REST list/detail endpoints ignored the caller's identity, so any authenticated user could read or be notified about objects in environments they had no permission for. Introduce a single AccessScope authorization layer (db/src/access.rs), resolved once per request / per websocket connection: - AccessScope::All for site admins and service identities (bots, in-cluster deployments); AccessScope::Envs(..) for regular users, limited to the environments they hold any permission in. REST: every environment-scoped model (Deployment, DeploymentTask, DeploymentResource, Env, Secret, EnvUserPermission, DeploymentPermission) now applies the scope as a SQL filter inside the single paginated query (no fetch-then-filter), and exposes find_scoped() for detail endpoints. Out-of-scope objects return the same 404 as missing ones, so existence is not leaked through status codes. Websocket: the /api/v2/ws endpoint now requires authentication. Since browsers cannot set an Authorization header on a WebSocket, the access token is carried via the Sec-WebSocket-Protocol subprotocol. A per-connection task filters the event firehose down to the events the identity is allowed to see, resolving each event's environment on demand and preserving event ordering. Known limitation (to be addressed in a later phase): because events are not enriched with their environment, DELETE events whose row is already gone cannot be resolved and are withheld from restricted users (fail closed). --- api/Cargo.toml | 2 +- api/src/routes/v2/deployment_permissions.rs | 12 +- api/src/routes/v2/deployment_resources.rs | 12 +- api/src/routes/v2/deployment_tasks.rs | 14 +- api/src/routes/v2/deployments.rs | 11 +- api/src/routes/v2/env_user_permissions.rs | 12 +- api/src/routes/v2/envs.rs | 11 +- api/src/routes/v2/secrets.rs | 12 +- api/src/routes/v2/ws.rs | 119 +++++++++--- auth/src/access_token.rs | 19 +- auth/src/actix_traits.rs | 26 +-- auth/src/identity.rs | 34 +++- db/src/access.rs | 202 ++++++++++++++++++++ db/src/identity.rs | 2 +- db/src/lib.rs | 2 + db/src/schema/deployment.rs | 38 +++- db/src/schema/deployment_permission.rs | 22 ++- db/src/schema/deployment_resource.rs | 29 ++- db/src/schema/deployment_task.rs | 24 ++- db/src/schema/env.rs | 19 +- db/src/schema/env_user_permission.rs | 22 ++- db/src/schema/k8s_cluster.rs | 11 ++ db/src/schema/secret.rs | 23 ++- 23 files changed, 576 insertions(+), 102 deletions(-) create mode 100644 db/src/access.rs diff --git a/api/Cargo.toml b/api/Cargo.toml index 9fd09d6..5f02a66 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -26,7 +26,7 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" strum = "0.28.0" thiserror = "2.0.18" -tokio = { version = "1.52.3", features = ["rt-multi-thread", "signal"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "signal", "sync"] } tokio-stream = { version = "0.1.18", features = ["sync"] } tracing = "0.1.44" url = "2.5.8" 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..17ee4a1 100644 --- a/api/src/routes/v2/deployment_resources.rs +++ b/api/src/routes/v2/deployment_resources.rs @@ -3,6 +3,7 @@ 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, @@ -34,12 +35,14 @@ use uuid::Uuid; )] #[get("/deployment-resources")] async fn get_all( - _identity: ApiIdentity, + identity: ApiIdentity, filters: web::Query, pagination: web::Query, ) -> ApiResult { + let scope = AccessScope::for_identity(identity.inner()).await?; let mut result = - DeploymentResource::all_filtered(filters.into_inner(), pagination.into_inner()).await?; + DeploymentResource::all_filtered(filters.into_inner(), pagination.into_inner(), &scope) + .await?; result.items = try_join_all( result .items @@ -66,9 +69,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..559eade 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -1,28 +1,34 @@ 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, db}; use std::time::Duration; -use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}; -use tracing::error; +use tokio::sync::{broadcast::error::RecvError, mpsc}; +use tokio_stream::wrappers::UnboundedReceiverStream; +use tracing::{error, warn}; -#[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 websocket connection that streams database change events to a single +/// authenticated client. The events have already been filtered to the client's +/// [`AccessScope`] by a per-connection task before reaching the actor. +struct DbEventsWs { + /// Stream of authorized events. Taken in `started` to feed the actor. + events: Option>, +} impl Actor for DbEventsWs { type Context = ws::WebsocketContext; fn started(&mut self, ctx: &mut Self::Context) { - let rx = match db() { - Ok(db) => db.subscribe_to_events(), - Err(err) => { - error!("Could not subscribe to DB events: {err}"); - ctx.stop(); - return; - } - }; - let stream = BroadcastStream::new(rx); - ctx.add_stream(stream); + if let Some(events) = self.events.take() { + ctx.add_stream(events); + } ctx.run_interval(Duration::from_secs(30), Self::keepalive); } } @@ -44,30 +50,81 @@ impl StreamHandler> for DbEventsWs { } } -impl StreamHandler> for DbEventsWs { - fn handle( - &mut self, - event: Result, - 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(); - } - }, +/// Authorized database events forwarded by the per-connection filter task. +impl StreamHandler for DbEventsWs { + fn handle(&mut self, event: DbEvent, ctx: &mut Self::Context) { + match serde_json::to_string(&event) { + Ok(payload) => ctx.text(payload), Err(err) => { - error!("Error in websocket stream handler: {:?}", err); + error!("Error serializing DB event for websocket: {err}"); ctx.stop(); } } } } +/// 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()))?; + + let db = db().map_err(|err| actix_web::error::ErrorServiceUnavailable(err.to_string()))?; + let mut events = db.subscribe_to_events(); + + // The actor runs synchronously and cannot await the per-event authorization + // queries. A dedicated task does that work, forwarding only the events this + // identity may see. Decisions are awaited sequentially so event ordering + // (e.g. UPDATE before DELETE for the same row) is preserved. + let (tx, rx) = mpsc::unbounded_channel::(); + actix_web::rt::spawn(async move { + loop { + match events.recv().await { + Ok(event) => match scope.can_receive_event(&event).await { + Ok(true) => { + if tx.send(event).is_err() { + // Receiver (the websocket actor) is gone. + break; + } + } + Ok(false) => {} + Err(err) => error!("Error authorizing websocket event: {err}"), + }, + Err(RecvError::Lagged(skipped)) => { + warn!("Websocket event listener lagged, skipped {skipped} events"); + } + Err(RecvError::Closed) => break, + } + } + }); + + let actor = DbEventsWs { + events: Some(UnboundedReceiverStream::new(rx)), + }; + // Echo the auth subprotocol back so the browser's WebSocket handshake + // succeeds. + ws::WsResponseBuilder::new(actor, &req, stream) + .protocols(&[WS_AUTH_PROTOCOL]) + .start() } pub fn config(cfg: &mut web::ServiceConfig) { 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/src/access.rs b/db/src/access.rs new file mode 100644 index 0000000..437833f --- /dev/null +++ b/db/src/access.rs @@ -0,0 +1,202 @@ +//! 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::{ + deployment::deployments, deployment_resource::deployment_resources, + deployment_task::deployment_tasks, env_user_permission::env_user_permissions, + k8s_cluster::k8s_clusters, 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 is allowed to receive a + /// websocket [`DbEvent`]. + /// + /// The environment of the changed row is resolved on demand with a small + /// primary-key lookup (no cache). Catalog/global tables are visible to any + /// authenticated identity. For env-scoped rows we fail closed: if the + /// environment cannot be resolved (for example a `DELETE`, where the row no + /// longer exists) a restricted user does not receive the event. + pub async fn can_receive_event(&self, event: &DbEvent) -> DbResult { + let env_ids = match self { + Self::All => return Ok(true), + Self::Envs(env_ids) => env_ids, + }; + + let row_env_id = match event.table { + // Global catalog and infrastructure tables: not environment-scoped, + // so they 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 => return Ok(true), + + DbTable::Envs => Some(event.data.id), + DbTable::Secrets => secret_env_id(event.data.id).await?, + DbTable::EnvUserPermissions => env_user_permission_env_id(event.data.id).await?, + DbTable::DeploymentPermissions => deployment_permission_env_id(event.data.id).await?, + DbTable::Deployments => deployment_env_id(event.data.id).await?, + DbTable::DeploymentTasks => deployment_task_env_id(event.data.id).await?, + DbTable::DeploymentResources => deployment_resource_env_id(event.data.id).await?, + }; + + Ok(row_env_id.is_some_and(|env_id| env_ids.contains(&env_id))) + } +} + +/// Resolve the environment of a k8s cluster (clusters may be detached, hence +/// the nested `Option`). +async fn cluster_env_id(cluster_id: Uuid) -> DbResult> { + Ok(k8s_clusters::table + .find(cluster_id) + .select(k8s_clusters::env_id) + .get_result::>(db_conn().await?.deref_mut()) + .await + .optional()? + .flatten()) +} + +/// Resolve the environment of a deployment via its cluster. +async fn deployment_env_id(deployment_id: Uuid) -> DbResult> { + let cluster_id = deployments::table + .find(deployment_id) + .select(deployments::cluster_id) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()?; + match cluster_id { + Some(cluster_id) => cluster_env_id(cluster_id).await, + None => Ok(None), + } +} + +/// Resolve the environment of a deployment task via its cluster. +async fn deployment_task_env_id(task_id: Uuid) -> DbResult> { + let cluster_id = deployment_tasks::table + .find(task_id) + .select(deployment_tasks::cluster_id) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()?; + match cluster_id { + Some(cluster_id) => cluster_env_id(cluster_id).await, + None => Ok(None), + } +} + +/// Resolve the environment of a deployment resource via its deployment. +async fn deployment_resource_env_id(resource_id: Uuid) -> DbResult> { + let deployment_id = deployment_resources::table + .find(resource_id) + .select(deployment_resources::deployment_id) + .get_result::>(db_conn().await?.deref_mut()) + .await + .optional()? + .flatten(); + match deployment_id { + Some(deployment_id) => deployment_env_id(deployment_id).await, + None => Ok(None), + } +} + +/// Resolve the (non-null) environment of a secret by primary key. +async fn secret_env_id(id: Uuid) -> DbResult> { + use crate::schema::secret::secrets; + Ok(secrets::table + .find(id) + .select(secrets::env_id) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()?) +} + +/// Resolve the (non-null) environment of an env-user permission by primary key. +async fn env_user_permission_env_id(id: Uuid) -> DbResult> { + Ok(env_user_permissions::table + .find(id) + .select(env_user_permissions::env_id) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()?) +} + +/// Resolve the (non-null) environment of a deployment permission by primary key. +async fn deployment_permission_env_id(id: Uuid) -> DbResult> { + use crate::schema::deployment_permission::deployment_permissions; + Ok(deployment_permissions::table + .find(id) + .select(deployment_permissions::env_id) + .get_result::(db_conn().await?.deref_mut()) + .await + .optional()?) +} 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..4814a7c 100644 --- a/db/src/lib.rs +++ b/db/src/lib.rs @@ -1,3 +1,4 @@ +mod access; mod config; mod db_table; mod errors; @@ -9,6 +10,7 @@ mod stats; pub mod tls; mod ui_collection; +pub use access::AccessScope; use crate::config::{DbPoolOptions, database_url, db_pool_options}; pub use config::{SslMode, SslSettings}; pub use db_table::*; 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..3a71b6d 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; @@ -76,8 +76,17 @@ impl DeploymentResource { pub async fn all_filtered( filters: DeploymentResourceFilters, pagination: PaginationParams, + scope: &AccessScope, ) -> DbResult> { - Ok(Self::filter(filters) + let mut filtered = Self::filter(filters); + // 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 +99,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()) From 75f9ffd661b0b1f43620421893116147a79bf4f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 13 Jun 2026 18:49:10 +0000 Subject: [PATCH 2/4] Enrich DB change events with env_id and filter the websocket in O(1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolve the changed row's environment inside the notification trigger and include it in the event payload, instead of having the API resolve it per event. A new migration replaces notify_specific_trigger_name() so each NOTIFY payload carries "env_id", 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 (helm_tag_formats) carry a null env_id. Because the trigger uses OLD on DELETE, the environment is resolved even for deletions — this removes the previous fail-closed limitation where restricted users missed DELETE events for rows they were entitled to see. DbEvent gains an env_id field, so AccessScope::can_receive_event is now a synchronous in-memory check with no per-event database query. The websocket handler filters events inline again, dropping the per-connection forwarding task and channel that the async resolution previously required. Verified the trigger against a live PostgreSQL: INSERT/UPDATE/DELETE on env-scoped tables emit the correct env_id (including DELETE), detached clusters and global tables emit null, and the down migration reverts the payload shape. --- api/Cargo.toml | 2 +- api/src/routes/v2/ws.rs | 91 +++++----- .../down.sql | 44 +++++ .../2026-06-13-184500_db-events-env-id/up.sql | 79 +++++++++ db/src/access.rs | 158 ++++-------------- db/src/events.rs | 7 + 6 files changed, 201 insertions(+), 180 deletions(-) create mode 100644 db/migrations/2026-06-13-184500_db-events-env-id/down.sql create mode 100644 db/migrations/2026-06-13-184500_db-events-env-id/up.sql diff --git a/api/Cargo.toml b/api/Cargo.toml index 5f02a66..9fd09d6 100644 --- a/api/Cargo.toml +++ b/api/Cargo.toml @@ -26,7 +26,7 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" strum = "0.28.0" thiserror = "2.0.18" -tokio = { version = "1.52.3", features = ["rt-multi-thread", "signal", "sync"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "signal"] } tokio-stream = { version = "0.1.18", features = ["sync"] } tracing = "0.1.44" url = "2.5.8" diff --git a/api/src/routes/v2/ws.rs b/api/src/routes/v2/ws.rs index 559eade..f21779f 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -4,9 +4,8 @@ use actix_web_actors::ws; use platz_auth::ApiIdentity; use platz_db::{AccessScope, DbEvent, DbEventData, DbEventOperation, db}; use std::time::Duration; -use tokio::sync::{broadcast::error::RecvError, mpsc}; -use tokio_stream::wrappers::UnboundedReceiverStream; -use tracing::{error, warn}; +use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}; +use tracing::error; /// Subprotocol used to carry the access token. Browsers cannot set an /// `Authorization` header on a WebSocket, so the client authenticates by @@ -15,20 +14,25 @@ use tracing::{error, warn}; const WS_AUTH_PROTOCOL: &str = "platz-auth-bearer"; /// A websocket connection that streams database change events to a single -/// authenticated client. The events have already been filtered to the client's -/// [`AccessScope`] by a per-connection task before reaching the actor. +/// authenticated client, filtered to the environments the client may access. struct DbEventsWs { - /// Stream of authorized events. Taken in `started` to feed the actor. - events: Option>, + scope: AccessScope, } impl Actor for DbEventsWs { type Context = ws::WebsocketContext; fn started(&mut self, ctx: &mut Self::Context) { - if let Some(events) = self.events.take() { - ctx.add_stream(events); - } + let rx = match db() { + Ok(db) => db.subscribe_to_events(), + Err(err) => { + error!("Could not subscribe to DB events: {err}"); + ctx.stop(); + return; + } + }; + let stream = BroadcastStream::new(rx); + ctx.add_stream(stream); ctx.run_interval(Duration::from_secs(30), Self::keepalive); } } @@ -50,13 +54,29 @@ impl StreamHandler> for DbEventsWs { } } -/// Authorized database events forwarded by the per-connection filter task. -impl StreamHandler for DbEventsWs { - fn handle(&mut self, event: DbEvent, ctx: &mut Self::Context) { - match serde_json::to_string(&event) { - Ok(payload) => ctx.text(payload), +impl StreamHandler> for DbEventsWs { + fn handle( + &mut self, + event: Result, + ctx: &mut Self::Context, + ) { + match event { + 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; + } + 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 serializing DB event for websocket: {err}"); + error!("Error in websocket stream handler: {:?}", err); ctx.stop(); } } @@ -75,7 +95,10 @@ fn extract_token(req: &HttpRequest) -> Option { if parts.next()? != WS_AUTH_PROTOCOL { return None; } - parts.next().filter(|token| !token.is_empty()).map(String::from) + parts + .next() + .filter(|token| !token.is_empty()) + .map(String::from) } async fn connect_ws(req: HttpRequest, stream: web::Payload) -> Result { @@ -88,41 +111,9 @@ async fn connect_ws(req: HttpRequest, stream: web::Payload) -> Result(); - actix_web::rt::spawn(async move { - loop { - match events.recv().await { - Ok(event) => match scope.can_receive_event(&event).await { - Ok(true) => { - if tx.send(event).is_err() { - // Receiver (the websocket actor) is gone. - break; - } - } - Ok(false) => {} - Err(err) => error!("Error authorizing websocket event: {err}"), - }, - Err(RecvError::Lagged(skipped)) => { - warn!("Websocket event listener lagged, skipped {skipped} events"); - } - Err(RecvError::Closed) => break, - } - } - }); - - let actor = DbEventsWs { - events: Some(UnboundedReceiverStream::new(rx)), - }; // Echo the auth subprotocol back so the browser's WebSocket handshake // succeeds. - ws::WsResponseBuilder::new(actor, &req, stream) + ws::WsResponseBuilder::new(DbEventsWs { scope }, &req, stream) .protocols(&[WS_AUTH_PROTOCOL]) .start() } 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 index 437833f..3c2139f 100644 --- a/db/src/access.rs +++ b/db/src/access.rs @@ -9,11 +9,7 @@ use crate::{ DbEvent, DbResult, DbTable, Identity, db_conn, - schema::{ - deployment::deployments, deployment_resource::deployment_resources, - deployment_task::deployment_tasks, env_user_permission::env_user_permissions, - k8s_cluster::k8s_clusters, user::users, - }, + schema::{env_user_permission::env_user_permissions, user::users}, }; use diesel::prelude::*; use diesel_async::RunQueryDsl; @@ -73,130 +69,34 @@ impl AccessScope { } } - /// Decide whether the identity behind this scope is allowed to receive a - /// websocket [`DbEvent`]. + /// Decide whether the identity behind this scope may receive a websocket + /// [`DbEvent`]. /// - /// The environment of the changed row is resolved on demand with a small - /// primary-key lookup (no cache). Catalog/global tables are visible to any - /// authenticated identity. For env-scoped rows we fail closed: if the - /// environment cannot be resolved (for example a `DELETE`, where the row no - /// longer exists) a restricted user does not receive the event. - pub async fn can_receive_event(&self, event: &DbEvent) -> DbResult { - let env_ids = match self { - Self::All => return Ok(true), - Self::Envs(env_ids) => env_ids, - }; - - let row_env_id = match event.table { - // Global catalog and infrastructure tables: not environment-scoped, - // so they 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 => return Ok(true), - - DbTable::Envs => Some(event.data.id), - DbTable::Secrets => secret_env_id(event.data.id).await?, - DbTable::EnvUserPermissions => env_user_permission_env_id(event.data.id).await?, - DbTable::DeploymentPermissions => deployment_permission_env_id(event.data.id).await?, - DbTable::Deployments => deployment_env_id(event.data.id).await?, - DbTable::DeploymentTasks => deployment_task_env_id(event.data.id).await?, - DbTable::DeploymentResources => deployment_resource_env_id(event.data.id).await?, - }; - - Ok(row_env_id.is_some_and(|env_id| env_ids.contains(&env_id))) - } -} - -/// Resolve the environment of a k8s cluster (clusters may be detached, hence -/// the nested `Option`). -async fn cluster_env_id(cluster_id: Uuid) -> DbResult> { - Ok(k8s_clusters::table - .find(cluster_id) - .select(k8s_clusters::env_id) - .get_result::>(db_conn().await?.deref_mut()) - .await - .optional()? - .flatten()) -} - -/// Resolve the environment of a deployment via its cluster. -async fn deployment_env_id(deployment_id: Uuid) -> DbResult> { - let cluster_id = deployments::table - .find(deployment_id) - .select(deployments::cluster_id) - .get_result::(db_conn().await?.deref_mut()) - .await - .optional()?; - match cluster_id { - Some(cluster_id) => cluster_env_id(cluster_id).await, - None => Ok(None), - } -} - -/// Resolve the environment of a deployment task via its cluster. -async fn deployment_task_env_id(task_id: Uuid) -> DbResult> { - let cluster_id = deployment_tasks::table - .find(task_id) - .select(deployment_tasks::cluster_id) - .get_result::(db_conn().await?.deref_mut()) - .await - .optional()?; - match cluster_id { - Some(cluster_id) => cluster_env_id(cluster_id).await, - None => Ok(None), - } -} - -/// Resolve the environment of a deployment resource via its deployment. -async fn deployment_resource_env_id(resource_id: Uuid) -> DbResult> { - let deployment_id = deployment_resources::table - .find(resource_id) - .select(deployment_resources::deployment_id) - .get_result::>(db_conn().await?.deref_mut()) - .await - .optional()? - .flatten(); - match deployment_id { - Some(deployment_id) => deployment_env_id(deployment_id).await, - None => Ok(None), + /// 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)), + }, + } } } - -/// Resolve the (non-null) environment of a secret by primary key. -async fn secret_env_id(id: Uuid) -> DbResult> { - use crate::schema::secret::secrets; - Ok(secrets::table - .find(id) - .select(secrets::env_id) - .get_result::(db_conn().await?.deref_mut()) - .await - .optional()?) -} - -/// Resolve the (non-null) environment of an env-user permission by primary key. -async fn env_user_permission_env_id(id: Uuid) -> DbResult> { - Ok(env_user_permissions::table - .find(id) - .select(env_user_permissions::env_id) - .get_result::(db_conn().await?.deref_mut()) - .await - .optional()?) -} - -/// Resolve the (non-null) environment of a deployment permission by primary key. -async fn deployment_permission_env_id(id: Uuid) -> DbResult> { - use crate::schema::deployment_permission::deployment_permissions; - Ok(deployment_permissions::table - .find(id) - .select(deployment_permissions::env_id) - .get_result::(db_conn().await?.deref_mut()) - .await - .optional()?) -} 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, } From 6f0dae7c55894c5029ab189d793a0bb63e5f8b8f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 13 Jun 2026 19:34:03 +0000 Subject: [PATCH 3/4] Add websocket subscriptions so clients receive only what they view Turn the websocket from a per-identity firehose into a subscription feed. The connection now starts with no subscriptions; the client sends subscribe/unsubscribe messages over the socket for the (collection, environment) pairs the current view needs: { "type": "subscribe", "table": "deployments", "env_id": "" } { "type": "unsubscribe", "table": "deployments", "env_id": "" } Global (non-environment-scoped) collections subscribe without an env_id. The actor tracks the subscription set and forwards an event only when it is both permitted (AccessScope) and matches an active subscription, so a user viewing one environment no longer receives changes for the others. Also add an env_id filter to the deployment-resources list endpoint (mirroring deployments and deployment-tasks) so a view can load just the resources of the environment it shows, enabling per-environment lazy loading on the frontend. --- api/src/routes/v2/deployment_resources.rs | 16 ++++-- api/src/routes/v2/ws.rs | 68 ++++++++++++++++++++--- db/src/db_table.rs | 2 +- db/src/schema/deployment_resource.rs | 13 +++++ 4 files changed, 85 insertions(+), 14 deletions(-) diff --git a/api/src/routes/v2/deployment_resources.rs b/api/src/routes/v2/deployment_resources.rs index 17ee4a1..168e2d6 100644 --- a/api/src/routes/v2/deployment_resources.rs +++ b/api/src/routes/v2/deployment_resources.rs @@ -8,8 +8,9 @@ use platz_db::{ schema::{ deployment::Deployment, deployment_resource::{ - DeploymentResource, DeploymentResourceFilters, DeploymentResourceSyncStatus, - NewDeploymentResource, UpdateDeploymentResource, UpdateDeploymentResourceSyncStatus, + DeploymentResource, DeploymentResourceExtraFilters, DeploymentResourceFilters, + DeploymentResourceSyncStatus, NewDeploymentResource, UpdateDeploymentResource, + UpdateDeploymentResourceSyncStatus, }, deployment_resource_type::DeploymentResourceType, }, @@ -37,12 +38,17 @@ use uuid::Uuid; async fn get_all( identity: ApiIdentity, filters: web::Query, + extra_filters: web::Query, pagination: web::Query, ) -> ApiResult { let scope = AccessScope::for_identity(identity.inner()).await?; - let mut result = - DeploymentResource::all_filtered(filters.into_inner(), pagination.into_inner(), &scope) - .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 diff --git a/api/src/routes/v2/ws.rs b/api/src/routes/v2/ws.rs index f21779f..91bbf02 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -2,10 +2,13 @@ use actix::prelude::*; use actix_web::{Error, HttpRequest, HttpResponse, web}; use actix_web_actors::ws; use platz_auth::ApiIdentity; -use platz_db::{AccessScope, DbEvent, DbEventData, DbEventOperation, db}; +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; /// Subprotocol used to carry the access token. Browsers cannot set an /// `Authorization` header on a WebSocket, so the client authenticates by @@ -13,10 +16,37 @@ use tracing::error; /// 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. +/// 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 { @@ -43,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), _ => (), } } @@ -67,6 +110,10 @@ impl StreamHandler> for DbEventsWs { 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) => { @@ -112,8 +159,13 @@ async fn connect_ws(req: HttpRequest, stream: web::Payload) -> Result, } +#[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,10 +81,17 @@ impl DeploymentResource { pub async fn all_filtered( filters: DeploymentResourceFilters, + extra_filters: DeploymentResourceExtraFilters, pagination: PaginationParams, scope: &AccessScope, ) -> DbResult> { 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. From 2e3c29f9f743e7fdd8a63c68aaf0a85172db7dce Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 13 Jun 2026 20:10:43 +0000 Subject: [PATCH 4/4] Apply rustfmt formatting --- api/src/routes/v2/ws.rs | 6 +----- db/src/lib.rs | 2 +- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/api/src/routes/v2/ws.rs b/api/src/routes/v2/ws.rs index 91bbf02..6f02483 100644 --- a/api/src/routes/v2/ws.rs +++ b/api/src/routes/v2/ws.rs @@ -133,11 +133,7 @@ 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 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; diff --git a/db/src/lib.rs b/db/src/lib.rs index 4814a7c..924a6a3 100644 --- a/db/src/lib.rs +++ b/db/src/lib.rs @@ -10,8 +10,8 @@ mod stats; pub mod tls; mod ui_collection; -pub use access::AccessScope; 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::{