refactor: merge authenticate and authorize api (#1360)

* chore: add auth api

* chore: update pg using auth api

* chore: update grpc using auth api

* chore: update http using auth api

src/servers/src/auth.rs

@@ -35,6 +35,20 @@ pub trait UserProvider: Send + Sync {
     /// from a certain user to a certain catalog/schema is legal.
     /// This method should be called after [`authenticate`].
     async fn authorize(&self, catalog: &str, schema: &str, user_info: &UserInfo) -> Result<()>;
+
+    /// [`auth`] is a combination of [`authenticate`] and [`authorize`].
+    /// In most cases it's preferred for both convenience and performance.
+    async fn auth(
+        &self,
+        id: Identity<'_>,
+        password: Password<'_>,
+        catalog: &str,
+        schema: &str,
+    ) -> Result<UserInfo> {
+        let user_info = self.authenticate(id, password).await?;
+        self.authorize(catalog, schema, &user_info).await?;
+        Ok(user_info)
+    }
 }
 
 pub type UserProviderRef = Arc<dyn UserProvider>;

src/servers/src/grpc/handler.rs

@@ -98,11 +98,13 @@ impl GreptimeRequestHandler {
             })
             .context(NotFoundAuthHeaderSnafu)?;
 
-        let user_info = match auth_scheme {
+        match auth_scheme {
             AuthScheme::Basic(Basic { username, password }) => user_provider
-                .authenticate(
+                .auth(
                     Identity::UserId(&username, None),
                     Password::PlainText(&password),
+                    &query_ctx.current_catalog(),
+                    &query_ctx.current_schema(),
                 )
                 .await
                 .map_err(|e| Auth { source: e }),
@@ -111,15 +113,7 @@ impl GreptimeRequestHandler {
             }),
         }
         .map_err(|e| Status::unauthenticated(e.to_string()))?;
-
-        user_provider
-            .authorize(
-                &query_ctx.current_catalog(),
-                &query_ctx.current_schema(),
-                &user_info,
-            )
-            .await
-            .map_err(|e| Status::permission_denied(e.to_string()))
+        Ok(())
     }
 }
 

src/servers/src/http/authorize.rs

@@ -25,7 +25,7 @@ use tower_http::auth::AsyncAuthorizeRequest;
 
 use super::PUBLIC_APIS;
 use crate::auth::Error::IllegalParam;
-use crate::auth::{Identity, IllegalParamSnafu, InternalStateSnafu, UserProviderRef};
+use crate::auth::{Identity, IllegalParamSnafu, UserProviderRef};
 use crate::error::Error::Auth;
 use crate::error::{
     self, InvalidAuthorizationHeaderSnafu, InvisibleASCIISnafu, NotFoundInfluxAuthSnafu, Result,
@@ -77,21 +77,37 @@ where
                 return Ok(request);
             };
 
-            // do authenticate
-            match authenticate(&user_provider, &request).await {
-                Ok(user_info) => {
-                    request.extensions_mut().insert(user_info);
+            let (username, password) = match extract_username_and_password(&request) {
+                Ok((username, password)) => (username, password),
+                Err(e) => {
+                    error!("extract username and password failed: {}", e);
+                    return Err(unauthorized_resp());
+                }
+            };
+
+            let (catalog, schema) = match extract_catalog_and_schema(&request) {
+                Ok((catalog, schema)) => (catalog, schema),
+                Err(e) => {
+                    error!("extract catalog and schema failed: {}", e);
+                    return Err(unauthorized_resp());
+                }
+            };
+
+            match user_provider
+                .auth(
+                    Identity::UserId(username.as_str(), None),
+                    crate::auth::Password::PlainText(password.as_str()),
+                    catalog,
+                    schema,
+                )
+                .await
+            {
+                Ok(userinfo) => {
+                    request.extensions_mut().insert(userinfo);
+                    Ok(request)
                 }
                 Err(e) => {
                     error!("authenticate failed: {}", e);
-                    return Err(unauthorized_resp());
-                }
-            }
-
-            match authorize(&user_provider, &request).await {
-                Ok(_) => Ok(request),
-                Err(e) => {
-                    error!("authorize failed: {}", e);
                     Err(unauthorized_resp())
                 }
             }
@@ -99,27 +115,18 @@ where
     }
 }
 
-async fn authorize<B: Send + Sync + 'static>(
-    user_provider: &UserProviderRef,
+fn extract_catalog_and_schema<B: Send + Sync + 'static>(
     request: &Request<B>,
-) -> crate::auth::Result<()> {
+) -> crate::auth::Result<(&str, &str)> {
     // try get database name
     let query = request.uri().query().unwrap_or_default();
     let input_database = extract_db_from_query(query).context(IllegalParamSnafu {
         msg: "db not provided or corrupted",
     })?;
 
-    let (catalog, database) =
-        crate::parse_catalog_and_schema_from_client_database_name(input_database);
-
-    let user_info = request
-        .extensions()
-        .get::<UserInfo>()
-        .context(InternalStateSnafu {
-            msg: "no user info provided while authorizing",
-        })?;
-
-    user_provider.authorize(catalog, database, user_info).await
+    Ok(crate::parse_catalog_and_schema_from_client_database_name(
+        input_database,
+    ))
 }
 
 fn get_influxdb_credentials<B: Send + Sync + 'static>(
