fix: make the JWT audit gate atomic again; refuse-only negative cache; JWKS key_ops

- Gate the audit with a conditional upsert (`DO UPDATE ... WHERE NOT
  guest_activity.jwt_entry RETURNING 1`) read with fetch_optional. The row comes
  back exactly once per email per day, decided by the conflicting tuple, so it
  keeps the atomicity `xmax = 0` had (no double audit when two first requests race
  on a metered instance, which takes no advisory lock) and still fires on the
  first JWT after an IdP sign-in created today's row. The prior CTE decided this
  from the statement snapshot and could double-audit.
- Negative-cache only a real allowance refusal (`PermissionDenied`); a transient
  DB error inside guest_admission denies this request but no longer locks the
  email out for 30 seconds.
- Refuse a JWKS key whose `key_ops` is present and omits `verify`: it is published
  for something other than signature verification. Unit-tested.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VF3v6LA9399gNphmZaHYG3

backend/.sqlx/query-0fc900f73ef119e4c89186cf120938a298bfe29afe1fd7111340252877f8b86c.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO guest_activity (email, workspace_id, day, jwt_entry)\n         VALUES ($1, $2, CURRENT_DATE, true)\n         ON CONFLICT (email, workspace_id, day)\n         DO UPDATE SET jwt_entry = true, last_seen_at = now()\n         WHERE NOT guest_activity.jwt_entry\n         RETURNING 1 AS \"audited!\"",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "audited!",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "0fc900f73ef119e4c89186cf120938a298bfe29afe1fd7111340252877f8b86c"
+}

backend/.sqlx/query-7c7298a76978f4e4d0f7b644ea6f84e01dda54b9342a02d6645a4462e5e6bb19.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "WITH prior AS (\n             SELECT jwt_entry FROM guest_activity\n             WHERE email = $1 AND workspace_id = $2 AND day = CURRENT_DATE\n         )\n         INSERT INTO guest_activity (email, workspace_id, day, jwt_entry)\n         VALUES ($1, $2, CURRENT_DATE, true)\n         ON CONFLICT (email, workspace_id, day)\n         DO UPDATE SET jwt_entry = true, last_seen_at = now()\n         RETURNING (NOT COALESCE((SELECT jwt_entry FROM prior), false)) AS \"first_jwt!\"",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "first_jwt!",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "7c7298a76978f4e4d0f7b644ea6f84e01dda54b9342a02d6645a4462e5e6bb19"
-}

backend/windmill-api-auth/src/auth.rs

@@ -759,34 +759,40 @@ async fn admit_and_record_guest_jwt(db: &DB, w_id: &str, email: &str, app_path:
     };
     // The allowance and the row that changes it, in one transaction: guest_admission
     // takes a transaction-scoped advisory lock, so the count check and the insert cannot
-    // race two strangers past the cap. A refusal (a stranger past the cap on a capped
-    // instance) or a DB error rolls the tx back and denies the guest.
-    if let Err(e) = windmill_common::workspaces::guest_admission(&mut *tx, email).await {
-        tracing::info!("guest JWT not admitted for {w_id}: {e:#}");
-        GUEST_JWT_REFUSED_CACHE.insert(cache_key, std::time::Instant::now());
-        return false;
+    // race two strangers past the cap. Only a real allowance refusal is negative-cached;
+    // a transient DB error denies this request but must not lock the email out for 30s.
+    match windmill_common::workspaces::guest_admission(&mut *tx, email).await {
+        Ok(()) => {}
+        Err(e @ windmill_common::error::Error::PermissionDenied(_)) => {
+            tracing::info!("guest JWT not admitted for {w_id}: {e:#}");
+            GUEST_JWT_REFUSED_CACHE.insert(cache_key, std::time::Instant::now());
+            return false;
+        }
+        Err(e) => {
+            tracing::error!("guest JWT allowance check failed for {w_id}: {e:#}");
+            return false;
+        }
     }
-    // `first_jwt` is true when this is the first JWT entry for the email today: a fresh
-    // row, or one an identity-provider sign-in created earlier with `jwt_entry = false`.
-    // The audit is gated on it, decided atomically in the upsert, so `users.login_guest`
-    // (entry=jwt) fires once a day even when today's row already existed.
+    // The conditional `WHERE NOT jwt_entry` flips the flag only on its false-to-true
+    // transition, so the upsert returns a row exactly once per email per day: on the
+    // fresh insert, or on the first JWT after an identity-provider sign-in created
+    // today's row with `jwt_entry = false`. The audit is gated on that, decided
+    // atomically by the conflicting tuple, so concurrent first requests (a metered
+    // instance takes no advisory lock) audit at most once.
     let first_jwt = sqlx::query_scalar!(
-        r#"WITH prior AS (
-             SELECT jwt_entry FROM guest_activity
-             WHERE email = $1 AND workspace_id = $2 AND day = CURRENT_DATE
-         )
-         INSERT INTO guest_activity (email, workspace_id, day, jwt_entry)
+        r#"INSERT INTO guest_activity (email, workspace_id, day, jwt_entry)
          VALUES ($1, $2, CURRENT_DATE, true)
          ON CONFLICT (email, workspace_id, day)
          DO UPDATE SET jwt_entry = true, last_seen_at = now()
-         RETURNING (NOT COALESCE((SELECT jwt_entry FROM prior), false)) AS "first_jwt!""#,
+         WHERE NOT guest_activity.jwt_entry
+         RETURNING 1 AS "audited!""#,
         email,
         w_id,
     )
-    .fetch_one(&mut *tx)
+    .fetch_optional(&mut *tx)
     .await;
     let first_jwt = match first_jwt {
-        Ok(v) => v,
+        Ok(v) => v.is_some(),
         Err(e) => {
             tracing::error!("recording guest JWT activity for {w_id}: {e:#}");
             return false;

backend/windmill-common/src/guest_jwt.rs

@@ -102,6 +102,16 @@ pub fn jwk_algorithms(jwk: &Jwk) -> Option<Vec<Algorithm>> {
     {
         return None;
     }
+    // A key that lists its operations must allow verifying signatures; otherwise it is
+    // published for something else (encryption, key wrapping) and is not ours to use.
+    if jwk
+        .common
+        .key_operations
+        .as_ref()
+        .is_some_and(|ops| !ops.contains(&jsonwebtoken::jwk::KeyOperations::Verify))
+    {
+        return None;
+    }
     match (&jwk.algorithm, jwk.common.algorithm) {
         (AlgorithmParameters::RSA(_), Some(alg)) if RSA_ALGORITHMS.contains(&alg) => Some(vec![alg]),
         (AlgorithmParameters::RSA(_), None) => Some(RSA_ALGORITHMS.to_vec()),
@@ -434,6 +444,14 @@ mod tests {
         assert_eq!(jwk_algorithms(&k), None);
     }
 
+    #[test]
+    fn key_ops_without_verify_is_refused() {
+        let enc = jwk(serde_json::json!({"kty":"RSA","key_ops":["encrypt"],"n":"aa","e":"AQAB"}));
+        assert_eq!(jwk_algorithms(&enc), None);
+        let ver = jwk(serde_json::json!({"kty":"RSA","key_ops":["verify"],"n":"aa","e":"AQAB"}));
+        assert_eq!(jwk_algorithms(&ver), Some(RSA_ALGORITHMS.to_vec()));
+    }
+
     // PUB1's coordinates and its matching PKCS8 private key, for the JWKS-derived
     // verification test.
     const PUB1_X: &str = "zAfqyCh34iYOCW0vg4ejq_zzJlzLSZScjnVyPjLGTao";