352 lines
12 KiB
Rust
352 lines
12 KiB
Rust
use std::collections::BTreeMap;
|
|
|
|
use sqlx::MySqlConnection;
|
|
use sqlx::prelude::*;
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::core::protocol::CompleteDatabaseNameResponse;
|
|
use crate::core::protocol::request_validation::GroupDenylist;
|
|
use crate::core::protocol::request_validation::validate_db_or_user_request;
|
|
use crate::core::types::DbOrUser;
|
|
use crate::core::types::MySQLDatabase;
|
|
use crate::core::types::MySQLUser;
|
|
use crate::{
|
|
core::{
|
|
common::UnixUser,
|
|
protocol::{
|
|
CreateDatabaseError, CreateDatabasesResponse, DropDatabaseError, DropDatabasesResponse,
|
|
ListAllDatabasesError, ListAllDatabasesResponse, ListDatabasesError,
|
|
ListDatabasesResponse,
|
|
},
|
|
},
|
|
server::{common::create_user_group_matching_regex, sql::quote_identifier},
|
|
};
|
|
|
|
// NOTE: this function is unsafe because it does no input validation.
|
|
pub(super) async fn unsafe_database_exists(
|
|
database_name: &str,
|
|
connection: &mut MySqlConnection,
|
|
) -> Result<bool, sqlx::Error> {
|
|
let result =
|
|
sqlx::query("SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?")
|
|
.bind(database_name)
|
|
.fetch_optional(connection)
|
|
.await;
|
|
|
|
if let Err(err) = &result {
|
|
tracing::error!(
|
|
"Failed to check if database '{}' exists: {:?}",
|
|
&database_name,
|
|
err
|
|
);
|
|
}
|
|
|
|
Ok(result?.is_some())
|
|
}
|
|
|
|
pub async fn complete_database_name(
|
|
database_prefix: &str,
|
|
unix_user: &UnixUser,
|
|
connection: &mut MySqlConnection,
|
|
_db_is_mariadb: bool,
|
|
group_denylist: &GroupDenylist,
|
|
) -> CompleteDatabaseNameResponse {
|
|
let result = sqlx::query(
|
|
r"
|
|
SELECT CAST(`SCHEMA_NAME` AS CHAR(64)) AS `database`
|
|
FROM `information_schema`.`SCHEMATA`
|
|
WHERE `SCHEMA_NAME` NOT IN ('information_schema', 'performance_schema', 'mysql', 'sys')
|
|
AND `SCHEMA_NAME` REGEXP ?
|
|
AND `SCHEMA_NAME` LIKE ?
|
|
",
|
|
)
|
|
.bind(create_user_group_matching_regex(unix_user, group_denylist))
|
|
.bind(format!("{database_prefix}%"))
|
|
.fetch_all(connection)
|
|
.await;
|
|
|
|
match result {
|
|
Ok(rows) => rows
|
|
.into_iter()
|
|
.filter_map(|row| {
|
|
let database: String = row.try_get("database").ok()?;
|
|
Some(database.into())
|
|
})
|
|
.collect(),
|
|
Err(err) => {
|
|
tracing::error!(
|
|
"Failed to complete database name for prefix '{}' and user '{}': {:?}",
|
|
database_prefix,
|
|
unix_user.username,
|
|
err
|
|
);
|
|
vec![]
|
|
}
|
|
}
|
|
}
|
|
|
|
pub async fn create_databases(
|
|
database_names: &[MySQLDatabase],
|
|
unix_user: &UnixUser,
|
|
connection: &mut MySqlConnection,
|
|
_db_is_mariadb: bool,
|
|
group_denylist: &GroupDenylist,
|
|
) -> CreateDatabasesResponse {
|
|
let mut results = BTreeMap::new();
|
|
|
|
for database_name in database_names.iter().cloned() {
|
|
if let Err(err) = validate_db_or_user_request(
|
|
&DbOrUser::Database(database_name.clone()),
|
|
unix_user,
|
|
group_denylist,
|
|
)
|
|
.map_err(CreateDatabaseError::ValidationError)
|
|
{
|
|
results.insert(database_name.clone(), Err(err));
|
|
continue;
|
|
}
|
|
|
|
match unsafe_database_exists(&database_name, &mut *connection).await {
|
|
Ok(true) => {
|
|
results.insert(
|
|
database_name.clone(),
|
|
Err(CreateDatabaseError::DatabaseAlreadyExists),
|
|
);
|
|
continue;
|
|
}
|
|
Err(err) => {
|
|
results.insert(
|
|
database_name.clone(),
|
|
Err(CreateDatabaseError::MySqlError(err.to_string())),
|
|
);
|
|
continue;
|
|
}
|
|
_ => {}
|
|
}
|
|
|
|
let result =
|
|
sqlx::query(format!("CREATE DATABASE {}", quote_identifier(&database_name)).as_str())
|
|
.execute(&mut *connection)
|
|
.await
|
|
.map(|_| ())
|
|
.map_err(|err| CreateDatabaseError::MySqlError(err.to_string()));
|
|
|
|
if let Err(err) = &result {
|
|
tracing::error!("Failed to create database '{}': {:?}", &database_name, err);
|
|
}
|
|
|
|
results.insert(database_name, result);
|
|
}
|
|
|
|
results
|
|
}
|
|
|
|
pub async fn drop_databases(
|
|
database_names: &[MySQLDatabase],
|
|
unix_user: &UnixUser,
|
|
connection: &mut MySqlConnection,
|
|
_db_is_mariadb: bool,
|
|
group_denylist: &GroupDenylist,
|
|
) -> DropDatabasesResponse {
|
|
let mut results = BTreeMap::new();
|
|
|
|
for database_name in database_names.iter().cloned() {
|
|
if let Err(err) = validate_db_or_user_request(
|
|
&DbOrUser::Database(database_name.clone()),
|
|
unix_user,
|
|
group_denylist,
|
|
)
|
|
.map_err(DropDatabaseError::ValidationError)
|
|
{
|
|
results.insert(database_name.clone(), Err(err));
|
|
continue;
|
|
}
|
|
|
|
match unsafe_database_exists(&database_name, &mut *connection).await {
|
|
Ok(false) => {
|
|
results.insert(
|
|
database_name.clone(),
|
|
Err(DropDatabaseError::DatabaseDoesNotExist),
|
|
);
|
|
continue;
|
|
}
|
|
Err(err) => {
|
|
results.insert(
|
|
database_name.clone(),
|
|
Err(DropDatabaseError::MySqlError(err.to_string())),
|
|
);
|
|
continue;
|
|
}
|
|
_ => {}
|
|
}
|
|
|
|
let result =
|
|
sqlx::query(format!("DROP DATABASE {}", quote_identifier(&database_name)).as_str())
|
|
.execute(&mut *connection)
|
|
.await
|
|
.map(|_| ())
|
|
.map_err(|err| DropDatabaseError::MySqlError(err.to_string()));
|
|
|
|
if let Err(err) = &result {
|
|
tracing::error!("Failed to drop database '{}': {:?}", &database_name, err);
|
|
}
|
|
|
|
results.insert(database_name, result);
|
|
}
|
|
|
|
results
|
|
}
|
|
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub struct DatabaseRow {
|
|
pub database: MySQLDatabase,
|
|
pub tables: Vec<String>,
|
|
pub users: Vec<MySQLUser>,
|
|
pub collation: Option<String>,
|
|
pub character_set: Option<String>,
|
|
pub size_bytes: u64,
|
|
}
|
|
|
|
impl FromRow<'_, sqlx::mysql::MySqlRow> for DatabaseRow {
|
|
fn from_row(row: &sqlx::mysql::MySqlRow) -> Result<Self, sqlx::Error> {
|
|
Ok(DatabaseRow {
|
|
database: row.try_get::<String, _>("database")?.into(),
|
|
tables: {
|
|
let s: Option<String> = row.try_get("tables")?;
|
|
s.and_then(|s| {
|
|
if s.is_empty() {
|
|
None
|
|
} else {
|
|
Some(s.split(',').map(std::borrow::ToOwned::to_owned).collect())
|
|
}
|
|
})
|
|
.unwrap_or_default()
|
|
},
|
|
users: {
|
|
let s: Option<String> = row.try_get("users")?;
|
|
s.and_then(|s| {
|
|
if s.is_empty() {
|
|
None
|
|
} else {
|
|
Some(s.split(',').map(|s| s.to_owned().into()).collect())
|
|
}
|
|
})
|
|
.unwrap_or_default()
|
|
},
|
|
collation: row.try_get::<Option<String>, _>("collation")?,
|
|
character_set: row.try_get::<Option<String>, _>("character_set")?,
|
|
size_bytes: row.try_get::<u64, _>("size_bytes")?,
|
|
})
|
|
}
|
|
}
|
|
|
|
pub async fn list_databases(
|
|
database_names: &[MySQLDatabase],
|
|
unix_user: &UnixUser,
|
|
connection: &mut MySqlConnection,
|
|
_db_is_mariadb: bool,
|
|
group_denylist: &GroupDenylist,
|
|
) -> ListDatabasesResponse {
|
|
let mut results = BTreeMap::new();
|
|
|
|
for database_name in database_names.iter().cloned() {
|
|
if let Err(err) = validate_db_or_user_request(
|
|
&DbOrUser::Database(database_name.clone()),
|
|
unix_user,
|
|
group_denylist,
|
|
)
|
|
.map_err(ListDatabasesError::ValidationError)
|
|
{
|
|
results.insert(database_name.clone(), Err(err));
|
|
continue;
|
|
}
|
|
|
|
let result = sqlx::query_as::<_, DatabaseRow>(
|
|
r"
|
|
SELECT
|
|
CAST(`information_schema`.`SCHEMATA`.`SCHEMA_NAME` AS CHAR(64)) AS `database`,
|
|
GROUP_CONCAT(DISTINCT CAST(`information_schema`.`TABLES`.`TABLE_NAME` AS CHAR(64)) SEPARATOR ',') AS `tables`,
|
|
GROUP_CONCAT(DISTINCT CAST(`mysql`.`db`.`User` AS CHAR(64)) SEPARATOR ',') AS `users`,
|
|
MAX(`information_schema`.`SCHEMATA`.`DEFAULT_COLLATION_NAME`) AS `collation`,
|
|
MAX(`information_schema`.`SCHEMATA`.`DEFAULT_CHARACTER_SET_NAME`) AS `character_set`,
|
|
CAST(IFNULL(
|
|
SUM(`information_schema`.`TABLES`.`DATA_LENGTH` + `information_schema`.`TABLES`.`INDEX_LENGTH`),
|
|
0
|
|
) AS UNSIGNED INTEGER) AS `size_bytes`
|
|
FROM `information_schema`.`SCHEMATA`
|
|
LEFT OUTER JOIN `information_schema`.`TABLES`
|
|
ON `information_schema`.`SCHEMATA`.`SCHEMA_NAME` = `TABLES`.`TABLE_SCHEMA`
|
|
LEFT OUTER JOIN `mysql`.`db`
|
|
ON `information_schema`.`SCHEMATA`.`SCHEMA_NAME` = `mysql`.`db`.`DB`
|
|
WHERE `information_schema`.`SCHEMATA`.`SCHEMA_NAME` = ?
|
|
GROUP BY `information_schema`.`SCHEMATA`.`SCHEMA_NAME`
|
|
",
|
|
|
|
)
|
|
.bind(database_name.to_string())
|
|
.fetch_optional(&mut *connection)
|
|
.await
|
|
.map_err(|err| ListDatabasesError::MySqlError(err.to_string()))
|
|
.and_then(|database| {
|
|
database.map_or_else(|| Err(ListDatabasesError::DatabaseDoesNotExist), Ok)
|
|
});
|
|
|
|
if let Err(err) = &result {
|
|
tracing::error!("Failed to list database '{}': {:?}", &database_name, err);
|
|
}
|
|
|
|
// TODO: should we assert that the users are also owned by the unix_user from the request?
|
|
|
|
results.insert(database_name, result);
|
|
}
|
|
|
|
results
|
|
}
|
|
|
|
pub async fn list_all_databases_for_user(
|
|
unix_user: &UnixUser,
|
|
connection: &mut MySqlConnection,
|
|
_db_is_mariadb: bool,
|
|
group_denylist: &GroupDenylist,
|
|
) -> ListAllDatabasesResponse {
|
|
let result = sqlx::query_as::<_, DatabaseRow>(
|
|
r"
|
|
SELECT
|
|
CAST(`information_schema`.`SCHEMATA`.`SCHEMA_NAME` AS CHAR(64)) AS `database`,
|
|
GROUP_CONCAT(DISTINCT CAST(`information_schema`.`TABLES`.`TABLE_NAME` AS CHAR(64)) SEPARATOR ',') AS `tables`,
|
|
GROUP_CONCAT(DISTINCT CAST(`mysql`.`db`.`User` AS CHAR(64)) SEPARATOR ',') AS `users`,
|
|
MAX(`information_schema`.`SCHEMATA`.`DEFAULT_COLLATION_NAME`) AS `collation`,
|
|
MAX(`information_schema`.`SCHEMATA`.`DEFAULT_CHARACTER_SET_NAME`) AS `character_set`,
|
|
CAST(IFNULL(
|
|
SUM(`information_schema`.`TABLES`.`DATA_LENGTH` + `information_schema`.`TABLES`.`INDEX_LENGTH`),
|
|
0
|
|
) AS UNSIGNED INTEGER) AS `size_bytes`
|
|
FROM `information_schema`.`SCHEMATA`
|
|
LEFT OUTER JOIN `information_schema`.`TABLES`
|
|
ON `information_schema`.`SCHEMATA`.`SCHEMA_NAME` = `TABLES`.`TABLE_SCHEMA`
|
|
LEFT OUTER JOIN `mysql`.`db`
|
|
ON `information_schema`.`SCHEMATA`.`SCHEMA_NAME` = `mysql`.`db`.`DB`
|
|
WHERE `information_schema`.`SCHEMATA`.`SCHEMA_NAME` NOT IN ('information_schema', 'performance_schema', 'mysql', 'sys')
|
|
AND `information_schema`.`SCHEMATA`.`SCHEMA_NAME` REGEXP ?
|
|
GROUP BY `information_schema`.`SCHEMATA`.`SCHEMA_NAME`
|
|
",
|
|
)
|
|
.bind(create_user_group_matching_regex(unix_user, group_denylist))
|
|
.fetch_all(connection)
|
|
.await
|
|
.map_err(|err| ListAllDatabasesError::MySqlError(err.to_string()));
|
|
|
|
// TODO: should we assert that the users are also owned by the unix_user from the request?
|
|
|
|
if let Err(err) = &result {
|
|
tracing::error!(
|
|
"Failed to list databases for user '{}': {:?}",
|
|
unix_user.username,
|
|
err
|
|
);
|
|
}
|
|
|
|
result
|
|
}
|