@@ -162,11 +169,10 @@ fn get_influxdb_credentials<B: Send + Sync + 'static>(
     }
 }
 
-async fn authenticate<B: Send + Sync + 'static>(
-    user_provider: &UserProviderRef,
+fn extract_username_and_password<B: Send + Sync + 'static>(
     request: &Request<B>,
-) -> Result<UserInfo> {
-    let (username, password) = if request.uri().path().contains("influxdb") {
+) -> Result<(Username, Password)> {
+    Ok(if request.uri().path().contains("influxdb") {
         // compatible with influxdb auth
         get_influxdb_credentials(request)?.context(NotFoundInfluxAuthSnafu)?
     } else {
@@ -175,14 +181,7 @@ async fn authenticate<B: Send + Sync + 'static>(
         match scheme {
             AuthScheme::Basic(username, password) => (username, password),
         }
-    };
-
-    Ok(user_provider
-        .authenticate(
-            Identity::UserId(&username, None),
-            crate::auth::Password::PlainText(&password),
-        )
-        .await?)
+    })
 }
 
 fn unauthorized_resp<RespBody>() -> Response<RespBody>

src/servers/src/postgres/auth_handler.rs

@@ -22,12 +22,10 @@ use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 use pgwire::messages::response::ErrorResponse;
 use pgwire::messages::startup::Authentication;
 use pgwire::messages::{PgWireBackendMessage, PgWireFrontendMessage};
-use session::context::{QueryContextRef, UserInfo};
-use snafu::ResultExt;
+use session::context::QueryContextRef;
 
 use super::PostgresServerHandler;
 use crate::auth::{Identity, Password, UserProviderRef};
-use crate::error;
 use crate::error::Result;
 use crate::query_handler::sql::ServerSqlQueryHandlerRef;
 
@@ -70,45 +68,34 @@ impl LoginInfo {
 }
 
 impl PgLoginVerifier {
-    async fn verify_pwd(&self, password: &str, login: &LoginInfo) -> Result<bool> {
-        if let Some(user_provider) = &self.user_provider {
-            let user_name = match &login.user {
-                Some(name) => name,
-                None => return Ok(false),
-            };
+    async fn auth(&self, login: &LoginInfo, password: &str) -> Result<bool> {
+        let user_provider = match &self.user_provider {
+            Some(provider) => provider,
+            None => return Ok(false),
+        };
 
-            // TODO(fys): pass user_info to context
-            let _user_info = user_provider
-                .authenticate(
-                    Identity::UserId(user_name, None),
-                    Password::PlainText(password),
-                )
-                .await
-                .context(error::AuthSnafu)?;
-        }
-        Ok(true)
-    }
+        let user_name = match &login.user {
+            Some(name) => name,
+            None => return Ok(false),
+        };
+        let catalog = match &login.catalog {
+            Some(name) => name,
+            None => return Ok(false),
+        };
+        let schema = match &login.schema {
+            Some(name) => name,
+            None => return Ok(false),
+        };
+
+        let _user_info = user_provider
+            .auth(
+                Identity::UserId(user_name, None),
+                Password::PlainText(password),
+                catalog,
+                schema,
+            )
+            .await?;
 
-    async fn authorize(&self, login: &LoginInfo) -> Result<bool> {
-        // at this time, username in login info should be valid
-        // TODO(shuiyisong): change to use actually user_info from session
-        if let Some(user_provider) = &self.user_provider {
-            let user_name = match &login.user {
-                Some(name) => name,
-                None => return Ok(false),
-            };
-            let catalog = match &login.catalog {
-                Some(name) => name,
-                None => return Ok(false),
-            };
-            let schema = match &login.schema {
-                Some(name) => name,
-                None => return Ok(false),
-            };
-            user_provider
-                .authorize(catalog, schema, &UserInfo::new(user_name))
-                .await?;
-        }
         Ok(true)
     }
 }
@@ -182,12 +169,10 @@ impl StartupHandler for PostgresServerHandler {
                 let pwd = pwd.into_password()?;
 
                 let login_info = LoginInfo::from_client_info(client);
+
                 // do authenticate
-                let authenticate_result = self
-                    .login_verifier
-                    .verify_pwd(pwd.password(), &login_info)
-                    .await;
-                if !matches!(authenticate_result, Ok(true)) {
+                let auth_result = self.login_verifier.auth(&login_info, pwd.password()).await;
+                if !matches!(auth_result, Ok(true)) {
                     return send_error(
                         client,
                         "FATAL",
@@ -196,17 +181,6 @@ impl StartupHandler for PostgresServerHandler {
                     )
                     .await;
                 }
-                // do authorize
-                let authorize_result = self.login_verifier.authorize(&login_info).await;
-                if !matches!(authorize_result, Ok(true)) {
-                    return send_error(
-                        client,
-                        "FATAL",
-                        "28P01",
-                        "password authorization failed".to_owned(),
-                    )
-                    .await;
-                }
                 set_query_context_from_client_info(client, self.query_ctx.clone());
                 auth::finish_authentication(client, self.param_provider.as_ref()).await;
             }