sql_data_context.cc

/*
 * Copyright (c) 2015, 2025, Oracle and/or its affiliates.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2.0,
 * as published by the Free Software Foundation.
 *
 * This program is designed to work with certain software (including
 * but not limited to OpenSSL) that is licensed under separate terms,
 * as designated in a particular file or component or in included license
 * documentation. The authors of MySQL hereby grant you an additional
 * permission to link the program and your derivative works with the
 * separately licensed software that they have either included with
 * the program or referenced in the documentation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License, version 2.0, for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/

#include"plugin/x/src/sql_data_context.h"

#include<algorithm>
#include<cinttypes>
#include<sstream>
#include<string>

#include"mysql/components/my_service.h"
#include"mysql/components/services/mysql_admin_session.h"
#include"mysql/plugin.h"
#include"mysql/service_command.h"
#include"sql/debug_sync.h"

#include"plugin/x/src/helper/get_system_variable.h"
#include"plugin/x/src/module_mysqlx.h"
#include"plugin/x/src/mysql_variables.h"
#include"plugin/x/src/notices.h"
#include"plugin/x/src/query_string_builder.h"
#include"plugin/x/src/sql_user_require.h"
#include"plugin/x/src/xpl_error.h"
#include"plugin/x/src/xpl_log.h"
#include"plugin/x/src/xpl_resultset.h"

#include"my_systime.h"// my_sleep() NOLINT(build/include_subdir)

namespacexpl {

namespacedetails {

classAdmin_session_factory {
public:
 MYSQL_SESSION create(srv_session_error_cb error_cb, void *context) {
if (!m_admin_session.is_valid()) returnnullptr;

return m_admin_session->open(error_cb, context);
 }

private:
classRegistry_holder {
public:
Registry_holder() : m_registry{mysql_plugin_registry_acquire()} {}
~Registry_holder() { mysql_plugin_registry_release(m_registry); }

SERVICE_TYPE(registry) *const m_registry;
 };

 Registry_holder m_registry_holder;
 my_service<SERVICE_TYPE(mysql_admin_session)> m_admin_session{
"mysql_admin_session", m_registry_holder.m_registry};
};

} // namespace details

ngs::Error_code Sql_data_context::init(constint client_port,
const Connection_type type,
constbool is_admin) {
 ngs::Error_code error = init(is_admin);
if (error) return error;

if ((error = set_connection_type(type))) return error;

if (0 != srv_session_info_set_client_port(m_mysql_session, client_port))
returnngs::Error_code(ER_X_SESSION, "Could not set session client port");

returnngs::Error_code();
}

ngs::Error_code Sql_data_context::init(constbool is_admin) {
if (is_admin) {
 details::Admin_session_factory factory;
 m_mysql_session =
 factory.create(&Sql_data_context::default_completion_handler, this);

 } else {
 m_mysql_session =
srv_session_open(&Sql_data_context::default_completion_handler, this);
 }

log_debug(
"sqlsession init: %p [%i]", m_mysql_session,
 m_mysql_session ? srv_session_info_get_session_id(m_mysql_session) : -1);
if (!m_mysql_session) {
if (ER_SERVER_ISNT_AVAILABLE == m_last_sql_errno)
returnngs::Error_code(ER_SERVER_ISNT_AVAILABLE, "Server API not ready");

if (ER_CON_COUNT_ERROR == m_last_sql_errno)
returnngs::Error_code(m_last_sql_errno, m_last_sql_error);

if (ER_CON_COUNT_ERROR == m_last_sql_errno)
returnngs::Error_code(m_last_sql_errno, m_last_sql_error);

log_error(ER_XPLUGIN_FAILED_TO_OPEN_INTERNAL_SESSION);
returnngs::Error_code(ER_X_SESSION, "Could not open session");
 }

 m_pre_authenticate_event_fired = false;
 m_authentication_code = ngs::Error_code();

returnngs::Error_code();
}

voidSql_data_context::deinit() {
if (m_mysql_session) {
if (m_pre_authenticate_event_fired && !attach()) {
if (!m_authentication_code) {
// In case of successfull login, connect event has been generated
// earlier. Time for generating disconnect event.
modules::Module_mysqlx::get_instance_services()->m_audit_api->emit(
get_thd(), MYSQL_AUDIT_CONNECTION_DISCONNECT);
 } else {
modules::Module_mysqlx::get_instance_services()
 ->m_audit_api->emit_with_errorcode(get_thd(),
 MYSQL_AUDIT_CONNECTION_CONNECT,
 m_authentication_code.error);
 }
 }

 m_pre_authenticate_event_fired = false;
 m_authentication_code = ngs::Error_code();

srv_session_detach(m_mysql_session);

log_debug("sqlsession deinit: %p [%i]", m_mysql_session,
srv_session_info_get_session_id(m_mysql_session));

srv_session_close(m_mysql_session);
 m_mysql_session = nullptr;
 }

#ifdef HAVE_PSI_THREAD_INTERFACE
PSI_THREAD_CALL(delete_current_thread)();

 PSI_thread *psi =
PSI_THREAD_CALL(new_thread)(KEY_thread_x_worker, 0, nullptr, 0);
PSI_THREAD_CALL(set_thread_os_id)(psi);
PSI_THREAD_CALL(set_thread)(psi);
#endif/* HAVE_PSI_THREAD_INTERFACE */
} // namespace xpl

