From 08cbb100c91bf03f9df2a364738004fc73f59e69 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 12 Aug 2026 11:40:55 +0000 Subject: [PATCH] control-plane: authorize create-data-plane and update-l2-reporting via the request Snapshot create-data-plane and update-l2-reporting replace their raw ops/-admin SQL pre-check (internal.user_roles) with the standard Envelope idiom (evaluate_names_authorization + authorization_outcome), extracted as server::authorize_ops_admin. The request's pinned Snapshot now serves both the pre-check and the ops publication: one request, one authorization view. A denial under a Snapshot which predates the request is now a 307 AuthZRetry rather than an immediate 403; the flow-plane-link unit gains curl -L so a redirect can't read as success. --- ...bf45b2ea36fd6fb8be6e988627b2bdf45e21d.json | 22 -- .../src/server/create_data_plane.rs | 194 ++++++++++++++++-- crates/control-plane-api/src/server/mod.rs | 16 ++ .../src/server/update_l2_reporting.rs | 55 +++-- local/systemd/flow-plane-link@.service | 11 + 5 files changed, 241 insertions(+), 57 deletions(-) delete mode 100644 .sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json diff --git a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json b/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json deleted file mode 100644 index 9dcadd077f7..00000000000 --- a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "role_prefix", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d" -} diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index cacdd507887..d876e382aa2 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -1,7 +1,7 @@ -use crate::directives::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping}; use crate::publications::{ DoNotRetry, DraftPublication, NoopInitialize, NoopWithCommit, PruneUnboundCollections, }; +use crate::directives::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping}; use anyhow::Context; use validator::Validate; @@ -64,19 +64,7 @@ pub async fn create_data_plane( }): super::Request, ) -> Result, crate::server::error::ApiError> { let models::authorizations::ControlClaims { sub: user_id, .. } = env.claims()?; - - if let None = sqlx::query!( - "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - user_id, - ) - .fetch_optional(&env.pg_pool) - .await? - { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + super::authorize_ops_admin(&env).await?; let (data_plane_fqdn, base_name, pulumi_stack) = match &private { None => ( @@ -232,7 +220,6 @@ pub async fn create_data_plane( .unwrap() .into(); - let snapshot = app.snapshot_watch.token(); let publication = DraftPublication { user_id: *user_id, logs_token: insert.logs_token, @@ -241,9 +228,7 @@ pub async fn create_data_plane( detail: Some(format!("publication for data-plane {base_name}")), // A one-shot handler invocation, with no queued row to anchor on. started_at: None, - snapshot: snapshot - .result() - .expect("authorization snapshot is not ready"), + snapshot: env.snapshot(), // We've already validated that the user can admin `ops/`, // so further authZ checks are unnecessary. verify_user_authz: false, @@ -332,3 +317,176 @@ impl Validate for Category { } } } + +/// The `ops/`-admin pre-check, evaluated against the request's pinned +/// Snapshot: a caller without `ops/` admin is rejected before any data-plane +/// state is touched — terminally (403) when the Snapshot postdates the +/// request, and with the platform-standard 307 retry when it doesn't. +#[cfg(test)] +mod test { + use crate::test_server; + + // From `fixtures/alice.sql`: admin of `aliceCo/` and nothing else. + const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111"); + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_create_data_plane_denied_for_non_ops_admin(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let response = server + .rest_client() + .post( + "/admin/create-data-plane", + &serde_json::json!({"name": "test-plane-c1", "category": "managed"}), + Some(&token), + ) + .send() + .await + .unwrap(); + + assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status()); + } + + /// A denial evaluated against a Snapshot which predates the request is + /// provisional: the endpoint answers with the platform-standard 307 + /// `AuthZRetry` (Retry-After + `started`/`retryAfter` params) rather than + /// a terminal 403, and a retry against the refreshed (authoritative) + /// Snapshot then resolves the denial terminally. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_create_data_plane_stale_snapshot_retries_then_denies(pool: sqlx::PgPool) { + let _guard = test_server::init(); + // gate=true serves an empty epoch-taken Snapshot first: every denial + // under it is provisional. The revoke cancelled by the first request + // refreshes the watch to the real (+1h) Snapshot. + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let client = flow_client_next::rest::Client { + base_url: server.base_url(), + http_client: reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(), + }; + let body = serde_json::json!({"name": "test-plane-c1", "category": "managed"}); + + let response = client + .post("/admin/create-data-plane", &body, Some(&token)) + .send() + .await + .unwrap(); + assert_eq!( + reqwest::StatusCode::TEMPORARY_REDIRECT, + response.status(), + "a stale denial must be provisional" + ); + assert!( + response + .headers() + .contains_key(reqwest::header::RETRY_AFTER) + ); + let location = response + .headers() + .get(reqwest::header::LOCATION) + .and_then(|l| l.to_str().ok()) + .expect("redirect carries a Location"); + assert!( + location.contains("started=") && location.contains("retryAfter="), + "Location must carry retry bookkeeping: {location}" + ); + + // The cancelled revoke triggers a refresh to the authoritative + // Snapshot; denials then become terminal. Bound the wait, since the + // refresh races this retry loop. + for attempt in 0..50 { + let response = client + .post("/admin/create-data-plane", &body, Some(&token)) + .send() + .await + .unwrap(); + match response.status() { + reqwest::StatusCode::FORBIDDEN => return, + reqwest::StatusCode::TEMPORARY_REDIRECT => { + tokio::time::sleep(std::time::Duration::from_millis(20 * attempt)).await; + } + other => panic!("unexpected interim status {other}"), + } + } + panic!("denial never became terminal under the refreshed snapshot"); + } + + /// The seeded system user — these endpoints' routine caller — holds a + /// direct `('ops/', 'admin')` row in `user_grants` (seed.sql); this pins + /// that exactly that grant shape resolves through the Snapshot's grant + /// walk, and that it doesn't leak into unrelated tenants. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_ops_admin_authorizes_via_snapshot(pool: sqlx::PgPool) { + let ops_admin = uuid::uuid!("99999999-9999-9999-9999-999999999999"); + sqlx::query("insert into auth.users (id, email) values ($1, 'ops-admin@example.com')") + .bind(ops_admin) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "insert into user_grants (user_id, object_role, capability) values ($1, 'ops/', 'admin')", + ) + .bind(ops_admin) + .execute(&pool) + .await + .unwrap(); + + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = crate::snapshot::try_fetch(&pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch snapshot"); + let snapshot = crate::Snapshot::new(tokens::now(), data); + + let claims = models::authorizations::ControlClaims { + iat: 0, + exp: u64::MAX, + sub: ops_admin, + role: "authenticated".to_string(), + aud: "authenticated".to_string(), + email: Some("ops-admin@example.com".to_string()), + }; + assert!( + crate::evaluate_names_authorization( + &snapshot, + &claims, + models::Capability::Admin, + ["ops/"], + ) + .is_ok(), + "an ops/ admin user_grant must satisfy the snapshot walk" + ); + assert!( + crate::evaluate_names_authorization( + &snapshot, + &claims, + models::Capability::Admin, + ["aliceCo/"], + ) + .is_err(), + "ops/ admin must not leak into unrelated tenants" + ); + } +} diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index 406393688d2..19e36b9c045 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -134,6 +134,22 @@ where Ok((None, ())) } +/// Require that the requesting user holds `admin` over the `ops/` tenant, +/// evaluated against the request's pinned Snapshot — the same Snapshot the +/// rest of the operation then uses: one request, one authorization view. +/// A denial is terminal only once the Snapshot postdates the request's start; +/// otherwise `authorization_outcome` yields the standard retry response. +pub(crate) async fn authorize_ops_admin(env: &crate::Envelope) -> Result<(), crate::ApiError> { + let policy_result = crate::evaluate_names_authorization( + env.snapshot(), + env.claims()?, + models::Capability::Admin, + ["ops/"], + ); + let (_expiry, ()) = env.authorization_outcome(policy_result).await?; + Ok(()) +} + /// Looks up the user's authorization grants for each item in /// `prefixes_or_names`, and calls the provided `attach` function with each /// item and its capability. The `Some` results are returned in a vec. diff --git a/crates/control-plane-api/src/server/update_l2_reporting.rs b/crates/control-plane-api/src/server/update_l2_reporting.rs index f7fc33ec5a0..253455fa6a7 100644 --- a/crates/control-plane-api/src/server/update_l2_reporting.rs +++ b/crates/control-plane-api/src/server/update_l2_reporting.rs @@ -33,19 +33,7 @@ pub async fn update_l2_reporting( }): super::Request, ) -> Result, crate::ApiError> { let crate::ControlClaims { sub: user_id, .. } = env.claims()?; - - if let None = sqlx::query!( - "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - user_id, - ) - .fetch_optional(&env.pg_pool) - .await? - { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + super::authorize_ops_admin(&env).await?; let template = include_str!("../../../../ops-catalog/reporting-L2-template.bundle.json"); let tables::DraftCatalog { collections, .. } = @@ -295,7 +283,6 @@ export class Derivation extends Types.IDerivation {"# }; let logs_token = uuid::Uuid::new_v4(); - let snapshot = app.snapshot_watch.token(); let publication = DraftPublication { user_id: *user_id, logs_token, @@ -304,9 +291,7 @@ export class Derivation extends Types.IDerivation {"# detail: Some(format!("publication for updating L2 reporting")), // A one-shot handler invocation, with no queued row to anchor on. started_at: None, - snapshot: snapshot - .result() - .expect("authorization snapshot is not ready"), + snapshot: env.snapshot(), default_data_plane_name: if default_data_plane.trim().is_empty() { None } else { @@ -352,6 +337,42 @@ export class Derivation extends Types.IDerivation {"# })) } +/// See `create_data_plane::test` — same pre-check, same contract: a caller +/// without `ops/` admin is rejected with 403 before any template work. +#[cfg(test)] +mod test { + use crate::test_server; + + const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111"); + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_update_l2_reporting_denied_for_non_ops_admin(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let response = server + .rest_client() + .post( + "/admin/update-l2-reporting", + &serde_json::json!({"defaultDataPlane": "", "dryRun": true}), + Some(&token), + ) + .send() + .await + .unwrap(); + + assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status()); + } +} + // Copied from crates/derive-typescript/src/codegen/mod.rs fn camel_case(name: &str, mut upper: bool) -> String { let mut w = String::new(); diff --git a/local/systemd/flow-plane-link@.service b/local/systemd/flow-plane-link@.service index 8cc19f12821..802d70b6a3d 100644 --- a/local/systemd/flow-plane-link@.service +++ b/local/systemd/flow-plane-link@.service @@ -20,6 +20,11 @@ EnvironmentFile=%h/flow-local/env/plane-link-%i.env # during warm-up, bounded to 120s of 2s-spaced attempts. The agent binds its API # port only once it can serve (crates/agent/src/main.rs), so a connection being # accepted means ready; a non-transient 4xx still fails fast with its body. +# +# The -L flag follows the 307 AuthZRetry the /admin endpoints answer with when +# an authorization denial is evaluated against a not-yet-authoritative +# snapshot (307 preserves the POST and body; the server paces early retries +# itself). Without it curl treats the redirect as success and silently no-ops. ExecStart=/bin/bash -c ' \ if [ -n "${DEKAF_ADDRESS:-}" ] && [ -n "${DEKAF_REGISTRY_ADDRESS:-}" ]; then \ printf "{\\"name\\":\\"%%s\\",\\"category\\":{\\"manual\\":{\\"brokerAddress\\":\\"%%s\\",\\"reactorAddress\\":\\"%%s\\",\\"hmacKeys\\":[\\"%%s\\"],\\"dekafAddress\\":\\"%%s\\",\\"dekafRegistryAddress\\":\\"%%s\\"}}}" \ @@ -32,6 +37,8 @@ ExecStart=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --retry 60 \ --retry-connrefused \ --retry-delay 2 \ @@ -49,6 +56,8 @@ ExecStart=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --retry 60 \ --retry-connrefused \ --retry-delay 2 \ @@ -76,6 +85,8 @@ ExecStop=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --fail-with-body \ --data-binary @- \ ${AGENT_API}/admin/update-l2-reporting \