diff --git a/.sqlx/query-3577a825b849ccd125309ed485c64124914c7c9d9203213837fbf2a62796fe9b.json b/.sqlx/query-3577a825b849ccd125309ed485c64124914c7c9d9203213837fbf2a62796fe9b.json new file mode 100644 index 00000000000..6ef5686e67e --- /dev/null +++ b/.sqlx/query-3577a825b849ccd125309ed485c64124914c7c9d9203213837fbf2a62796fe9b.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO internal.service_accounts (user_id, catalog_name, created_by)\n VALUES ($1, $2::text::catalog_name, $3)\n RETURNING created_at AS \"created_at!: chrono::DateTime\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "created_at!: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "3577a825b849ccd125309ed485c64124914c7c9d9203213837fbf2a62796fe9b" +} diff --git a/.sqlx/query-3ca07bc1e85a922a7b12de7a6dfeac5f15d5b8c1e5b01d10874e84710b8768af.json b/.sqlx/query-3ca07bc1e85a922a7b12de7a6dfeac5f15d5b8c1e5b01d10874e84710b8768af.json new file mode 100644 index 00000000000..01c08e645ce --- /dev/null +++ b/.sqlx/query-3ca07bc1e85a922a7b12de7a6dfeac5f15d5b8c1e5b01d10874e84710b8768af.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS(SELECT 1 FROM internal.service_accounts WHERE user_id = $1) AS \"exists!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "3ca07bc1e85a922a7b12de7a6dfeac5f15d5b8c1e5b01d10874e84710b8768af" +} diff --git a/.sqlx/query-45e004513d51bbe95e3b6e6ceff4b0bf4f9a0c8da1830dcd304f5c4157ba2441.json b/.sqlx/query-45e004513d51bbe95e3b6e6ceff4b0bf4f9a0c8da1830dcd304f5c4157ba2441.json new file mode 100644 index 00000000000..6de68677589 --- /dev/null +++ b/.sqlx/query-45e004513d51bbe95e3b6e6ceff4b0bf4f9a0c8da1830dcd304f5c4157ba2441.json @@ -0,0 +1,58 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n ak.id AS \"id!: models::Id\",\n ak.service_account_id,\n ak.label,\n ak.created_by,\n ak.created_at AS \"created_at!: chrono::DateTime\",\n ak.expires_at AS \"expires_at!: chrono::DateTime\",\n ak.last_used_at AS \"last_used_at: chrono::DateTime\"\n FROM internal.api_keys ak\n WHERE ak.service_account_id = ANY($1)\n AND ak.revoked_at IS NULL\n ORDER BY ak.created_at DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "service_account_id", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "label", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "created_by", + "type_info": "Uuid" + }, + { + "ordinal": 4, + "name": "created_at!: chrono::DateTime", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "expires_at!: chrono::DateTime", + "type_info": "Timestamptz" + }, + { + "ordinal": 6, + "name": "last_used_at: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + true + ] + }, + "hash": "45e004513d51bbe95e3b6e6ceff4b0bf4f9a0c8da1830dcd304f5c4157ba2441" +} diff --git a/.sqlx/query-629063373c60c75e6f7f1cb17b7b6438ee7bb6cb9005397703d413f93b6a4c59.json b/.sqlx/query-629063373c60c75e6f7f1cb17b7b6438ee7bb6cb9005397703d413f93b6a4c59.json new file mode 100644 index 00000000000..49fffcd968d --- /dev/null +++ b/.sqlx/query-629063373c60c75e6f7f1cb17b7b6438ee7bb6cb9005397703d413f93b6a4c59.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE internal.api_keys SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Macaddr8" + ] + }, + "nullable": [] + }, + "hash": "629063373c60c75e6f7f1cb17b7b6438ee7bb6cb9005397703d413f93b6a4c59" +} diff --git a/.sqlx/query-3284a4b8b8368d36bb286e8aaad6a12642ebfe7815281a62705bd3d7a32e2eb0.json b/.sqlx/query-7903341c9acaf216eafd315680b31fbe471070810a12f453b3e26b4f1cd70fcc.json similarity index 51% rename from .sqlx/query-3284a4b8b8368d36bb286e8aaad6a12642ebfe7815281a62705bd3d7a32e2eb0.json rename to .sqlx/query-7903341c9acaf216eafd315680b31fbe471070810a12f453b3e26b4f1cd70fcc.json index dbaada791a7..3a12243cf30 100644 --- a/.sqlx/query-3284a4b8b8368d36bb286e8aaad6a12642ebfe7815281a62705bd3d7a32e2eb0.json +++ b/.sqlx/query-7903341c9acaf216eafd315680b31fbe471070810a12f453b3e26b4f1cd70fcc.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n select users.email as email\n from user_grants\n join auth.users as users on user_grants.user_id = users.id\n where users.email is not null and user_grants.object_role = $1\n and user_grants.capability = 'admin'\n order by users.created_at asc\n ", + "query": "\n select users.email as email\n from user_grants\n join auth.users as users on user_grants.user_id = users.id\n where users.email is not null and user_grants.object_role = $1\n and user_grants.capability = 'admin'\n -- Exclude service accounts: their synthetic sa+ addresses\n -- must never be chosen as a tenant's Stripe billing contact.\n and not exists (\n select 1 from internal.service_accounts sa where sa.user_id = users.id\n )\n order by users.created_at asc\n ", "describe": { "columns": [ { @@ -18,5 +18,5 @@ true ] }, - "hash": "3284a4b8b8368d36bb286e8aaad6a12642ebfe7815281a62705bd3d7a32e2eb0" + "hash": "7903341c9acaf216eafd315680b31fbe471070810a12f453b3e26b4f1cd70fcc" } diff --git a/.sqlx/query-8397efbae3f2744e7ef47aa5f9adf3c93ab0e7def33b016e24b42a561e27c11d.json b/.sqlx/query-8397efbae3f2744e7ef47aa5f9adf3c93ab0e7def33b016e24b42a561e27c11d.json new file mode 100644 index 00000000000..01ec82ffd62 --- /dev/null +++ b/.sqlx/query-8397efbae3f2744e7ef47aa5f9adf3c93ab0e7def33b016e24b42a561e27c11d.json @@ -0,0 +1,31 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH new_key AS (\n SELECT\n internal.id_generator() AS id,\n -- 256 bits from pgcrypto's CSPRNG; the SHA-256 hashing\n -- below rests on secrets being high-entropy.\n encode(gen_random_bytes(32), 'hex') AS secret\n )\n INSERT INTO internal.api_keys (id, service_account_id, secret_hash, label, expires_at, created_by)\n SELECT\n nk.id,\n $1,\n -- SHA-256 rather than bcrypt: the secret is high-entropy random,\n -- so bcrypt isn't necessary here.\n -- Refresh-token secrets will migrate to the same scheme once\n -- the legacy postgREST token functions are retired.\n encode(digest(nk.secret, 'sha256'), 'hex'),\n $2,\n now() + $3::text::interval,\n $4\n FROM new_key nk\n RETURNING\n id AS \"id!: models::Id\",\n (SELECT secret FROM new_key) AS \"secret!: String\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!: models::Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "secret!: String", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Text", + "Uuid" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "8397efbae3f2744e7ef47aa5f9adf3c93ab0e7def33b016e24b42a561e27c11d" +} diff --git a/.sqlx/query-87c3c08fe92a4e2ecdc866ca6837fe80d08329efd2ea90dc56c42c722cb35f22.json b/.sqlx/query-87c3c08fe92a4e2ecdc866ca6837fe80d08329efd2ea90dc56c42c722cb35f22.json new file mode 100644 index 00000000000..ea3dd12cbcd --- /dev/null +++ b/.sqlx/query-87c3c08fe92a4e2ecdc866ca6837fe80d08329efd2ea90dc56c42c722cb35f22.json @@ -0,0 +1,54 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n sa.user_id,\n sa.catalog_name AS \"catalog_name!: String\",\n sa.created_by,\n sa.created_at AS \"created_at!: chrono::DateTime\",\n sa.updated_at AS \"updated_at!: chrono::DateTime\",\n -- The account's \"last used\" is the max across all its\n -- keys (revoked included): verify_api_key only ever\n -- stamps the key it authenticated, so api_keys is the\n -- single source of truth.\n (\n SELECT max(ak.last_used_at)\n FROM internal.api_keys ak\n WHERE ak.service_account_id = sa.user_id\n ) AS \"last_used_at: chrono::DateTime\"\n FROM internal.service_accounts sa\n WHERE sa.catalog_name::text ^@ ANY($1)\n AND ($2::timestamptz IS NULL OR sa.created_at < $2)\n ORDER BY sa.created_at DESC\n LIMIT $3 + 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "user_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "catalog_name!: String", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "created_by", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "created_at!: chrono::DateTime", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "updated_at!: chrono::DateTime", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "last_used_at: chrono::DateTime", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Timestamptz", + "Int4" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + null + ] + }, + "hash": "87c3c08fe92a4e2ecdc866ca6837fe80d08329efd2ea90dc56c42c722cb35f22" +} diff --git a/.sqlx/query-9496328d2bec1ffe02bafd913580e9b0ca37f7615c5dc86ae33aaf06f7c777ee.json b/.sqlx/query-9496328d2bec1ffe02bafd913580e9b0ca37f7615c5dc86ae33aaf06f7c777ee.json new file mode 100644 index 00000000000..7f296666bc5 --- /dev/null +++ b/.sqlx/query-9496328d2bec1ffe02bafd913580e9b0ca37f7615c5dc86ae33aaf06f7c777ee.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT user_id\n FROM internal.service_accounts\n WHERE catalog_name = $1::text::catalog_name\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "user_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "9496328d2bec1ffe02bafd913580e9b0ca37f7615c5dc86ae33aaf06f7c777ee" +} diff --git a/.sqlx/query-b75dee229313efb8195899acd566515aaabc3a9ea6bc6c40502c4b28607a7bc8.json b/.sqlx/query-b75dee229313efb8195899acd566515aaabc3a9ea6bc6c40502c4b28607a7bc8.json new file mode 100644 index 00000000000..7d2652ce4ec --- /dev/null +++ b/.sqlx/query-b75dee229313efb8195899acd566515aaabc3a9ea6bc6c40502c4b28607a7bc8.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT sa.catalog_name AS \"catalog_name!: String\"\n FROM internal.api_keys ak\n JOIN internal.service_accounts sa ON sa.user_id = ak.service_account_id\n WHERE ak.id = $1 AND ak.revoked_at IS NULL\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "catalog_name!: String", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Macaddr8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "b75dee229313efb8195899acd566515aaabc3a9ea6bc6c40502c4b28607a7bc8" +} diff --git a/.sqlx/query-bdc6b63dae3c20d0dcf342f6353c92c86eb49dd18354c1ac11566ce29510489c.json b/.sqlx/query-bdc6b63dae3c20d0dcf342f6353c92c86eb49dd18354c1ac11566ce29510489c.json new file mode 100644 index 00000000000..541a0452b0c --- /dev/null +++ b/.sqlx/query-bdc6b63dae3c20d0dcf342f6353c92c86eb49dd18354c1ac11566ce29510489c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM public.user_grants WHERE user_id = $1 AND object_role = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "bdc6b63dae3c20d0dcf342f6353c92c86eb49dd18354c1ac11566ce29510489c" +} diff --git a/.sqlx/query-cb2678905898cd566802b5f77cff7b967c364bfdb01f4d7d30ad7a62a1b42131.json b/.sqlx/query-cb2678905898cd566802b5f77cff7b967c364bfdb01f4d7d30ad7a62a1b42131.json new file mode 100644 index 00000000000..8bb93fbf4a9 --- /dev/null +++ b/.sqlx/query-cb2678905898cd566802b5f77cff7b967c364bfdb01f4d7d30ad7a62a1b42131.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT $1::text::interval > interval '0'\n AND $1::text::interval <= interval '1 year' AS \"ok!: bool\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "ok!: bool", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "cb2678905898cd566802b5f77cff7b967c364bfdb01f4d7d30ad7a62a1b42131" +} diff --git a/.sqlx/query-cbe2f613660b912b54a98dac61847400cd78f0106597944391d04faaa133fcd7.json b/.sqlx/query-cbe2f613660b912b54a98dac61847400cd78f0106597944391d04faaa133fcd7.json new file mode 100644 index 00000000000..228bb20d816 --- /dev/null +++ b/.sqlx/query-cbe2f613660b912b54a98dac61847400cd78f0106597944391d04faaa133fcd7.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE internal.api_keys\n SET last_used_at = now()\n WHERE id = $1\n AND secret_hash = encode(digest($2, 'sha256'), 'hex')\n AND expires_at > now()\n AND revoked_at IS NULL\n RETURNING service_account_id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "service_account_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Macaddr8", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "cbe2f613660b912b54a98dac61847400cd78f0106597944391d04faaa133fcd7" +} diff --git a/.sqlx/query-f7562d1965e61fe0ad91b6daf0ccf3e2d32a6ee387d69af18ea7b06047335376.json b/.sqlx/query-f7562d1965e61fe0ad91b6daf0ccf3e2d32a6ee387d69af18ea7b06047335376.json new file mode 100644 index 00000000000..d59830a69f5 --- /dev/null +++ b/.sqlx/query-f7562d1965e61fe0ad91b6daf0ccf3e2d32a6ee387d69af18ea7b06047335376.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO auth.users (id, email, raw_user_meta_data)\n VALUES ($1, $2, $3)\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Varchar", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "f7562d1965e61fe0ad91b6daf0ccf3e2d32a6ee387d69af18ea7b06047335376" +} diff --git a/crates/billing-integrations/src/publish.rs b/crates/billing-integrations/src/publish.rs index ebe31955887..276a31c6958 100644 --- a/crates/billing-integrations/src/publish.rs +++ b/crates/billing-integrations/src/publish.rs @@ -825,6 +825,11 @@ async fn get_or_create_customer_for_tenant( join auth.users as users on user_grants.user_id = users.id where users.email is not null and user_grants.object_role = $1 and user_grants.capability = 'admin' + -- Exclude service accounts: their synthetic addresses must + -- never be chosen as a tenant's Stripe billing contact. + and not exists ( + select 1 from internal.service_accounts sa where sa.user_id = users.id + ) order by users.created_at asc "#, tenant diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs index 3ca7d720771..3565327996d 100644 --- a/crates/control-plane-api/src/envelope.rs +++ b/crates/control-plane-api/src/envelope.rs @@ -237,8 +237,25 @@ impl axum::extract::FromRequestParts> for Envelope { let mut token = auth.token(); let exchanged_token: Option; - // Is this is a refresh token? If so, first exchange for an access token. - if !token.contains(".") { + // A service-account API key (flow_sa_-prefixed) or a + // refresh token (base64 JSON, which never contains '.') may + // be presented in lieu of a signed JWT. We exchange either + // for a short-lived signed access token and then verify it + // exactly like a JWT presented directly; a JWT is verified + // by its signature. The exchange's stateful database check + // is the source of trust — the signature just lets the + // minted token flow through the normal verify path. + if token.starts_with("flow_sa_") { + exchanged_token = Some( + crate::server::exchange_api_key( + &state.pg_pool, + &state.control_plane_jwt_encode_key, + token, + ) + .await?, + ); + token = exchanged_token.as_ref().unwrap(); + } else if !token.contains('.') { exchanged_token = Some( crate::server::exchange_refresh_token(&state.pg_pool, token).await?, ); diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index c8247a163f5..f30a4dc6d8d 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -270,17 +270,132 @@ pub async fn exchange_refresh_token( .fetch_one(pg_pool) .await .map_err(|err| { - tonic::Status::unauthenticated(format!("failed to exchange refresh token: {err}")) + // `generate_access_token` signals an unusable credential (unknown id, + // bad secret, or expired token) by `raise`-ing, which surfaces as + // SQLSTATE P0001 — the only legitimate 401, collapsed into one generic + // message so the response doesn't reveal which check failed. Any other + // error is an internal fault: log the detail and don't leak it. + if err.as_database_error().and_then(|e| e.code()).as_deref() == Some("P0001") { + tonic::Status::unauthenticated("invalid, expired, or unknown refresh token") + } else { + tracing::error!(?err, "failed to exchange refresh token"); + tonic::Status::internal("failed to exchange refresh token") + } })?; let GenerateTokenResponse { access_token } = serde_json::from_value(response.token.unwrap_or_default()).map_err(|err| { - tonic::Status::internal(format!("invalid access token generated: {err}")) + tracing::error!( + ?err, + "generate_access_token returned an unparseable response" + ); + tonic::Status::internal("invalid token response") })?; Ok(access_token) } +/// Statefully verify a service-account API key against the database, returning +/// the verified service account's `user_id`. +/// +/// This is the sole check of an API key's validity — secret hash, expiry, and +/// revocation — and it stamps the key's `last_used_at` in the same round-trip. +/// Both the bearer path and the token-exchange endpoint route through +/// [`exchange_api_key`], which mints a signed access token atop the verified +/// identity. +/// +/// Secrets are hashed with SHA-256 rather than bcrypt: the secret is +/// high-entropy random (so the slow bcrypt hash adds no brute-force protection). +pub async fn verify_api_key(pg_pool: &sqlx::PgPool, api_key: &str) -> tonic::Result { + let raw = api_key + .strip_prefix("flow_sa_") + .expect("caller dispatches on the flow_sa_ prefix"); + + use base64::Engine; + let decoded = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(raw) + .map_err(|_| tonic::Status::invalid_argument("malformed api key: invalid base64"))?; + let decoded = String::from_utf8(decoded) + .map_err(|_| tonic::Status::invalid_argument("malformed api key: invalid UTF-8"))?; + + let (id_str, secret) = decoded + .split_once(':') + .ok_or_else(|| tonic::Status::invalid_argument("malformed api key payload"))?; + let key_id: models::Id = id_str + .parse() + .map_err(|_| tonic::Status::invalid_argument("malformed api key: invalid key id"))?; + + // Validate the secret, expiry, and revocation in one query, stamping the + // key's last_used_at as part of the same verification round-trip. The owning + // account's "last used" is derived from its keys at read time, so there's no + // second row to update here. + let row = sqlx::query!( + r#" + UPDATE internal.api_keys + SET last_used_at = now() + WHERE id = $1 + AND secret_hash = encode(digest($2, 'sha256'), 'hex') + AND expires_at > now() + AND revoked_at IS NULL + RETURNING service_account_id + "#, + key_id as models::Id, + secret, + ) + .fetch_optional(pg_pool) + .await + .map_err(|err| { + // Log the database detail server-side; don't leak it to the caller. + tracing::error!(?err, "failed to verify service-account API key"); + tonic::Status::internal("failed to verify API key") + })? + .ok_or_else(|| tonic::Status::unauthenticated("invalid, expired, or revoked api key"))?; + + Ok(row.service_account_id) +} + +/// Lifetime of an access token minted in the application layer. A signed token +/// can't be revoked before it expires, so this bounds how long a minted token +/// outlives a credential revocation. Used by [`exchange_api_key`] today; the +/// refresh-token exchange will reuse it once that minting moves out of the SQL +/// `generate_access_token` function and into the application layer. +const ACCESS_TOKEN_TTL: chrono::Duration = chrono::Duration::hours(1); + +/// Exchange a service-account API key for a short-lived signed access token. +/// +/// Verifies the key statefully via [`verify_api_key`] and mints a JWT for the +/// verified service-account identity. This mirrors how a refresh token is +/// exchanged (see [`exchange_refresh_token`]): the bearer path mints a token, +/// verifies it, and discards it within the same request, while the +/// `/api/v1/auth/token` endpoint returns it to the caller. Because both route +/// through the same verification, a revoked or expired key can never yield a +/// token. +/// +/// On the bearer path the token's lifetime is moot — it's verified once and +/// thrown away — but for the exchange endpoint it bounds how long a token +/// outlives a key revocation. See [`ACCESS_TOKEN_TTL`]. +pub async fn exchange_api_key( + pg_pool: &sqlx::PgPool, + jwt_encode_key: &tokens::jwt::EncodingKey, + api_key: &str, +) -> tonic::Result { + let user_id = verify_api_key(pg_pool, api_key).await?; + + let now = tokens::now(); + let exp = now + ACCESS_TOKEN_TTL; + + let claims = crate::ControlClaims { + iat: now.timestamp() as u64, + exp: exp.timestamp() as u64, + sub: user_id, + role: "authenticated".to_string(), + aud: "authenticated".to_string(), + email: None, + }; + + tokens::jwt::sign(&claims, jwt_encode_key) +} + /// Parse a data-plane claims token without verifying its signature. /// Returns an `Unverified` wrapper to make clear the claims have not been verified. fn parse_untrusted_data_plane_claims( diff --git a/crates/control-plane-api/src/server/public/graphql/mod.rs b/crates/control-plane-api/src/server/public/graphql/mod.rs index 9b4222d35d8..545c1ee8729 100644 --- a/crates/control-plane-api/src/server/public/graphql/mod.rs +++ b/crates/control-plane-api/src/server/public/graphql/mod.rs @@ -38,6 +38,7 @@ mod live_specs; mod prefixes; mod publication_history; mod refresh_tokens; +mod service_accounts; pub mod status; mod storage_mappings; mod tenant; @@ -120,6 +121,7 @@ pub struct QueryRoot( connectors::ConnectorsQuery, tenant::TenantQuery, refresh_tokens::RefreshTokensQuery, + service_accounts::ServiceAccountsQuery, ); // Represents the portion of the GraphQL schema that deals with mutations. @@ -132,6 +134,7 @@ pub struct MutationRoot( invite_links::InviteLinksMutation, data_planes::DataPlanesMutation, refresh_tokens::RefreshTokensMutation, + service_accounts::ServiceAccountsMutation, ); pub fn create_schema(alert_config_defaults: models::AlertConfig) -> GraphQLSchema { diff --git a/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs b/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs index 29821b8dd17..6a5583b5480 100644 --- a/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs +++ b/crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs @@ -137,6 +137,15 @@ impl RefreshTokensMutation { )); } + // Service accounts authenticate exclusively via API keys, which are + // expiring and revocable. A refresh token bypasses both, so deny + // issuance to SA principals. + if super::service_accounts::is_service_account(&env.pg_pool, claims.sub).await? { + return Err(async_graphql::Error::new( + "service accounts cannot create refresh tokens; authenticate with an API key instead", + )); + } + let row = sqlx::query!( r#" WITH new_token AS ( @@ -240,9 +249,10 @@ mod test { /// Covers the refresh-token GraphQL surface (create → list → revoke, plus /// the `validFor` validation and not-found idempotency guards), the - /// `/api/v1/auth/token` refresh-token dispatch, and rejection of a refresh + /// `/api/v1/auth/token` refresh-token dispatch, rejection of a refresh /// token presented as a bearer credential when its secret is bad or it has - /// been revoked. + /// been revoked, and the guard denying refresh tokens to service-account + /// principals. /// /// The happy-path *exchange* — `generate_access_token` actually signing a /// JWT — is intentionally not exercised here: it reads `app.jwt_secret` from @@ -472,5 +482,52 @@ mod test { ) .await; assert!(revoke_again["errors"].is_array()); + + // === Service accounts cannot create refresh tokens === + // They authenticate via API keys, which are expiring and revocable; a + // refresh token would bypass both, so issuance to an SA principal must + // be denied. + let create_sa: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/refresh-token-bot" + grants: [{ prefix: "aliceCo/", capability: admin }] + ) { catalogName } + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + create_sa["errors"].is_null(), + "create SA should succeed: {create_sa}" + ); + + // The API doesn't expose the SA's backing user_id, so read it from the + // DB to mint a token whose `sub` is the service account (an SA principal + // has no email), standing in for an authenticated SA caller. + let sa_user_id: uuid::Uuid = sqlx::query_scalar( + "SELECT user_id FROM internal.service_accounts WHERE catalog_name = 'aliceCo/refresh-token-bot'", + ) + .fetch_one(&pool) + .await + .unwrap(); + let sa_token = server.make_access_token(sa_user_id, None); + + let sa_create_rt: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#"mutation { createRefreshToken(validFor: "P30D") { id secret } }"# + }), + Some(&sa_token), + ) + .await; + assert!( + sa_create_rt["errors"].is_array(), + "service account should be denied a refresh token: {sa_create_rt}" + ); } } diff --git a/crates/control-plane-api/src/server/public/graphql/service_accounts.rs b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs new file mode 100644 index 00000000000..7dffb699754 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/service_accounts.rs @@ -0,0 +1,1538 @@ +use super::TimestampCursor; +use async_graphql::{Context, types::connection}; + +#[derive(Debug, Clone, async_graphql::SimpleObject)] +pub struct ServiceAccount { + // A service account is addressed by its `catalog_name` handle. Its backing + // auth.users id is an implementation detail and is deliberately not exposed + // in the public schema; it can be added later if a need arises. + pub catalog_name: models::Name, + pub created_by: uuid::Uuid, + pub created_at: chrono::DateTime, + pub updated_at: chrono::DateTime, + pub last_used_at: Option>, + pub api_keys: Vec, +} + +/// A user_grant to seed a service account with at creation time. +#[derive(Debug, Clone, async_graphql::InputObject)] +pub struct ServiceAccountGrantInput { + pub prefix: models::Prefix, + pub capability: models::Capability, +} + +#[derive(Debug, Clone, async_graphql::SimpleObject)] +pub struct ApiKeyInfo { + #[graphql(name = "id")] + pub key_id: models::Id, + pub label: String, + pub created_by: uuid::Uuid, + pub created_at: chrono::DateTime, + pub expires_at: chrono::DateTime, + pub last_used_at: Option>, +} + +#[derive(Debug, Clone, async_graphql::SimpleObject)] +pub struct CreateApiKeyResult { + #[graphql(name = "id")] + pub key_id: models::Id, + pub secret: String, +} + +pub type PaginatedServiceAccounts = connection::Connection< + TimestampCursor, + ServiceAccount, + connection::EmptyFields, + connection::EmptyFields, + connection::DefaultConnectionName, + connection::DefaultEdgeName, + connection::DisableNodesField, +>; + +#[derive(Debug, Default)] +pub struct ServiceAccountsQuery; + +const DEFAULT_PAGE_SIZE: usize = 25; +const MAX_PREFIXES: usize = 20; + +#[async_graphql::Object] +impl ServiceAccountsQuery { + async fn service_accounts( + &self, + ctx: &Context<'_>, + after: Option, + first: Option, + ) -> async_graphql::Result { + let env = ctx.data::()?; + + let snapshot = env.snapshot(); + // Service accounts are visible to callers who can manage them: those + // holding ManageServiceAccount on a prefix covering the account's + // catalog_name. + let user_accessible_prefixes = super::authorized_prefixes::authorized_prefixes( + &snapshot.role_grants, + &snapshot.user_grants, + env.claims()?.sub, + models::authz::Capability::ManageServiceAccount, + None, + ); + + if user_accessible_prefixes.is_empty() { + return Ok(PaginatedServiceAccounts::new(false, false)); + } + if user_accessible_prefixes.len() > MAX_PREFIXES { + return Err(async_graphql::Error::new("Too many accessible prefixes")); + } + + connection::query_with::( + after, + None, + first, + None, + |after, _, first, _| async move { + let after_created_at = after.map(|c| c.0); + let limit = first.unwrap_or(DEFAULT_PAGE_SIZE); + + let sa_rows = sqlx::query!( + r#" + SELECT + sa.user_id, + sa.catalog_name AS "catalog_name!: String", + sa.created_by, + sa.created_at AS "created_at!: chrono::DateTime", + sa.updated_at AS "updated_at!: chrono::DateTime", + -- The account's "last used" is the max across all its + -- keys (revoked included): verify_api_key only ever + -- stamps the key it authenticated, so api_keys is the + -- single source of truth. + ( + SELECT max(ak.last_used_at) + FROM internal.api_keys ak + WHERE ak.service_account_id = sa.user_id + ) AS "last_used_at: chrono::DateTime" + FROM internal.service_accounts sa + WHERE sa.catalog_name::text ^@ ANY($1) + AND ($2::timestamptz IS NULL OR sa.created_at < $2) + ORDER BY sa.created_at DESC + LIMIT $3 + 1 + "#, + &user_accessible_prefixes, + after_created_at, + limit as i64, + ) + .fetch_all(&env.pg_pool) + .await?; + + let has_next = sa_rows.len() > limit; + + let user_ids: Vec = + sa_rows.iter().take(limit).map(|r| r.user_id).collect(); + + // Keys are batch-loaded for the whole page in one query (no + // N+1). The tradeoff is that this runs even when the caller + // didn't select `apiKeys`. That's fine for a low-frequency admin + // listing against an indexed column. Revisit with a `DataLoader` + // (a `#[ComplexObject]` api_keys resolver backed by a batching + // loader keyed on service_account_id) if either changes: callers + // commonly list service accounts WITHOUT `apiKeys` (making this + // fetch mostly wasted), or more lazily-resolved per-account child + // collections get added — at which point one batching mechanism + // beats several conditional eager fetches. + let key_rows = if user_ids.is_empty() { + vec![] + } else { + sqlx::query!( + r#" + SELECT + ak.id AS "id!: models::Id", + ak.service_account_id, + ak.label, + ak.created_by, + ak.created_at AS "created_at!: chrono::DateTime", + ak.expires_at AS "expires_at!: chrono::DateTime", + ak.last_used_at AS "last_used_at: chrono::DateTime" + FROM internal.api_keys ak + WHERE ak.service_account_id = ANY($1) + AND ak.revoked_at IS NULL + ORDER BY ak.created_at DESC + "#, + &user_ids, + ) + .fetch_all(&env.pg_pool) + .await? + }; + + let mut keys_by_sa: std::collections::HashMap> = + std::collections::HashMap::new(); + for kr in key_rows { + keys_by_sa + .entry(kr.service_account_id) + .or_default() + .push(ApiKeyInfo { + key_id: kr.id, + label: kr.label, + created_by: kr.created_by, + created_at: kr.created_at, + expires_at: kr.expires_at, + last_used_at: kr.last_used_at, + }); + } + + let edges: Vec<_> = sa_rows + .into_iter() + .take(limit) + .map(|r| { + let api_keys = keys_by_sa.remove(&r.user_id).unwrap_or_default(); + connection::Edge::new( + TimestampCursor(r.created_at), + ServiceAccount { + catalog_name: models::Name::new(&r.catalog_name), + created_by: r.created_by, + created_at: r.created_at, + updated_at: r.updated_at, + last_used_at: r.last_used_at, + api_keys, + }, + ) + }) + .collect(); + + let mut conn = connection::Connection::new(after_created_at.is_some(), has_next); + conn.edges = edges; + Ok(conn) + }, + ) + .await + } +} + +#[derive(Debug, Default)] +pub struct ServiceAccountsMutation; + +#[async_graphql::Object] +impl ServiceAccountsMutation { + /// Create a service account homed at the specified catalog name, seeded + /// with the given user_grants. + /// + /// `catalogName` is a management anchor: admins of a prefix covering it + /// may manage the account. It determines who may manage the account, not + /// what the account may access. Access is determined solely by the + /// account's user_grants, which may span multiple prefixes. + /// + /// The caller must have ManageServiceAccount on the catalog name AND + /// CreateGrant on each granted prefix. Creates an auth.users row, an + /// internal.service_accounts row, and a user_grants row per requested + /// grant. + async fn create_service_account( + &self, + ctx: &Context<'_>, + catalog_name: models::Name, + grants: Vec, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + if let Err(err) = validator::Validate::validate(&catalog_name) { + return Err(async_graphql::Error::new(format!( + "invalid catalog name: {err}" + ))); + } + // Managing the account (here, creating it under this anchor) requires + // ManageServiceAccount on the catalog name — the same capability that + // gates listing in ServiceAccountsQuery, so the read and write surfaces + // agree. This is deliberately narrower than full Admin. + super::verify_authorization( + env, + catalog_name.as_str(), + models::authz::Capability::ManageServiceAccount, + ) + .await?; + + for grant in &grants { + if let Err(err) = validator::Validate::validate(&grant.prefix) { + return Err(async_graphql::Error::new(format!( + "invalid grant prefix {}: {err}", + grant.prefix.as_str(), + ))); + } + // `none` confers no access until bundles are wired, so reject it + // rather than mint a no-op grant. + if grant.capability == models::Capability::None { + return Err(async_graphql::Error::new( + "grant capability must be one of: read, write, admin", + )); + } + } + + // Granting the account access to a prefix requires CreateGrant on that + // prefix — the anti-escalation guard, distinct from managing the + // account: a caller can't hand a service account reach they couldn't + // grant anyone. (Human-user grant creation still lives in PostgREST; + // when it migrates to GraphQL it should gate on this same CreateGrant + // capability.) + for grant in &grants { + super::verify_authorization( + env, + grant.prefix.as_str(), + models::authz::Capability::CreateGrant, + ) + .await?; + } + + let mut txn = env.pg_pool.begin().await?; + + let sa_user_id = uuid::Uuid::new_v4(); + + // Both the synthetic email and the catalog_name are unique and derived + // from the same handle, so either insert can raise the duplicate + // (SQLSTATE 23505) — and which one fires first depends on the + // environment (real Supabase enforces a unique email; the local stub + // does not). Map either violation to one clear message. + let duplicate_err = |err: sqlx::Error| -> async_graphql::Error { + if err.as_database_error().and_then(|e| e.code()).as_deref() == Some("23505") { + async_graphql::Error::new(format!( + "a service account already exists for catalog name '{}'", + catalog_name.as_str(), + )) + } else { + err.into() + } + }; + + sqlx::query!( + r#" + INSERT INTO auth.users (id, email, raw_user_meta_data) + VALUES ($1, $2, $3) + "#, + sa_user_id, + format!("{}@service_accounts.estuary.dev", catalog_name.as_str()), + serde_json::json!({ + "full_name": catalog_name.as_str(), + }), + ) + .execute(&mut *txn) + .await + .map_err(duplicate_err)?; + + let now = sqlx::query_scalar!( + r#" + INSERT INTO internal.service_accounts (user_id, catalog_name, created_by) + VALUES ($1, $2::text::catalog_name, $3) + RETURNING created_at AS "created_at!: chrono::DateTime" + "#, + sa_user_id, + catalog_name.as_str(), + claims.sub, + ) + .fetch_one(&mut *txn) + .await + .map_err(duplicate_err)?; + + for grant in &grants { + crate::grants::upsert_user_grant( + sa_user_id, + grant.prefix.as_str(), + grant.capability, + Some("service account grant".to_string()), + &mut txn, + ) + .await?; + } + + txn.commit().await?; + + tracing::info!( + %catalog_name, + ?grants, + %claims.sub, + %sa_user_id, + "created service account" + ); + + Ok(ServiceAccount { + catalog_name, + created_by: claims.sub, + created_at: now, + updated_at: now, + last_used_at: None, + api_keys: vec![], + }) + } + + /// Add a user_grant to a service account. + /// + /// The caller must manage the service account (ManageServiceAccount on its + /// catalog name) AND have CreateGrant on the granted prefix. The second + /// requirement prevents a caller from extending an account's access beyond + /// what they could grant anyone. (Human-user grant creation still lives in + /// PostgREST; when it migrates to GraphQL it should gate on this same + /// CreateGrant capability.) + async fn add_service_account_grant( + &self, + ctx: &Context<'_>, + catalog_name: models::Name, + prefix: models::Prefix, + capability: models::Capability, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + super::verify_authorization( + env, + catalog_name.as_str(), + models::authz::Capability::ManageServiceAccount, + ) + .await?; + + if let Err(err) = validator::Validate::validate(&prefix) { + return Err(async_graphql::Error::new(format!( + "invalid grant prefix {}: {err}", + prefix.as_str(), + ))); + } + // `none` confers no access until bundles are wired, so reject it + // rather than mint a no-op grant. + if capability == models::Capability::None { + return Err(async_graphql::Error::new( + "grant capability must be one of: read, write, admin", + )); + } + + super::verify_authorization(env, prefix.as_str(), models::authz::Capability::CreateGrant) + .await?; + + let user_id = resolve_service_account(&env.pg_pool, catalog_name.as_str()).await?; + + let mut txn = env.pg_pool.begin().await?; + crate::grants::upsert_user_grant( + user_id, + prefix.as_str(), + capability, + Some("service account grant".to_string()), + &mut txn, + ) + .await?; + txn.commit().await?; + + tracing::info!( + %user_id, + %catalog_name, + %prefix, + ?capability, + %claims.sub, + "added service account grant" + ); + + Ok(true) + } + + /// Remove a user_grant from a service account. + /// + /// The caller must manage the service account (ManageServiceAccount on its + /// catalog name). Unlike addServiceAccountGrant, no capability on the + /// grant's prefix is required: removal only ever narrows the account's + /// access, so managers may remove ANY grant — including grants to + /// prefixes they don't themselves administer. + async fn remove_service_account_grant( + &self, + ctx: &Context<'_>, + catalog_name: models::Name, + prefix: models::Prefix, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + super::verify_authorization( + env, + catalog_name.as_str(), + models::authz::Capability::ManageServiceAccount, + ) + .await?; + + let user_id = resolve_service_account(&env.pg_pool, catalog_name.as_str()).await?; + + let deleted = sqlx::query!( + "DELETE FROM public.user_grants WHERE user_id = $1 AND object_role = $2", + user_id, + prefix.as_str(), + ) + .execute(&env.pg_pool) + .await?; + + if deleted.rows_affected() == 0 { + return Err(async_graphql::Error::new("grant not found")); + } + + tracing::info!( + %user_id, + %catalog_name, + %prefix, + %claims.sub, + "removed service account grant" + ); + + Ok(true) + } + + /// Create an API key for a service account. + /// + /// Returns the key_id and the plaintext secret (flow_sa_...). + /// The secret is returned exactly once and cannot be retrieved again. + /// + /// The API key can be exchanged for an 1-hr access token via `POST /api/v1/auth/token` + /// or used directly as an `Authorization: Bearer` credential + async fn create_api_key( + &self, + ctx: &Context<'_>, + catalog_name: models::Name, + label: String, + #[graphql(desc = "ISO 8601 duration for key validity (e.g. P90D, P1Y)")] valid_for: String, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + super::verify_authorization( + env, + catalog_name.as_str(), + models::authz::Capability::ManageServiceAccount, + ) + .await?; + + let user_id = resolve_service_account(&env.pg_pool, catalog_name.as_str()).await?; + + // valid_for is documented as an ISO 8601 duration (e.g. P90D, P1Y). + // Reject anything that isn't ISO 8601 up front: the `::interval` cast + // below would otherwise also accept Postgres's own syntax ("90 days"), + // silently widening the contract and contradicting the field's docs and + // error messages. ISO 8601 durations always start with 'P'; no Postgres + // traditional unit does, so this prefix check cleanly distinguishes them. + if !valid_for.trim_start().starts_with('P') { + return Err(async_graphql::Error::new( + "valid_for must be an ISO 8601 duration, e.g. P90D or P1Y", + )); + } + + // Bound the lifetime so a key can't become an effectively-permanent + // credential, and require it to be positive. Postgres does the interval + // math, which is calendar-aware for the P1Y / P3M cases. + let within_bounds = sqlx::query_scalar!( + r#" + SELECT $1::text::interval > interval '0' + AND $1::text::interval <= interval '1 year' AS "ok!: bool" + "#, + valid_for, + ) + .fetch_one(&env.pg_pool) + .await; + + let within_bounds = match within_bounds { + Ok(ok) => ok, + // A 'P'-prefixed value can still fail the `::interval` cast: Postgres + // raises SQLSTATE 22007 (invalid_datetime_format) / 22008 + // (datetime_field_overflow) for a malformed duration and 22015 + // (interval_field_overflow) for one too extreme to parse. All are + // client errors, not internal faults, so surface a sanitized message. + Err(sqlx::Error::Database(db)) + if matches!(db.code().as_deref(), Some("22007" | "22008" | "22015")) => + { + return Err(async_graphql::Error::new( + "invalid valid_for: expected an ISO 8601 duration, e.g. P90D or P1Y", + )); + } + // Any other database error is an internal fault: log the detail + // server-side and don't leak it to the caller. + Err(err) => { + tracing::error!(?err, "failed to validate api key valid_for"); + return Err(async_graphql::Error::new("failed to create api key")); + } + }; + + if !within_bounds { + return Err(async_graphql::Error::new( + "valid_for must be a positive duration no greater than 1 year", + )); + } + + let row = sqlx::query!( + r#" + WITH new_key AS ( + SELECT + internal.id_generator() AS id, + -- 256 bits from pgcrypto's CSPRNG; the SHA-256 hashing + -- below rests on secrets being high-entropy. + encode(gen_random_bytes(32), 'hex') AS secret + ) + INSERT INTO internal.api_keys (id, service_account_id, secret_hash, label, expires_at, created_by) + SELECT + nk.id, + $1, + -- SHA-256 rather than bcrypt: the secret is high-entropy random, + -- so bcrypt isn't necessary here. + encode(digest(nk.secret, 'sha256'), 'hex'), + $2, + now() + $3::text::interval, + $4 + FROM new_key nk + RETURNING + id AS "id!: models::Id", + (SELECT secret FROM new_key) AS "secret!: String" + "#, + user_id, + label, + valid_for, + claims.sub, + ) + .fetch_one(&env.pg_pool) + .await?; + + use base64::Engine; + let payload = format!("{}:{}", row.id, row.secret); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.as_bytes()); + let full_secret = format!("flow_sa_{encoded}"); + + tracing::info!( + key_id = %row.id, + %user_id, + %label, + %claims.sub, + "created api key for service account" + ); + + Ok(CreateApiKeyResult { + key_id: row.id, + secret: full_secret, + }) + } + + /// Revoke an API key. + /// + /// The caller must have ManageServiceAccount capability on the owning service account's + /// catalog name. + /// + /// Rather than deleting the row, we stamp `revoked_at`, which makes the key + /// inert (excluded from bearer authentication and listings) while + /// preserving the audit trail. Already-revoked keys are treated as not + /// found. + async fn revoke_api_key( + &self, + ctx: &Context<'_>, + #[graphql(name = "id")] key_id: models::Id, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + let catalog_name = sqlx::query_scalar!( + r#" + SELECT sa.catalog_name AS "catalog_name!: String" + FROM internal.api_keys ak + JOIN internal.service_accounts sa ON sa.user_id = ak.service_account_id + WHERE ak.id = $1 AND ak.revoked_at IS NULL + "#, + key_id as models::Id, + ) + .fetch_optional(&env.pg_pool) + .await?; + + let catalog_name = match catalog_name { + Some(name) => name, + None => return Err(async_graphql::Error::new("API key not found")), + }; + + super::verify_authorization( + env, + &catalog_name, + models::authz::Capability::ManageServiceAccount, + ) + .await?; + + sqlx::query!( + "UPDATE internal.api_keys SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL", + key_id as models::Id + ) + .execute(&env.pg_pool) + .await?; + + tracing::info!( + %key_id, + service_account = %catalog_name, + %claims.sub, + "revoked api key" + ); + + Ok(true) + } +} + +/// Returns whether `user_id` is backed by a service account. Used to deny +/// SA principals operations reserved for human users (e.g. refresh tokens). +pub(super) async fn is_service_account( + pg_pool: &sqlx::PgPool, + user_id: uuid::Uuid, +) -> async_graphql::Result { + let exists = sqlx::query_scalar!( + r#"SELECT EXISTS(SELECT 1 FROM internal.service_accounts WHERE user_id = $1) AS "exists!""#, + user_id, + ) + .fetch_one(pg_pool) + .await?; + + Ok(exists) +} + +/// Resolve a service account's backing `user_id` from its `catalog_name` handle. +/// +/// Service accounts are addressed publicly by catalog name; the writes still +/// need the backing auth.users id. Callers authorize against the catalog name +/// *before* resolving, so a "not found" here is for an authorized namespace. +async fn resolve_service_account( + pg_pool: &sqlx::PgPool, + catalog_name: &str, +) -> async_graphql::Result { + let row = sqlx::query_scalar!( + r#" + SELECT user_id + FROM internal.service_accounts + WHERE catalog_name = $1::text::catalog_name + "#, + catalog_name, + ) + .fetch_optional(pg_pool) + .await?; + + row.ok_or_else(|| async_graphql::Error::new("service account not found")) +} + +#[cfg(test)] +mod test { + use crate::test_server; + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../../../fixtures", scripts("data_planes", "alice")) + )] + async fn test_service_account_lifecycle(pool: sqlx::PgPool) { + let _guard = test_server::init(); + + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + + let alice_token = server.make_access_token( + uuid::Uuid::from_bytes([0x11; 16]), + Some("alice@example.test"), + ); + + // Create a bob user who does NOT have admin on aliceCo/. + sqlx::query("INSERT INTO auth.users (id, email) VALUES ('22222222-2222-2222-2222-222222222222', 'bob@example.test')") + .execute(&pool) + .await + .unwrap(); + + let bob_token = + server.make_access_token(uuid::Uuid::from_bytes([0x22; 16]), Some("bob@example.test")); + + // === Create a service account with multiple seeded grants === + let create_response: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($catalogName: Name!, $grants: [ServiceAccountGrantInput!]!) { + createServiceAccount( + catalogName: $catalogName + grants: $grants + ) { + catalogName + createdBy + createdAt + updatedAt + lastUsedAt + apiKeys { id } + } + }"#, + "variables": { + "catalogName": "aliceCo/ci-deploy-bot", + "grants": [ + { "prefix": "aliceCo/", "capability": "admin" }, + { "prefix": "aliceCo/data/", "capability": "read" } + ] + } + }), + Some(&alice_token), + ) + .await; + + assert!( + create_response["errors"].is_null(), + "create should succeed: {create_response}" + ); + let sa = &create_response["data"]["createServiceAccount"]; + // The public API doesn't expose the backing user_id; fetch it from the + // DB for the row-level assertions below. + let sa_user_id = service_account_user_id(&pool, "aliceCo/ci-deploy-bot").await; + assert_eq!(sa["catalogName"], "aliceCo/ci-deploy-bot"); + assert_eq!(sa["apiKeys"].as_array().unwrap().len(), 0); + + // === A catalog name is unique to one service account === + // A second account cannot claim the same handle, even for an authorized + // caller. + let dup: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/ci-deploy-bot" + grants: [{ prefix: "aliceCo/", capability: admin }] + ) { catalogName } + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + dup["errors"] + .as_array() + .is_some_and(|errs| errs.iter().any(|e| e["message"] + .as_str() + .is_some_and(|m| m.contains("already exists")))), + "duplicate catalog name should be rejected: {dup}" + ); + assert_eq!( + grant_count(&pool, &sa_user_id).await, + 2, + "each requested grant should mint a user_grants row" + ); + // Provenance and timestamp fields are populated on creation: createdBy + // is the calling admin (alice), the timestamps are set, and a freshly + // created account has never been used. + assert_eq!( + sa["createdBy"], "11111111-1111-1111-1111-111111111111", + "createdBy should be the calling admin: {create_response}" + ); + assert!(sa["createdAt"].is_string(), "createdAt should be set: {sa}"); + assert!(sa["updatedAt"].is_string(), "updatedAt should be set: {sa}"); + assert!( + sa["lastUsedAt"].is_null(), + "a never-used account should have null lastUsedAt: {sa}" + ); + + // === Bob cannot create a service account for aliceCo/ === + let unauthorized: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/hacker-bot" + grants: [{ prefix: "aliceCo/", capability: read }] + ) { catalogName } + }"# + }), + Some(&bob_token), + ) + .await; + + assert!(unauthorized["errors"].is_array()); + + // === create_service_account input validation === + // An invalid catalog name is rejected (before authorization), even + // for an admin caller. + let bad_name: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount(catalogName: "Not A Name", grants: []) { catalogName } + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + bad_name["errors"][0]["message"] + .as_str() + .unwrap_or_default() + .contains("invalid catalog name"), + "invalid catalog name should be rejected: {bad_name}" + ); + + // capability `none` confers no access until bundles are wired, so it is + // rejected rather than minting a no-op grant. + let none_capability: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/no-op-bot" + grants: [{ prefix: "aliceCo/", capability: none }] + ) { catalogName } + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + none_capability["errors"][0]["message"] + .as_str() + .unwrap_or_default() + .contains("capability must be one of"), + "Capability::None should be rejected: {none_capability}" + ); + + // Every requested grant is independently authorized: alice admins + // aliceCo/ but not bobCo/, so seeding a bobCo/ grant must fail even + // though the account itself is homed under her prefix. + let foreign_grant: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/overreach-bot" + grants: [ + { prefix: "aliceCo/", capability: read }, + { prefix: "bobCo/", capability: read } + ] + ) { catalogName } + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + foreign_grant["errors"].is_array(), + "a grant to an unadministered prefix should be rejected: {foreign_grant}" + ); + + // === Create an API key === + let create_key: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($label: String!, $validFor: String!) { + createApiKey( + catalogName: "aliceCo/ci-deploy-bot" + label: $label + validFor: $validFor + ) { + id + secret + } + }"#, + "variables": { + "label": "GitHub Actions", + "validFor": "P90D" + } + }), + Some(&alice_token), + ) + .await; + + assert!( + create_key["errors"].is_null(), + "create key should succeed: {create_key}" + ); + let key_data = &create_key["data"]["createApiKey"]; + let key_id = key_data["id"].as_str().expect("should have id"); + let secret = key_data["secret"].as_str().expect("should have secret"); + assert!(secret.starts_with("flow_sa_")); + + // === valid_for validation === + // Each case must be rejected, and the error message identifies the + // specific branch: non-ISO syntax, malformed ISO, interval overflow, + // non-positive, and over the one-year cap. + for (valid_for, want) in [ + ("90 days", "ISO 8601"), // Postgres syntax, not ISO 8601 + ("Pfoo", "invalid valid_for"), // 'P'-prefixed but unparseable + ("P300000000000Y", "invalid valid_for"), // overflows interval parsing (SQLSTATE 22015) + ("P0D", "positive"), // zero duration + ("P2Y", "no greater than 1 year"), // exceeds the cap + ] { + let rejected: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($label: String!, $validFor: String!) { + createApiKey(catalogName: "aliceCo/ci-deploy-bot", label: $label, validFor: $validFor) { id } + }"#, + "variables": { + "label": "bad valid_for", + "validFor": valid_for, + } + }), + Some(&alice_token), + ) + .await; + assert!( + rejected["errors"][0]["message"] + .as_str() + .unwrap_or_default() + .contains(want), + "valid_for {valid_for:?} should be rejected mentioning {want:?}: {rejected}" + ); + } + + // === The API key authenticates directly as a bearer credential === + // The Envelope exchanges it for a short-lived signed access token and + // verifies that, resolving to the service account's identity. The + // refreshTokens listing is empty (the account owns none), but a data + // response proves authentication succeeded. + let via_bearer: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#"query { refreshTokens { edges { node { id } } } }"# + }), + Some(secret), + ) + .await; + assert!( + via_bearer["data"]["refreshTokens"]["edges"].is_array(), + "bearer-authenticated request should succeed: {via_bearer}" + ); + + // === The API key can also be exchanged for an access token === + // POST /api/v1/auth/token with an `api_key` grant statefully verifies + // the key and returns a signed JWT (no refresh token — the key is the + // durable credential). The minted token then authenticates a request + // by its signature, resolving to the same service-account identity. + let exchanged = server + .rest_client() + .post( + "/api/v1/auth/token", + &serde_json::json!({ "grant_type": "api_key", "api_key": secret }), + None, + ) + .send() + .await + .unwrap(); + assert_eq!( + exchanged.status(), + reqwest::StatusCode::OK, + "api_key exchange should succeed" + ); + let exchanged: serde_json::Value = exchanged.json().await.unwrap(); + let access_token = exchanged["access_token"] + .as_str() + .expect("exchange returns an access_token"); + assert!( + exchanged["refresh_token"].is_null(), + "api_key exchange returns no refresh token: {exchanged}" + ); + + // The exchanged token is valid for one hour. + use base64::Engine; + let payload = access_token.split('.').nth(1).expect("jwt has a payload"); + let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload) + .expect("jwt payload is base64url"); + let payload: serde_json::Value = serde_json::from_slice(&payload).unwrap(); + let (iat, exp) = ( + payload["iat"].as_u64().unwrap(), + payload["exp"].as_u64().unwrap(), + ); + assert_eq!( + exp - iat, + 3600, + "exchanged token should be valid for one hour: {payload}" + ); + + let via_jwt: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#"query { refreshTokens { edges { node { id } } } }"# + }), + Some(access_token), + ) + .await; + assert!( + via_jwt["data"]["refreshTokens"]["edges"].is_array(), + "exchanged access token should authenticate: {via_jwt}" + ); + + // === List service accounts === + let list: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + query { + serviceAccounts { + edges { + node { + catalogName + lastUsedAt + apiKeys { + id + label + createdBy + createdAt + expiresAt + lastUsedAt + } + } + } + } + }"# + }), + Some(&alice_token), + ) + .await; + + let edges = list["data"]["serviceAccounts"]["edges"] + .as_array() + .expect("should have edges"); + assert_eq!(edges.len(), 1); + assert_eq!(edges[0]["node"]["catalogName"], "aliceCo/ci-deploy-bot"); + let listed_key = &edges[0]["node"]["apiKeys"][0]; + assert_eq!(edges[0]["node"]["apiKeys"].as_array().unwrap().len(), 1); + assert_eq!(listed_key["label"], "GitHub Actions"); + assert_eq!( + listed_key["createdBy"], "11111111-1111-1111-1111-111111111111", + "key createdBy should be the calling admin: {list}" + ); + assert!( + listed_key["createdAt"].is_string() && listed_key["expiresAt"].is_string(), + "key createdAt/expiresAt should be set: {list}" + ); + // The key was presented as a bearer credential above, so its + // last_used_at is now populated — the stamp is fused into the bearer + // verification query. + assert!( + listed_key["lastUsedAt"].is_string(), + "lastUsedAt should be set after a successful bearer use: {list}" + ); + // The account's lastUsedAt is derived as the max across its keys' last_used_at. + assert!( + edges[0]["node"]["lastUsedAt"].is_string(), + "account lastUsedAt should be derived from its keys' use: {list}" + ); + + // Bob sees no service accounts. + let bob_list: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + query { + serviceAccounts { edges { node { catalogName } } } + }"# + }), + Some(&bob_token), + ) + .await; + + let bob_edges = bob_list["data"]["serviceAccounts"]["edges"] + .as_array() + .expect("should have edges"); + assert_eq!(bob_edges.len(), 0); + + // === Revoke the API key === + let revoke: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($keyId: Id!) { + revokeApiKey(id: $keyId) + }"#, + "variables": { "keyId": key_id } + }), + Some(&alice_token), + ) + .await; + + assert!( + revoke["errors"].is_null(), + "revoke should succeed: {revoke}" + ); + + // The row is preserved with revoked_at stamped — revocation is a soft + // delete for audit purposes. The exchange and listing assertions below + // can't observe this distinction, so check the table directly. + let parsed_key_id: models::Id = key_id.parse().unwrap(); + let revoked_row = sqlx::query!( + r#"SELECT revoked_at FROM internal.api_keys WHERE id = $1"#, + parsed_key_id as models::Id, + ) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + revoked_row.revoked_at.is_some(), + "revocation must stamp revoked_at, not delete the row" + ); + + // Revoking again fails: already-revoked keys are treated as not found. + let revoke_again: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($keyId: Id!) { + revokeApiKey(id: $keyId) + }"#, + "variables": { "keyId": key_id } + }), + Some(&alice_token), + ) + .await; + assert!( + revoke_again["errors"][0]["message"] + .as_str() + .unwrap_or_default() + .contains("API key not found"), + "re-revoking should report not found: {revoke_again}" + ); + + // The revoked key no longer authenticates — immediately, since every + // request re-verifies the credential against the database. Revoked + // keys are excluded from the verification query, so this falls + // through to the same 401 rejection as a nonexistent key. + let rejected = server + .rest_client() + .post( + "/api/graphql", + &serde_json::json!({ "query": "query { refreshTokens { edges { node { id } } } }" }), + Some(secret), + ) + .send() + .await + .unwrap(); + let status = rejected.status(); + let body = rejected.text().await.unwrap(); + assert_eq!( + status, + reqwest::StatusCode::UNAUTHORIZED, + "revoked key should be rejected with 401: {body}" + ); + assert!( + body.contains("invalid, expired, or revoked api key"), + "revoked key rejection body: {body}" + ); + + // Exchange routes through the same stateful verification, so a revoked + // key can't be traded for an access token either. + let exchange_revoked = server + .rest_client() + .post( + "/api/v1/auth/token", + &serde_json::json!({ "grant_type": "api_key", "api_key": secret }), + None, + ) + .send() + .await + .unwrap(); + assert_eq!( + exchange_revoked.status(), + reqwest::StatusCode::UNAUTHORIZED, + "revoked key must not be exchangeable for an access token" + ); + + // The revoked key is excluded from listings, even though its row remains. + let list_after_revoke: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + query { + serviceAccounts { edges { node { apiKeys { id } } } } + }"# + }), + Some(&alice_token), + ) + .await; + assert_eq!( + list_after_revoke["data"]["serviceAccounts"]["edges"][0]["node"]["apiKeys"] + .as_array() + .unwrap() + .len(), + 0, + "revoked keys must not appear in listings: {list_after_revoke}" + ); + + // Count the service account's user_grants rows directly: the + // grant-management assertions below observe access changes through + // this, since bearer authentication succeeds whether or not grants + // remain (access is wholly determined by user_grants). + async fn grant_count(pool: &sqlx::PgPool, user_id: &str) -> i64 { + sqlx::query_scalar!( + r#"SELECT count(*) AS "count!" FROM public.user_grants WHERE user_id = $1"#, + uuid::Uuid::parse_str(user_id).unwrap(), + ) + .fetch_one(pool) + .await + .unwrap() + } + + // The public API addresses service accounts by catalog name; tests that + // assert at the row level still need the backing user_id. + async fn service_account_user_id(pool: &sqlx::PgPool, catalog_name: &str) -> String { + sqlx::query_scalar!( + r#"SELECT user_id FROM internal.service_accounts WHERE catalog_name = $1::text::catalog_name"#, + catalog_name, + ) + .fetch_one(pool) + .await + .unwrap() + .to_string() + } + + // === Grant management === + // Adding a grant requires BOTH managing the account and admin on the + // granted prefix. Bob has neither, so he can't add a grant. + let add_unmanaged: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + addServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "aliceCo/", capability: read) + }"# + }), + Some(&bob_token), + ) + .await; + assert!( + add_unmanaged["errors"].is_array(), + "a non-manager must not add grants: {add_unmanaged}" + ); + + // Alice manages the account but doesn't admin bobCo/: extending the + // account's access beyond what she administers is denied. + let add_foreign: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + addServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "bobCo/", capability: read) + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + add_foreign["errors"].is_array(), + "a grant to an unadministered prefix must be rejected: {add_foreign}" + ); + + // Happy path: alice manages the account and admins aliceCo/ops/. + let add: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + addServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "aliceCo/ops/", capability: write) + }"# + }), + Some(&alice_token), + ) + .await; + assert!(add["errors"].is_null(), "add grant should succeed: {add}"); + assert_eq!(grant_count(&pool, &sa_user_id).await, 3); + + // Removal requires only account management — no capability on the + // grant's prefix. Seed a grant to bobCo/ directly (adding one via the + // API requires admin on it), then alice removes it despite having no + // bobCo/ access of her own. + sqlx::query("INSERT INTO user_grants (user_id, object_role, capability) VALUES ($1, 'bobCo/', 'read')") + .bind(uuid::Uuid::parse_str(&sa_user_id).unwrap()) + .execute(&pool) + .await + .unwrap(); + + let remove_foreign: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + removeServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "bobCo/") + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + remove_foreign["errors"].is_null(), + "a manager may remove ANY grant, including one to a prefix they don't administer: {remove_foreign}" + ); + assert_eq!(grant_count(&pool, &sa_user_id).await, 3); + + // Removing an absent grant reports not found. + let remove_again: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + removeServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "bobCo/") + }"# + }), + Some(&alice_token), + ) + .await; + assert!( + remove_again["errors"][0]["message"] + .as_str() + .unwrap_or_default() + .contains("grant not found"), + "re-removing should report not found: {remove_again}" + ); + + // Bob cannot remove grants of an account he doesn't manage. + let remove_unmanaged: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + removeServiceAccountGrant(catalogName: "aliceCo/ci-deploy-bot", prefix: "aliceCo/data/") + }"# + }), + Some(&bob_token), + ) + .await; + assert!(remove_unmanaged["errors"].is_array()); + } + + /// The management gates accept the fine-grained capabilities the feature + /// defines, not only the full `Admin` bundle: a caller holding `TeamAdmin` + /// (which confers `ManageServiceAccount` + `CreateGrant`) but NOT `Admin` + /// can manage service accounts, while the per-grant `CreateGrant` check + /// still bounds how far they can extend an account's reach. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../../../fixtures", scripts("data_planes", "alice")) + )] + async fn test_team_admin_manages_without_full_admin(pool: sqlx::PgPool) { + let _guard = test_server::init(); + + // Carol holds the TeamAdmin bundle on aliceCo/ and nothing else: her + // grant carries no legacy capability ('none'), so her bits come solely + // from the bundle — ManageServiceAccount and CreateGrant, but none of + // the wider Admin-bundle bits. This is the caller class the gates were + // narrowed to admit. Seeded before the snapshot so authorization + // observes it. + let carol_uid = uuid::Uuid::from_bytes([0x33; 16]); + sqlx::query("INSERT INTO auth.users (id, email) VALUES ($1, 'carol@example.test')") + .bind(carol_uid) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO public.user_grants (user_id, object_role, capability, bundles) + VALUES ($1, 'aliceCo/', 'none', ARRAY['team_admin']::capability_bundle[])", + ) + .bind(carol_uid) + .execute(&pool) + .await + .unwrap(); + + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + + let carol_token = server.make_access_token(carol_uid, Some("carol@example.test")); + + // Create succeeds: the anchor gate accepts ManageServiceAccount, and + // the per-grant gate accepts CreateGrant on aliceCo/data/ (covered by + // Carol's aliceCo/ bundle) — all without her holding full Admin. + let create: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation($grants: [ServiceAccountGrantInput!]!) { + createServiceAccount( + catalogName: "aliceCo/team-bot" + grants: $grants + ) { catalogName createdBy } + }"#, + "variables": { + "grants": [ { "prefix": "aliceCo/data/", "capability": "read" } ] + } + }), + Some(&carol_token), + ) + .await; + assert!( + create["errors"].is_null(), + "a TeamAdmin without full Admin should create a service account: {create}" + ); + assert_eq!( + create["data"]["createServiceAccount"]["createdBy"], + "33333333-3333-3333-3333-333333333333", + "createdBy should be the calling team admin: {create}" + ); + + // The anchor-only mutation createApiKey also accepts ManageServiceAccount. + let key: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createApiKey(catalogName: "aliceCo/team-bot", label: "ci", validFor: "P30D") { id } + }"# + }), + Some(&carol_token), + ) + .await; + assert!( + key["errors"].is_null(), + "a TeamAdmin should mint an API key: {key}" + ); + + // addServiceAccountGrant to a prefix Carol can confer (she holds + // CreateGrant across aliceCo/) succeeds. + let add_ok: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + addServiceAccountGrant(catalogName: "aliceCo/team-bot", prefix: "aliceCo/ops/", capability: write) + }"# + }), + Some(&carol_token), + ) + .await; + assert!( + add_ok["errors"].is_null(), + "granting a prefix the team admin can confer should succeed: {add_ok}" + ); + + // Anti-escalation: Carol lacks CreateGrant on bobCo/, so she cannot + // extend the account there — managing an account does not let her widen + // its reach beyond what she could grant. This is the boundary now + // sitting at CreateGrant rather than Admin. + let add_escalation: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + addServiceAccountGrant(catalogName: "aliceCo/team-bot", prefix: "bobCo/", capability: read) + }"# + }), + Some(&carol_token), + ) + .await; + assert!( + add_escalation["errors"].is_array(), + "a team admin must not grant a prefix she lacks CreateGrant on: {add_escalation}" + ); + + // The same boundary binds at creation time: seeding a foreign grant fails. + let create_escalation: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + mutation { + createServiceAccount( + catalogName: "aliceCo/overreach-bot" + grants: [{ prefix: "bobCo/", capability: read }] + ) { catalogName } + }"# + }), + Some(&carol_token), + ) + .await; + assert!( + create_escalation["errors"].is_array(), + "seeding a grant beyond the team admin's CreateGrant must fail: {create_escalation}" + ); + } +} diff --git a/crates/control-plane-api/src/server/public/token_exchange.rs b/crates/control-plane-api/src/server/public/token_exchange.rs index e3f4ed0fd87..aa8c431765b 100644 --- a/crates/control-plane-api/src/server/public/token_exchange.rs +++ b/crates/control-plane-api/src/server/public/token_exchange.rs @@ -8,6 +8,11 @@ pub enum TokenRequest { refresh_token_id: models::Id, secret: String, }, + #[serde(rename = "api_key")] + ApiKey { + // The full `flow_sa_...` service-account API key. + api_key: String, + }, } #[derive(Debug, serde::Serialize, serde::Deserialize, schemars::JsonSchema)] @@ -34,7 +39,36 @@ pub async fn handle_post_token( refresh_token_id, secret, } => exchange_refresh_token(&app, refresh_token_id, &secret).await, + TokenRequest::ApiKey { api_key } => exchange_api_key(&app, &api_key).await, + } +} + +// Exchange a service-account API key for a short-lived access token. This is +// the standard way to authenticate a service account: present the long-lived +// key once, then carry the signed token. Presenting the key directly as an +// `Authorization: Bearer` credential is also supported, as a convenience for +// one-off requests; that path mints the same token internally, so both route +// through `server::exchange_api_key`. +async fn exchange_api_key( + app: &crate::App, + api_key: &str, +) -> Result, crate::ApiError> { + // `verify_api_key` dispatches on this prefix via `expect`; reject a + // non-key here so a malformed request is a 400 rather than a panic. + if !api_key.starts_with("flow_sa_") { + return Err(crate::ApiError::Status(tonic::Status::invalid_argument( + "api_key must be a flow_sa_ service-account key", + ))); } + + let access_token = + crate::server::exchange_api_key(&app.pg_pool, &app.control_plane_jwt_encode_key, api_key) + .await?; + + Ok(axum::Json(TokenResponse { + access_token, + refresh_token: None, + })) } // Exchange a refresh token for an access token. diff --git a/crates/flow-client/control-plane-api.graphql b/crates/flow-client/control-plane-api.graphql index 353b01be592..eee0fbfeefe 100644 --- a/crates/flow-client/control-plane-api.graphql +++ b/crates/flow-client/control-plane-api.graphql @@ -256,6 +256,15 @@ input AlertsBy { active: Boolean } +type ApiKeyInfo { + id: Id! + label: String! + createdBy: UUID! + createdAt: DateTime! + expiresAt: DateTime! + lastUsedAt: DateTime +} + type AutoDiscoverFailure { """ The number of consecutive failures that have been observed. @@ -372,6 +381,7 @@ enum CapabilityBit { CreateInviteLink ViewDataPlanePrivateNetworking ModifyDataPlanePrivateNetworking + ManageServiceAccount Delegate Assume } @@ -633,6 +643,11 @@ type Controller { updatedAt: DateTime! } +type CreateApiKeyResult { + id: Id! + secret: String! +} + type CreateBillingSetupIntentPayload { clientSecret: String! } @@ -1286,6 +1301,69 @@ type MutationRoot { Already-zeroed (revoked) tokens are treated as not found. """ revokeRefreshToken(id: Id!): Boolean! + """ + Create a service account homed at the specified catalog name, seeded + with the given user_grants. + + `catalogName` is a management anchor: admins of a prefix covering it + may manage the account. It determines who may manage the account, not + what the account may access. Access is determined solely by the + account's user_grants, which may span multiple prefixes. + + The caller must have ManageServiceAccount on the catalog name AND + CreateGrant on each granted prefix. Creates an auth.users row, an + internal.service_accounts row, and a user_grants row per requested + grant. + """ + createServiceAccount(catalogName: Name!, grants: [ServiceAccountGrantInput!]!): ServiceAccount! + """ + Add a user_grant to a service account. + + The caller must manage the service account (ManageServiceAccount on its + catalog name) AND have CreateGrant on the granted prefix. The second + requirement prevents a caller from extending an account's access beyond + what they could grant anyone. (Human-user grant creation still lives in + PostgREST; when it migrates to GraphQL it should gate on this same + CreateGrant capability.) + """ + addServiceAccountGrant(catalogName: Name!, prefix: Prefix!, capability: Capability!): Boolean! + """ + Remove a user_grant from a service account. + + The caller must manage the service account (ManageServiceAccount on its + catalog name). Unlike addServiceAccountGrant, no capability on the + grant's prefix is required: removal only ever narrows the account's + access, so managers may remove ANY grant — including grants to + prefixes they don't themselves administer. + """ + removeServiceAccountGrant(catalogName: Name!, prefix: Prefix!): Boolean! + """ + Create an API key for a service account. + + Returns the key_id and the plaintext secret (flow_sa_...). + The secret is returned exactly once and cannot be retrieved again. + + The API key can be exchanged for an 1-hr access token via `POST /api/v1/auth/token` + or used directly as an `Authorization: Bearer` credential + """ + createApiKey( catalogName: Name!, label: String!, + """ + ISO 8601 duration for key validity (e.g. P90D, P1Y) + """ + validFor: String! + ): CreateApiKeyResult! + """ + Revoke an API key. + + The caller must have ManageServiceAccount capability on the owning service account's + catalog name. + + Rather than deleting the row, we stamp `revoked_at`, which makes the key + inert (excluded from bearer authentication and listings) while + preserving the audit trail. Already-revoked keys are treated as not + found. + """ + revokeApiKey(id: Id!): Boolean! } """ @@ -1601,6 +1679,7 @@ type QueryRoot { List refresh tokens owned by the authenticated user. """ refreshTokens(after: String, first: Int): RefreshTokenInfoConnection! + serviceAccounts(after: String, first: Int): ServiceAccountConnection! } """ @@ -1679,6 +1758,48 @@ type RepublishRequested { lastBuildId: Id! } +type ServiceAccount { + catalogName: Name! + createdBy: UUID! + createdAt: DateTime! + updatedAt: DateTime! + lastUsedAt: DateTime + apiKeys: [ApiKeyInfo!]! +} + +type ServiceAccountConnection { + """ + Information to aid in pagination. + """ + pageInfo: PageInfo! + """ + A list of edges. + """ + edges: [ServiceAccountEdge!]! +} + +""" +An edge in a connection. +""" +type ServiceAccountEdge { + """ + The item at the end of the edge + """ + node: ServiceAccount! + """ + A cursor for use in pagination + """ + cursor: String! +} + +""" +A user_grant to seed a service account with at creation time. +""" +input ServiceAccountGrantInput { + prefix: Prefix! + capability: Capability! +} + """ The shape of a connector status, which matches that of an ops::Log. """ diff --git a/crates/models/src/authz.rs b/crates/models/src/authz.rs index 0cc9e48a54d..f7df620d8be 100644 --- a/crates/models/src/authz.rs +++ b/crates/models/src/authz.rs @@ -26,6 +26,7 @@ pub enum Capability { // `ModifyDataPlanePrivateNetworking` permits mutating that same // configuration; the data-plane controller converges to it. ModifyDataPlanePrivateNetworking, + ManageServiceAccount, Delegate, Assume, } @@ -108,7 +109,7 @@ impl CapabilityBundle { | Self::ManageDataPlane.capabilities() } Self::Billing => EnumSet::empty(), - Self::TeamAdmin => CreateGrant | DeleteGrant | CreateInviteLink, + Self::TeamAdmin => CreateGrant | DeleteGrant | CreateInviteLink | ManageServiceAccount, Self::ManageDataPlane => { ViewDataPlanePrivateNetworking | ModifyDataPlanePrivateNetworking } diff --git a/supabase/migrations/00_polyfill.sql b/supabase/migrations/00_polyfill.sql index b31ebc82fb8..b691d574783 100644 --- a/supabase/migrations/00_polyfill.sql +++ b/supabase/migrations/00_polyfill.sql @@ -27,8 +27,8 @@ BEGIN -- Roles are cluster-wide and already exist from migrations applied to the -- primary `postgres` database. We only need to stub Supabase schemas. - -- pgcrypto provides crypt()/gen_salt(), used for refresh-token - -- secret hashing. Real Supabase pre-installs it in the + -- pgcrypto provides crypt()/gen_salt(), used by refresh-token and + -- API-key secret hashing. Real Supabase pre-installs it in the -- `extensions` schema; here we install into `public` (on the default -- search_path) so the migrations' unqualified calls resolve. create extension if not exists pgcrypto with schema public; diff --git a/supabase/migrations/20260615120000_service_accounts.sql b/supabase/migrations/20260615120000_service_accounts.sql new file mode 100644 index 00000000000..74519e507e5 --- /dev/null +++ b/supabase/migrations/20260615120000_service_accounts.sql @@ -0,0 +1,57 @@ +begin; + +-- Service accounts: non-login identities that authenticate via API keys. +-- Used for programmatic access (CI/CD, automation) and scoped, time-limited +-- access grants. + +create table internal.service_accounts ( + user_id uuid primary key references auth.users (id), + -- `catalog_name` is a management anchor only: admins of a prefix covering + -- this name may manage the service account (mint keys, revoke grants) and + -- see it in listings. The account's *access* is determined solely by its + -- user_grants rows, which are managed like any other user's and may span + -- multiple prefixes. + catalog_name public.catalog_name not null, + created_by uuid not null references auth.users (id), + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); + +comment on table internal.service_accounts is + 'Non-login identities that authenticate via API keys and are authorized through user_grants.'; + +-- A catalog name is the service account's handle: it identifies the account +-- within this table. +create unique index service_accounts_catalog_name_key on internal.service_accounts + (catalog_name); + +-- The serviceAccounts query scopes results to a caller's admin prefixes with +-- `catalog_name::text ^@ ANY($1)`. SP-GiST natively supports the `^@` +-- (starts-with) operator; the unique btree above would not be used by it. +create index service_accounts_catalog_name_spgist on internal.service_accounts + using spgist ((catalog_name::text)); + +-- API keys: long-lived credentials for service accounts exchanged for a short-lived (1-hour) access token. + +create table internal.api_keys ( + id public.flowid primary key not null default internal.id_generator(), + service_account_id uuid not null references internal.service_accounts (user_id), + -- Hex-encoded SHA-256 of the key secret. + secret_hash text not null, + label text not null, + expires_at timestamptz not null, + created_by uuid not null references auth.users (id), + last_used_at timestamptz, + created_at timestamptz not null default now(), + -- Revocation stamps this rather than deleting the row, preserving the + -- audit trail. Revoked keys are inert: excluded from bearer authentication + -- and from listings. + revoked_at timestamptz +); + +create index api_keys_service_account_id on internal.api_keys (service_account_id); + +comment on table internal.api_keys is + 'Long-lived service-account credentials, exchanged for a short-lived access token.'; + +commit;