staticvoidkill_completion_handler(void *, unsignedint sql_errno,
constchar *err_msg) {
log_warning(ER_XPLUGIN_CLIENT_KILL_MSG, sql_errno, err_msg);
}

boolSql_data_context::kill() {
if (srv_session_server_is_available()) {
log_debug("sqlsession init (for kill): %p [%i]", m_mysql_session,
 m_mysql_session ? srv_session_info_get_session_id(m_mysql_session)
 : -1);
 MYSQL_SESSION session = srv_session_open(kill_completion_handler, nullptr);
bool ok = false;
if (session) {
 MYSQL_SECURITY_CONTEXT scontext;

if (thd_get_security_context(srv_session_info_get_thd(session),
 &scontext)) {
log_warning(ER_XPLUGIN_FAILED_TO_GET_SECURITY_CTX);
 } else {
constchar *user = MYSQL_SESSION_USER;
constchar *host = MYSQLXSYS_HOST;
if (security_context_lookup(scontext, user, host, nullptr, nullptr)) {
log_warning(ER_XPLUGIN_FAILED_TO_SWITCH_SECURITY_CTX, user);
 } else {
 COM_DATA data;
 Callback_command_delegate deleg;
 Query_string_builder qb;
 qb.put("KILL ").put(mysql_session_id());

memset(&data, 0, sizeof(data));
 data.com_query.query = qb.get().c_str();
 data.com_query.length = static_cast<unsignedint>(qb.get().length());

if (!command_service_run_command(
 session, COM_QUERY, &data, mysqld::get_default_charset(),
 deleg.callbacks(), deleg.representation(), &deleg)) {
if (!deleg.get_error())
 ok = true;
else
log_debug("Kill client: %i %s", deleg.get_error().error,
 deleg.get_error().message.c_str());
 }
 }
 }
srv_session_close(session);
 }
return ok;
 }
returnfalse;
}

ngs::Error_code Sql_data_context::set_connection_type(
const Connection_type type) {
 enum_vio_type vio_type = Connection_type_helper::convert_type(type);

if (NO_VIO_TYPE == vio_type)
returnngs::Error(ER_X_SESSION, "Connection type not known. type=%" PRIu32,
static_cast<uint32_t>(type));

if (0 != srv_session_info_set_connection_type(m_mysql_session, vio_type))
returnngs::Error_code(ER_X_SESSION,
"Could not set session connection type");

returnngs::Error_code();
}

boolSql_data_context::wait_api_ready(std::function<bool()> exiting) {
bool result = is_api_ready();

while (!result && !exiting()) {
my_sleep(250000); // wait for 0.25s

 result = is_api_ready();
 }

return result;
}

Sql_data_context::~Sql_data_context() {
if (m_mysql_session)
log_debug("sqlsession deinit~: %p [%i]", m_mysql_session,
srv_session_info_get_session_id(m_mysql_session));
if (m_mysql_session && srv_session_close(m_mysql_session))
log_warning(ER_XPLUGIN_FAILED_TO_CLOSE_SQL_SESSION);
}

voidSql_data_context::switch_to_local_user(const std::string &user) {
 ngs::Error_code error =
switch_to_user(user.c_str(), "localhost", nullptr, nullptr);
if (error) throw error;
}

