control-plane: authorize create-data-plane and update-l2-reporting via the request Snapshot (6/7)

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<Request>,
 ) -> Result<axum::Json<Response>, 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"
+        );
+    }
+}

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.

crates/control-plane-api/src/server/update_l2_reporting.rs

@@ -33,19 +33,7 @@ pub async fn update_l2_reporting(
     }): super::Request<Request>,
 ) -> Result<axum::Json<Response>, 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();

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 \