feat(pgwire): implement OAuth audience validation

src/frontend/src/meta_client.rs

@@ -171,6 +171,8 @@ pub trait FrontendMetaClient: Send + Sync {
     async fn expire_iceberg_table_snapshots(&self, sink_id: SinkId) -> Result<()>;
 
     async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse>;
+
+    fn cluster_id(&self) -> &str;
 }
 
 pub struct FrontendMetaClientImpl(pub MetaClient);
@@ -424,4 +426,8 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
     async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse> {
         self.0.refresh(request).await
     }
+
+    fn cluster_id(&self) -> &str {
+        self.0.cluster_id()
+    }
 }

src/frontend/src/session.rs

@@ -1548,7 +1548,10 @@ impl SessionManagerImpl {
                             salt,
                         }
                     } else if auth_info.encryption_type == EncryptionType::Oauth as i32 {
-                        UserAuthenticator::OAuth(auth_info.metadata.clone())
+                        UserAuthenticator::OAuth {
+                            metadata: auth_info.metadata.clone(),
+                            cluster_id: self.env.meta_client().cluster_id().to_owned(),
+                        }
                     } else {
                         return Err(Box::new(Error::new(
                             ErrorKind::Unsupported,

src/frontend/src/test_utils.rs

@@ -1236,6 +1236,10 @@ impl FrontendMetaClient for MockFrontendMetaClient {
     async fn refresh(&self, _request: RefreshRequest) -> RpcResult<RefreshResponse> {
         Ok(RefreshResponse { status: None })
     }
+
+    fn cluster_id(&self) -> &str {
+        "test-cluster-uuid"
+    }
 }
 
 #[cfg(test)]

src/utils/pgwire/Cargo.toml

@@ -39,6 +39,10 @@ tracing = "0.1"
 workspace-hack = { path = "../../workspace-hack" }
 
 [dev-dependencies]
+base64 = "0.22"
+jsonwebtoken = "9"
+rand = "0.8"
+rsa = "0.9"
 tempfile = "3"
 tokio-postgres = "0.7"
 

src/utils/pgwire/src/pg_protocol.rs

@@ -652,7 +652,7 @@ where
                     })?;
                 self.ready_for_query()?;
             }
-            UserAuthenticator::ClearText(_) | UserAuthenticator::OAuth(_) => {
+            UserAuthenticator::ClearText(_) | UserAuthenticator::OAuth { .. } => {
                 self.stream
                     .write_no_flush(&BeMessage::AuthenticationCleartextPassword)?;
             }