ngs::Error_code Sql_data_context::authenticate_internal(
constchar *user, constchar *host, constchar *ip, constchar *db,
const std::string &passwd,
const iface::Authentication &account_verification,
bool allow_expired_passwords) {
 std::string authenticated_user_name = get_authenticated_user_name();
 std::string authenticated_user_host = get_authenticated_user_host();

 ngs::Error_code error =
switch_to_user(MYSQL_SESSION_USER, MYSQLXSYS_HOST, nullptr, nullptr);

if (error) {
constchar *session_user = MYSQL_SESSION_USER;
log_error(ER_XPLUGIN_FAILED_TO_SWITCH_CONTEXT, session_user);
return error;
 }

if (!is_acl_disabled()) {
 error = account_verification.authenticate_account(
 authenticated_user_name, authenticated_user_host, passwd);
 }

 ngs::Error_code switch_error = switch_to_user(user, host, ip, db);

if (error.error == ER_MUST_CHANGE_PASSWORD_LOGIN) {
 m_password_expired = true;

// password is expired, client doesn't support it and server wants us to
// disconnect these users
if (error.severity == ngs::Error_code::FATAL && !allow_expired_passwords)
return error;
 } else {
if (error) return error;
 }

 error = switch_error;

if (!error) {
if (db && *db) {
 COM_DATA data;

 data.com_init_db.db_name = db;
 data.com_init_db.length =
static_cast<unsignedlong>(strlen(db)); // NOLINT(runtime/int)

 Callback_command_delegate callback_delegate;
if (command_service_run_command(
 m_mysql_session, COM_INIT_DB, &data,
mysqld::get_default_charset(), callback_delegate.callbacks(),
 callback_delegate.representation(), &callback_delegate))
returnngs::Error_code(ER_NO_DB_ERROR, "Could not set database");
 error = callback_delegate.get_error();
 }

 std::string user_name = get_user_name();
 std::string host_or_ip = get_host_or_ip();

#ifdef HAVE_PSI_THREAD_INTERFACE
PSI_THREAD_CALL(set_thread_account)
 (user_name.c_str(), static_cast<int>(user_name.length()),
 host_or_ip.c_str(), static_cast<int>(host_or_ip.length()));
#endif// HAVE_PSI_THREAD_INTERFACE

return error;
 }

log_error(ER_XPLUGIN_FAILED_TO_SWITCH_CONTEXT, user);

return error;
}

ngs::Error_code Sql_data_context::authenticate(
constchar *user, constchar *host, constchar *ip, constchar *db,
const std::string &passwd,
const iface::Authentication &account_verification,
bool allow_expired_passwords) {
 m_password_expired = false;

 m_using_password = passwd.size() > 0;

 ngs::Error_code error = switch_to_user(user, host, ip, db);

if (error) error = ngs::SQLError_access_denied();

 m_authentication_code = error;

if (!m_pre_authenticate_event_fired) {
attach();
int audit_result =
modules::Module_mysqlx::get_instance_services()->m_audit_api->emit(
static_cast<void *>(get_thd()),
 MYSQL_AUDIT_CONNECTION_PRE_AUTHENTICATE);
detach();

if (audit_result != 0)
returnngs::Fatal(ER_AUDIT_API_ABORT, my_get_err_msg(ER_AUDIT_API_ABORT),
"MYSQL_AUDIT_CONNECTION_PRE_AUTHENTICATE",
 audit_result);

 m_pre_authenticate_event_fired = true;
 }

if (error) return error;

 error = authenticate_internal(user, host, ip, db, passwd,
 account_verification, allow_expired_passwords);

 m_authentication_code = error;

if (!error) {
attach();
int audit_result =
modules::Module_mysqlx::get_instance_services()->m_audit_api->emit(
static_cast<void *>(get_thd()), MYSQL_AUDIT_CONNECTION_CONNECT);
detach();

if (audit_result != 0)
 m_authentication_code = error =
ngs::Fatal(ER_AUDIT_API_ABORT, my_get_err_msg(ER_AUDIT_API_ABORT),
"MYSQL_AUDIT_CONNECTION_CONNECT", audit_result);
 }

return error;
}

template <typename Result_type>
boolget_security_context_value(MYSQL_THD thd, constchar *option,
 Result_type *result) {
 MYSQL_SECURITY_CONTEXT scontext;

if (thd_get_security_context(thd, &scontext)) returnfalse;

returnfalse == security_context_get_option(scontext, option, result);
}

boolSql_data_context::is_acl_disabled() {
 MYSQL_LEX_CSTRING value{"", 0};

if (get_security_context_value(get_thd(), "priv_user", &value)) {
return0 != value.length && nullptr != strstr(value.str, "skip-grants ");
 }

returnfalse;
}

