fix(query): add JWT verification time tolerance

src/query/users/src/jwt/authenticator.rs

@@ -26,6 +26,7 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::jwk;
+use super::key_pair::verification_options;
 
 #[derive(Debug, Clone)]
 pub enum PubKey {
@@ -125,8 +126,12 @@ impl JwtAuthenticator {
         pub_key: &PubKey,
     ) -> Result<JWTClaims<CustomClaims>> {
         let result = match pub_key {
-            PubKey::RSA256(pk) => pk.verify_token::<CustomClaims>(token, None),
-            PubKey::ES256(pk) => pk.verify_token::<CustomClaims>(token, None),
+            PubKey::RSA256(pk) => {
+                pk.verify_token::<CustomClaims>(token, Some(verification_options()))
+            }
+            PubKey::ES256(pk) => {
+                pk.verify_token::<CustomClaims>(token, Some(verification_options()))
+            }
         };
         let claims = result.map_err(|err| ErrorCode::AuthenticateFailure(err.to_string()))?;
         match claims.subject {

src/query/users/src/jwt/key_pair.rs

@@ -27,10 +27,22 @@ use jwt_simple::algorithms::RS512PublicKey;
 use jwt_simple::algorithms::RSAPublicKeyLike;
 use jwt_simple::prelude::JWTClaims;
 use jwt_simple::prelude::NoCustomClaims;
+use jwt_simple::prelude::VerificationOptions;
 
 /// Minimum RSA key size in bits.
 const RSA_MIN_KEY_BITS: usize = 2048;
 
+const JWT_TIME_TOLERANCE_SECS: u64 = 5;
+
+pub(crate) fn verification_options() -> VerificationOptions {
+    VerificationOptions {
+        time_tolerance: Some(jwt_simple::prelude::Duration::from_secs(
+            JWT_TIME_TOLERANCE_SECS,
+        )),
+        ..Default::default()
+    }
+}
+
 pub enum PublicKeyType {
     RSA(String), // PEM string; parsed into RS256/RS384/RS512 at verify time
     ES256(ES256PublicKey),
@@ -91,27 +103,39 @@ fn verify_token_with_key(
             // which hash the signer used; we attempt all three since the underlying
             // RSA key is the same.
             if let Ok(k) = RS256PublicKey::from_pem(pem) {
-                if let Ok(claims) = k.verify_token::<NoCustomClaims>(token, None) {
+                if let Ok(claims) =
+                    k.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+                {
                     return Ok(claims);
                 }
             }
             if let Ok(k) = RS384PublicKey::from_pem(pem) {
-                if let Ok(claims) = k.verify_token::<NoCustomClaims>(token, None) {
+                if let Ok(claims) =
+                    k.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+                {
                     return Ok(claims);
                 }
             }
             if let Ok(k) = RS512PublicKey::from_pem(pem) {
-                if let Ok(claims) = k.verify_token::<NoCustomClaims>(token, None) {
+                if let Ok(claims) =
+                    k.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+                {
                     return Ok(claims);
                 }
             }
             Err(jwt_simple::Error::msg(
                 "RSA signature verification failed with RS256, RS384, and RS512",
             ))
         }
-        PublicKeyType::ES256(pk) => pk.verify_token::<NoCustomClaims>(token, None),
-        PublicKeyType::ES384(pk) => pk.verify_token::<NoCustomClaims>(token, None),
-        PublicKeyType::Ed25519(pk) => pk.verify_token::<NoCustomClaims>(token, None),
+        PublicKeyType::ES256(pk) => {
+            pk.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+        }
+        PublicKeyType::ES384(pk) => {
+            pk.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+        }
+        PublicKeyType::Ed25519(pk) => {
+            pk.verify_token::<NoCustomClaims>(token, Some(verification_options()))
+        }
     }
 }
 

src/query/users/tests/it/jwt/authenticator.rs

@@ -79,6 +79,49 @@ async fn test_parse_non_custom_claim() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn test_jwks_jwt_time_tolerance() -> anyhow::Result<()> {
+    let (pair1, pbkey1) = get_jwks_file_rs256("test_kid");
+    let template1 = ResponseTemplate::new(200).set_body_raw(pbkey1, "application/json");
+    let server = MockServer::start().await;
+    let json_path = "/jwks.json";
+    Mock::given(method("GET"))
+        .and(path(json_path))
+        .respond_with(template1)
+        .expect(1..)
+        .mount(&server)
+        .await;
+    let first_url = format!("http://{}{}", server.address(), json_path);
+    let mut cfg = QueryConfig {
+        tenant_id: Tenant::new_literal("test-tenant"),
+        ..Default::default()
+    };
+    cfg.common.cluster_id = "test-cluster".to_string();
+    cfg.common.jwt_key_file = first_url;
+    cfg.common.jwks_refresh_interval = 86400;
+    cfg.common.jwks_refresh_timeout = 10;
+    let auth = JwtAuthenticator::create(&cfg, &BUILD_INFO).unwrap();
+
+    let now = Clock::now_since_epoch();
+    let user_name = "test-user";
+
+    let mut tolerated_claims =
+        Claims::with_custom_claims(CustomClaims::new(), Duration::from_hours(2))
+            .with_subject(user_name.to_string());
+    tolerated_claims.issued_at = Some(now + Duration::from_secs(4));
+    let tolerated_token = pair1.sign(tolerated_claims)?;
+    auth.parse_jwt_claims(&tolerated_token).await?;
+
+    let mut rejected_claims =
+        Claims::with_custom_claims(CustomClaims::new(), Duration::from_hours(2))
+            .with_subject(user_name.to_string());
+    rejected_claims.issued_at = Some(now + Duration::from_secs(30));
+    let rejected_token = pair1.sign(rejected_claims)?;
+    assert!(auth.parse_jwt_claims(&rejected_token).await.is_err());
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn test_parse_jwt_claims_with_ensure_user_scenarios() -> anyhow::Result<()> {
     let (pair1, pbkey1) = get_jwks_file_rs256("test_kid");

src/query/users/tests/it/jwt/key_pair.rs

@@ -0,0 +1,47 @@
+// Copyright 2026 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use databend_common_meta_app::principal::PublicKeyEntry;
+use databend_common_meta_app::principal::normalize_public_key;
+use databend_common_users::verify_key_pair_jwt;
+use jwt_simple::prelude::*;
+
+fn public_key_entry(public_key_pem: &str) -> anyhow::Result<PublicKeyEntry> {
+    Ok(PublicKeyEntry {
+        key: normalize_public_key(public_key_pem)?,
+        label: "test-key".to_string(),
+        created_at: 0,
+    })
+}
+
+#[test]
+fn test_key_pair_jwt_time_tolerance() -> anyhow::Result<()> {
+    let key_pair = RS256KeyPair::generate(2048)?;
+    let public_key = public_key_entry(&key_pair.public_key().to_pem()?)?;
+    let public_keys = [public_key];
+
+    let now = Clock::now_since_epoch();
+
+    let mut tolerated_claims = Claims::create(Duration::from_hours(2));
+    tolerated_claims.issued_at = Some(now + Duration::from_secs(4));
+    let tolerated_token = key_pair.sign(tolerated_claims)?;
+    verify_key_pair_jwt(&tolerated_token, &public_keys)?;
+
+    let mut rejected_claims = Claims::create(Duration::from_hours(2));
+    rejected_claims.issued_at = Some(now + Duration::from_secs(30));
+    let rejected_token = key_pair.sign(rejected_claims)?;
+    assert!(verify_key_pair_jwt(&rejected_token, &public_keys).is_err());
+
+    Ok(())
+}

src/query/users/tests/it/jwt/mod.rs

@@ -14,3 +14,4 @@
 
 mod authenticator;
 mod jwk;
+mod key_pair;