feat: add a new status code for "external" errors (#4775)

* feat: add a new status code for "external" errors

* Update src/auth/src/error.rs

Co-authored-by: shuiyisong <113876041+shuiyisong@users.noreply.github.com>

* support mysql cli cleartext auth

* resolve PR comments

---------

Co-authored-by: shuiyisong <113876041+shuiyisong@users.noreply.github.com>

src/auth/src/common.rs

@@ -75,6 +75,16 @@ pub enum Password<'a> {
     PgMD5(HashedPassword<'a>, Salt<'a>),
 }
 
+impl Password<'_> {
+    pub fn r#type(&self) -> &str {
+        match self {
+            Password::PlainText(_) => "plain_text",
+            Password::MysqlNativePassword(_, _) => "mysql_native_password",
+            Password::PgMD5(_, _) => "pg_md5",
+        }
+    }
+}
+
 pub fn auth_mysql(
     auth_data: HashedPassword,
     salt: Salt,

src/auth/src/error.rs

@@ -89,7 +89,7 @@ impl ErrorExt for Error {
             Error::FileWatch { .. } => StatusCode::InvalidArguments,
             Error::InternalState { .. } => StatusCode::Unexpected,
             Error::Io { .. } => StatusCode::StorageUnavailable,
-            Error::AuthBackend { .. } => StatusCode::Internal,
+            Error::AuthBackend { source, .. } => source.status_code(),
 
             Error::UserNotFound { .. } => StatusCode::UserNotFound,
             Error::UnsupportedPasswordType { .. } => StatusCode::UnsupportedPasswordType,

src/auth/src/user_provider.rs

@@ -57,6 +57,11 @@ pub trait UserProvider: Send + Sync {
         self.authorize(catalog, schema, &user_info).await?;
         Ok(user_info)
     }
+
+    /// Returns whether this user provider implementation is backed by an external system.
+    fn external(&self) -> bool {
+        false
+    }
 }
 
 fn load_credential_from_file(filepath: &str) -> Result<Option<HashMap<String, Vec<u8>>>> {

src/common/error/src/status_code.rs

@@ -38,6 +38,8 @@ pub enum StatusCode {
     Cancelled = 1005,
     /// Illegal state, can be exposed to users.
     IllegalState = 1006,
+    /// Caused by some error originated from external system.
+    External = 1007,
     // ====== End of common status code ================
 
     // ====== Begin of SQL related status code =========
@@ -162,7 +164,8 @@ impl StatusCode {
             | StatusCode::InvalidAuthHeader
             | StatusCode::AccessDenied
             | StatusCode::PermissionDenied
-            | StatusCode::RequestOutdated => false,
+            | StatusCode::RequestOutdated
+            | StatusCode::External => false,
         }
     }
 
@@ -177,7 +180,9 @@ impl StatusCode {
             | StatusCode::IllegalState
             | StatusCode::EngineExecuteQuery
             | StatusCode::StorageUnavailable
-            | StatusCode::RuntimeResourcesExhausted => true,
+            | StatusCode::RuntimeResourcesExhausted
+            | StatusCode::External => true,
+
             StatusCode::Success
             | StatusCode::Unsupported
             | StatusCode::InvalidArguments
@@ -256,7 +261,7 @@ macro_rules! define_into_tonic_status {
 pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
     match status_code {
         StatusCode::Success => Code::Ok,
-        StatusCode::Unknown => Code::Unknown,
+        StatusCode::Unknown | StatusCode::External => Code::Unknown,
         StatusCode::Unsupported => Code::Unimplemented,
         StatusCode::Unexpected
         | StatusCode::IllegalState

src/servers/src/http/error_result.rs

@@ -122,7 +122,8 @@ pub fn status_code_to_http_status(status_code: &StatusCode) -> HttpStatusCode {
         StatusCode::RegionNotReady
         | StatusCode::TableUnavailable
         | StatusCode::RegionBusy
-        | StatusCode::StorageUnavailable => HttpStatusCode::SERVICE_UNAVAILABLE,
+        | StatusCode::StorageUnavailable
+        | StatusCode::External => HttpStatusCode::SERVICE_UNAVAILABLE,
 
         StatusCode::Internal
         | StatusCode::Unexpected

src/servers/src/mysql/handler.rs

@@ -54,6 +54,9 @@ use crate::mysql::writer::{create_mysql_column, handle_err};
 use crate::query_handler::sql::ServerSqlQueryHandlerRef;
 use crate::SqlPlan;
 
+const MYSQL_NATIVE_PASSWORD: &str = "mysql_native_password";
+const MYSQL_CLEAR_PASSWORD: &str = "mysql_clear_password";
+
 // An intermediate shim for executing MySQL queries.
 pub struct MysqlInstanceShim {
     query_handler: ServerSqlQueryHandlerRef,
@@ -219,6 +222,19 @@ impl MysqlInstanceShim {
         let mut guard = self.prepared_stmts.write();
         let _ = guard.remove(&stmt_key);
     }
+
+    fn auth_plugin(&self) -> &str {
+        if self
+            .user_provider
+            .as_ref()
+            .map(|x| x.external())
+            .unwrap_or(false)
+        {
+            MYSQL_CLEAR_PASSWORD
+        } else {
+            MYSQL_NATIVE_PASSWORD
+        }
+    }
 }
 
 #[async_trait]
@@ -229,6 +245,14 @@ impl<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> for MysqlInstanceShi
         std::env::var("GREPTIMEDB_MYSQL_SERVER_VERSION").unwrap_or_else(|_| "8.4.2".to_string())
     }
 
+    fn default_auth_plugin(&self) -> &str {
+        self.auth_plugin()
+    }
+
+    async fn auth_plugin_for_username(&self, _user: &[u8]) -> &str {
+        self.auth_plugin()
+    }
+
     fn salt(&self) -> [u8; 20] {
         self.salt
     }
@@ -253,7 +277,17 @@ impl<W: AsyncWrite + Send + Sync + Unpin> AsyncMysqlShim<W> for MysqlInstanceShi
             let user_id = Identity::UserId(&username, addr.as_deref());
 
             let password = match auth_plugin {
-                "mysql_native_password" => Password::MysqlNativePassword(auth_data, salt),
+                MYSQL_NATIVE_PASSWORD => Password::MysqlNativePassword(auth_data, salt),
+                MYSQL_CLEAR_PASSWORD => {
+                    // The raw bytes received could be represented in C-like string, ended in '\0'.
+                    // We must "trim" it to get the real password string.
+                    let password = if let &[password @ .., 0] = &auth_data {
+                        password
+                    } else {
+                        auth_data
+                    };
+                    Password::PlainText(String::from_utf8_lossy(password).to_string().into())
+                }
                 other => {
                     error!("Unsupported mysql auth plugin: {}", other);
                     return false;

src/servers/src/mysql/writer.rs

@@ -328,7 +328,7 @@ pub fn create_mysql_column_def(schema: &SchemaRef) -> Result<Vec<Column>> {
 fn mysql_error_kind(status_code: &StatusCode) -> ErrorKind {
     match status_code {
         StatusCode::Success => ErrorKind::ER_YES,
-        StatusCode::Unknown => ErrorKind::ER_UNKNOWN_ERROR,
+        StatusCode::Unknown | StatusCode::External => ErrorKind::ER_UNKNOWN_ERROR,
         StatusCode::Unsupported => ErrorKind::ER_NOT_SUPPORTED_YET,
         StatusCode::Cancelled => ErrorKind::ER_QUERY_INTERRUPTED,
         StatusCode::RuntimeResourcesExhausted => ErrorKind::ER_OUT_OF_RESOURCES,

src/servers/src/postgres/types/error.rs

@@ -374,6 +374,7 @@ impl From<StatusCode> for PgErrorCode {
             StatusCode::Unsupported => PgErrorCode::Ec0A000,
             StatusCode::InvalidArguments => PgErrorCode::Ec22023,
             StatusCode::Cancelled => PgErrorCode::Ec57000,
+            StatusCode::External => PgErrorCode::Ec58000,
 
             StatusCode::Unknown
             | StatusCode::Unexpected