boolSql_data_context::has_authenticated_user_a_super_priv() const {
 my_svc_bool value = 0;
if (get_security_context_value(get_thd(), "privilege_super", &value))
return value != 0;

returnfalse;
}

std::string Sql_data_context::get_user_name() const {
 MYSQL_LEX_CSTRING result{"", 0};

if (get_security_context_value(get_thd(), "user", &result)) return result.str;

return"";
}

std::string Sql_data_context::get_host_or_ip() const {
 MYSQL_LEX_CSTRING result{"", 0};

if (get_security_context_value(get_thd(), "host_or_ip", &result))
return result.str;

return"";
}

std::string Sql_data_context::get_authenticated_user_name() const {
 MYSQL_LEX_CSTRING result{"", 0};

if (get_security_context_value(get_thd(), "priv_user", &result))
return result.str;

return"";
}

std::string Sql_data_context::get_authenticated_user_host() const {
 MYSQL_LEX_CSTRING result{"", 0};

if (get_security_context_value(get_thd(), "priv_host", &result))
return result.str;

return"";
}

ngs::Error_code Sql_data_context::switch_to_user(constchar *username,
constchar *hostname,
constchar *address,
constchar *db) {
 MYSQL_SECURITY_CONTEXT scontext;

if (thd_get_security_context(get_thd(), &scontext))
returnngs::Fatal(ER_X_SERVICE_ERROR,
"Error getting security context for session");

attach();
CONDITIONAL_SYNC_POINT("xpl_switch_to_user_enter");

// security_context_lookup - doesn't make a copy of username, hostname, addres
// or db thus we need to make a copy of them
// and pass our pointers to security_context_lookup.
// If we change pass different pointers than in last
// invocation of `switch_to_user`, then it might
// cause an invalid memory access while listing all
// sessions. Thus we must use memory regions that
// can hold values of maximum length.
memset(m_username, 0, sizeof(m_username));
memset(m_hostname, 0, sizeof(m_hostname));
memset(m_address, 0, sizeof(m_address));
memset(m_db, 0, sizeof(m_db));

if (username) {
strncpy(m_username, username, sizeof(m_username) - 1);
 }

if (hostname) {
strncpy(m_hostname, hostname, sizeof(m_hostname) - 1);
 }

if (address) {
strncpy(m_address, address, sizeof(m_address) - 1);
 }

if (db) {
strncpy(m_db, db, sizeof(m_db) - 1);
 }

CONDITIONAL_SYNC_POINT("xpl_switch_to_user_execute");
detach();

log_debug("Switching security context to user %s@%s [%s]", username, hostname,
 address);

if (security_context_lookup(scontext, m_username, m_hostname, m_address,
 m_db)) {
log_debug("Unable to switch security context to user %s@%s [%s]", username,
 hostname, address);
returnngs::Fatal(ER_X_SERVICE_ERROR, "Unable to switch context to user %s",
 username);
 }

returnngs::Success();
}

ngs::Error_code Sql_data_context::execute_kill_sql_session(
uint64_t mysql_session_id) {
 Query_string_builder qb;
 qb.put("KILL ").put(mysql_session_id);
 Empty_resultset rset;
returnexecute(qb.get().data(), qb.get().length(), &rset);
}

voidSql_data_context::default_completion_handler(void *ctx,
unsignedint sql_errno,
constchar *err_msg) {
 Sql_data_context *self = reinterpret_cast<Sql_data_context *>(ctx);
 self->m_last_sql_errno = sql_errno;
 self->m_last_sql_error = err_msg ? err_msg : "";
}

boolSql_data_context::is_killed() const {
constautokill = thd_killed(get_thd());
DBUG_LOG("debug", "is_killed:" << kill);

if (0 == kill) returnfalse;

return ER_QUERY_INTERRUPTED != kill;
}

boolSql_data_context::is_api_ready() {
return0 != srv_session_server_is_available();
}

uint64_tSql_data_context::mysql_session_id() const {
returnsrv_session_info_get_session_id(m_mysql_session);
}

MYSQL_THD Sql_data_context::get_thd() const {
returnsrv_session_info_get_thd(m_mysql_session);
}

ngs::Error_code Sql_data_context::execute(constchar *sql, std::size_t sql_len,
 iface::Resultset *rset) {
constauto error = execute_sql(sql, sql_len, rset);
if (error) return error;

if (m_password_expired) {
// if a SQL command succeeded while password is expired, it means the user
// probably changed the password
// we run a command to check just in case... (some commands are still
// allowed in expired password mode)
 Empty_resultset rs;
staticconst std::string cmd("select 1");
if (!execute_sql(cmd.c_str(), cmd.length(), &rs))
 m_password_expired = false;
 }
returnngs::Success();
}

ngs::Error_code Sql_data_context::execute_sql(constchar *sql,
 std::size_t sql_len,
 iface::Resultset *rset) {
 COM_DATA data;
memset(&data, 0, sizeof(data));
 data.com_query.query = sql;
 data.com_query.length = static_cast<unsignedint>(sql_len);
returnexecute_server_command(COM_QUERY, data, rset);
}

ngs::Error_code Sql_data_context::fetch_cursor(constuint32_t statement_id,
constuint32_t row_count,
 iface::Resultset *rset) {
 COM_DATA data;
 data.com_stmt_fetch.stmt_id = statement_id;
 data.com_stmt_fetch.num_rows = row_count;
returnexecute_server_command(COM_STMT_FETCH, data, rset);
}

ngs::Error_code Sql_data_context::attach() {
 THD *previous_thd = nullptr;

if (nullptr == m_mysql_session ||
srv_session_attach(m_mysql_session, &previous_thd)) {
returnngs::Error_code(ER_X_SERVICE_ERROR, "Internal error attaching");
 }

return {};
}

ngs::Error_code Sql_data_context::detach() {
if (nullptr == m_mysql_session || srv_session_detach(m_mysql_session)) {
returnngs::Error_code(ER_X_SERVICE_ERROR, "Internal error when detaching");
 }

return {};
}

ngs::Error_code Sql_data_context::prepare_prep_stmt(constchar *sql,
 std::size_t sql_len,
 iface::Resultset *rset) {
 COM_DATA data;
 data.com_stmt_prepare = {sql, static_cast<unsigned>(sql_len)};
returnexecute_server_command(COM_STMT_PREPARE, data, rset);
}

ngs::Error_code Sql_data_context::deallocate_prep_stmt(constuint32_t stmt_id,
 iface::Resultset *rset) {
 COM_DATA data;
 data.com_stmt_close = {static_cast<unsigned>(stmt_id)};
returnexecute_server_command(COM_STMT_CLOSE, data, rset);
}

ngs::Error_code Sql_data_context::execute_prep_stmt(
constuint32_t stmt_id, constbool has_cursor, const PS_PARAM *parameters,
const std::size_t parameters_count, iface::Resultset *rset) {
 COM_DATA cmd;
 cmd.com_stmt_execute = {
static_cast<unsignedlong>(stmt_id), // NOLINT(runtime/int)
static_cast<unsignedlong>(has_cursor), // NOLINT(runtime/int)
const_cast<PS_PARAM *>(parameters),
static_cast<unsignedlong>(parameters_count), // NOLINT(runtime/int)
static_cast<unsignedchar>(true)}; // NOLINT(runtime/int)

returnexecute_server_command(COM_STMT_EXECUTE, cmd, rset);
}

ngs::Error_code Sql_data_context::reset() {
 COM_DATA data{};
 Empty_resultset rset;
returnexecute_server_command(COM_RESET_CONNECTION, data, &rset);
}

ngs::Error_code Sql_data_context::execute_server_command(
const enum_server_command cmd, const COM_DATA &cmd_data,
 iface::Resultset *rset) {
 ngs::Command_delegate &deleg = rset->get_callbacks();
 deleg.reset();
constauto fail = command_service_run_command(
 m_mysql_session, cmd, &cmd_data, mysqld::get_default_charset(),
 deleg.callbacks(), deleg.representation(), &deleg);

 ngs::Error_code error = deleg.get_error();

if (fail) {
if (error) {
log_debug("Error running server command: (%i:%s)", error.error,
 error.message.c_str());
return error;
 }
returnngs::Error(ER_X_SERVICE_ERROR, "Internal error executing command");
 }

if (!error) returnngs::Success();

log_debug("Error running server command: (%i:%s; killed: %i:%i)", error.error,
 error.message.c_str(), static_cast<int>(is_killed()),
static_cast<int>(deleg.killed()));

returnis_killed() ? ngs::Fatal(error) : error;
}

boolSql_data_context::is_sql_mode_set(const std::string &mode) {
 std::string modes;
 get_system_variable<std::string>(this, "sql_mode", &modes);
 std::istringstream str(modes);
 std::string m;
while (std::getline(str, m, ','))
if (m == mode) returntrue;
returnfalse;
}

} // namespace xpl