diff options
Diffstat (limited to 'sql/sql_connect.cc')
| -rw-r--r-- | sql/sql_connect.cc | 1582 |
1 files changed, 1582 insertions, 0 deletions
diff --git a/sql/sql_connect.cc b/sql/sql_connect.cc new file mode 100644 index 00000000..999830ff --- /dev/null +++ b/sql/sql_connect.cc @@ -0,0 +1,1582 @@ +/* + Copyright (c) 2007, 2013, Oracle and/or its affiliates. + Copyright (c) 2008, 2020, MariaDB + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + 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 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-1335 USA +*/ + +/* + Functions to autenticate and handle reqests for a connection +*/ + +#include "mariadb.h" +#include "mysqld.h" +#include "sql_priv.h" +#ifndef __WIN__ +#include <netdb.h> // getservbyname, servent +#endif +#include "sql_audit.h" +#include "sql_connect.h" +#include "thread_cache.h" +#include "probes_mysql.h" +#include "sql_parse.h" // sql_command_flags, + // execute_init_command, + // do_command +#include "sql_db.h" // mysql_change_db +#include "hostname.h" // inc_host_errors, ip_to_hostname, + // reset_host_errors +#include "sql_callback.h" + +#ifdef WITH_WSREP +#include "wsrep_trans_observer.h" /* wsrep open/close */ +#include "wsrep_mysqld.h" +#endif /* WITH_WSREP */ +#include "proxy_protocol.h" +#include <ssl_compat.h> + +HASH global_user_stats, global_client_stats, global_table_stats; +HASH global_index_stats; +/* Protects the above global stats */ +extern mysql_mutex_t LOCK_global_user_client_stats; +extern mysql_mutex_t LOCK_global_table_stats; +extern mysql_mutex_t LOCK_global_index_stats; +extern vio_keepalive_opts opt_vio_keepalive; + +/* + Get structure for logging connection data for the current user +*/ + +#ifndef NO_EMBEDDED_ACCESS_CHECKS +static HASH hash_user_connections; + +int get_or_create_user_conn(THD *thd, const char *user, + const char *host, + const USER_RESOURCES *mqh) +{ + int return_val= 0; + size_t temp_len, user_len; + char temp_user[USER_HOST_BUFF_SIZE]; + struct user_conn *uc; + + DBUG_ASSERT(user != 0); + DBUG_ASSERT(host != 0); + DBUG_ASSERT(thd->user_connect == 0); + + user_len= strlen(user); + temp_len= (strmov(strmov(temp_user, user)+1, host) - temp_user)+1; + mysql_mutex_lock(&LOCK_user_conn); + if (!(uc = (struct user_conn *) my_hash_search(&hash_user_connections, + (uchar*) temp_user, temp_len))) + { + /* First connection for user; Create a user connection object */ + if (!(uc= ((struct user_conn*) + my_malloc(key_memory_user_conn, + sizeof(struct user_conn) + temp_len+1, MYF(MY_WME))))) + { + /* MY_WME ensures an error is set in THD. */ + return_val= 1; + goto end; + } + uc->user=(char*) (uc+1); + memcpy(uc->user,temp_user,temp_len+1); + uc->host= uc->user + user_len + 1; + uc->len= (uint)temp_len; + uc->connections= uc->questions= uc->updates= uc->conn_per_hour= 0; + uc->reset_utime= thd->thr_create_utime; + if (my_hash_insert(&hash_user_connections, (uchar*) uc)) + { + /* The only possible error is out of memory, MY_WME sets an error. */ + my_free(uc); + return_val= 1; + goto end; + } + } + uc->user_resources= *mqh; + thd->user_connect=uc; + uc->connections++; +end: + mysql_mutex_unlock(&LOCK_user_conn); + return return_val; +} + + +/* + check if user has already too many connections + + SYNOPSIS + check_for_max_user_connections() + thd Thread handle + uc User connect object + + NOTES + If check fails, we decrease user connection count, which means one + shouldn't call decrease_user_connections() after this function. + + RETURN + 0 ok + 1 error +*/ + +int check_for_max_user_connections(THD *thd, USER_CONN *uc) +{ + int error= 1; + Host_errors errors; + DBUG_ENTER("check_for_max_user_connections"); + + mysql_mutex_lock(&LOCK_user_conn); + + /* Root is not affected by the value of max_user_connections */ + if (global_system_variables.max_user_connections && + !uc->user_resources.user_conn && + global_system_variables.max_user_connections < uc->connections && + !(thd->security_ctx->master_access & PRIV_IGNORE_MAX_USER_CONNECTIONS)) + { + my_error(ER_TOO_MANY_USER_CONNECTIONS, MYF(0), uc->user); + error=1; + errors.m_max_user_connection= 1; + goto end; + } + time_out_user_resource_limits(thd, uc); + if (uc->user_resources.user_conn && + uc->user_resources.user_conn < uc->connections) + { + my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, + "max_user_connections", + (long) uc->user_resources.user_conn); + error= 1; + errors.m_max_user_connection= 1; + goto end; + } + if (uc->user_resources.conn_per_hour && + uc->user_resources.conn_per_hour <= uc->conn_per_hour) + { + my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, + "max_connections_per_hour", + (long) uc->user_resources.conn_per_hour); + error=1; + errors.m_max_user_connection_per_hour= 1; + goto end; + } + uc->conn_per_hour++; + error= 0; + +end: + if (unlikely(error)) + { + uc->connections--; // no need for decrease_user_connections() here + /* + The thread may returned back to the pool and assigned to a user + that doesn't have a limit. Ensure the user is not using resources + of someone else. + */ + thd->user_connect= NULL; + } + mysql_mutex_unlock(&LOCK_user_conn); + if (unlikely(error)) + { + inc_host_errors(thd->main_security_ctx.ip, &errors); + } + DBUG_RETURN(error); +} + + +/* + Decrease user connection count + + SYNOPSIS + decrease_user_connections() + uc User connection object + + NOTES + If there is a n user connection object for a connection + (which only happens if 'max_user_connections' is defined or + if someone has created a resource grant for a user), then + the connection count is always incremented on connect. + + The user connect object is not freed if some users has + 'max connections per hour' defined as we need to be able to hold + count over the lifetime of the connection. +*/ + +void decrease_user_connections(USER_CONN *uc) +{ + DBUG_ENTER("decrease_user_connections"); + mysql_mutex_lock(&LOCK_user_conn); + DBUG_ASSERT(uc->connections); + if (!--uc->connections && !mqh_used) + { + /* Last connection for user; Delete it */ + (void) my_hash_delete(&hash_user_connections,(uchar*) uc); + } + mysql_mutex_unlock(&LOCK_user_conn); + DBUG_VOID_RETURN; +} + + +/* + Reset per-hour user resource limits when it has been more than + an hour since they were last checked + + SYNOPSIS: + time_out_user_resource_limits() + thd Thread handler + uc User connection details + + NOTE: + This assumes that the LOCK_user_conn mutex has been acquired, so it is + safe to test and modify members of the USER_CONN structure. +*/ + +void time_out_user_resource_limits(THD *thd, USER_CONN *uc) +{ + ulonglong check_time= thd->start_utime; + DBUG_ENTER("time_out_user_resource_limits"); + + /* If more than a hour since last check, reset resource checking */ + if (check_time - uc->reset_utime >= 3600000000ULL) + { + uc->questions=0; + uc->updates=0; + uc->conn_per_hour=0; + uc->reset_utime= check_time; + } + + DBUG_VOID_RETURN; +} + +/* + Check if maximum queries per hour limit has been reached + returns 0 if OK. +*/ + +bool check_mqh(THD *thd, uint check_command) +{ + bool error= 0; + USER_CONN *uc=thd->user_connect; + DBUG_ENTER("check_mqh"); + DBUG_ASSERT(uc != 0); + + mysql_mutex_lock(&LOCK_user_conn); + + time_out_user_resource_limits(thd, uc); + + /* Check that we have not done too many questions / hour */ + if (uc->user_resources.questions && + uc->questions++ >= uc->user_resources.questions) + { + my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, "max_queries_per_hour", + (long) uc->user_resources.questions); + error=1; + goto end; + } + if (check_command < (uint) SQLCOM_END) + { + /* Check that we have not done too many updates / hour */ + if (uc->user_resources.updates && + (sql_command_flags[check_command] & CF_CHANGES_DATA) && + uc->updates++ >= uc->user_resources.updates) + { + my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, "max_updates_per_hour", + (long) uc->user_resources.updates); + error=1; + goto end; + } + } +end: + mysql_mutex_unlock(&LOCK_user_conn); + DBUG_RETURN(error); +} + +#endif /* NO_EMBEDDED_ACCESS_CHECKS */ + +/* + Check for maximum allowable user connections, if the mysqld server is + started with corresponding variable that is greater then 0. +*/ + +extern "C" uchar *get_key_conn(user_conn *buff, size_t *length, + my_bool not_used __attribute__((unused))) +{ + *length= buff->len; + return (uchar*) buff->user; +} + + +extern "C" void free_user(struct user_conn *uc) +{ + my_free(uc); +} + + +void init_max_user_conn(void) +{ +#ifndef NO_EMBEDDED_ACCESS_CHECKS + my_hash_init(key_memory_user_conn, &hash_user_connections, + system_charset_info, max_connections, 0, 0, (my_hash_get_key) + get_key_conn, (my_hash_free_key) free_user, 0); +#endif +} + + +void free_max_user_conn(void) +{ +#ifndef NO_EMBEDDED_ACCESS_CHECKS + my_hash_free(&hash_user_connections); +#endif /* NO_EMBEDDED_ACCESS_CHECKS */ +} + + +void reset_mqh(LEX_USER *lu, bool get_them= 0) +{ +#ifndef NO_EMBEDDED_ACCESS_CHECKS + mysql_mutex_lock(&LOCK_user_conn); + if (lu) // for GRANT + { + USER_CONN *uc; + size_t temp_len=lu->user.length+lu->host.length+2; + char temp_user[USER_HOST_BUFF_SIZE]; + + memcpy(temp_user,lu->user.str,lu->user.length); + memcpy(temp_user+lu->user.length+1,lu->host.str,lu->host.length); + temp_user[lu->user.length]='\0'; temp_user[temp_len-1]=0; + if ((uc = (struct user_conn *) my_hash_search(&hash_user_connections, + (uchar*) temp_user, + temp_len))) + { + uc->questions=0; + get_mqh(temp_user,&temp_user[lu->user.length+1],uc); + uc->updates=0; + uc->conn_per_hour=0; + } + } + else + { + /* for FLUSH PRIVILEGES and FLUSH USER_RESOURCES */ + for (uint idx=0;idx < hash_user_connections.records; idx++) + { + USER_CONN *uc=(struct user_conn *) + my_hash_element(&hash_user_connections, idx); + if (get_them) + get_mqh(uc->user,uc->host,uc); + uc->questions=0; + uc->updates=0; + uc->conn_per_hour=0; + } + } + mysql_mutex_unlock(&LOCK_user_conn); +#endif /* NO_EMBEDDED_ACCESS_CHECKS */ +} + +/***************************************************************************** + Handle users statistics +*****************************************************************************/ + +/* 'mysql_system_user' is used for when the user is not defined for a THD. */ +static const char mysql_system_user[]= "#mysql_system#"; + +// Returns 'user' if it's not NULL. Returns 'mysql_system_user' otherwise. +static const char * get_valid_user_string(const char* user) +{ + return user ? user : mysql_system_user; +} + +/* + Returns string as 'IP' for the client-side of the connection represented by + 'client'. Does not allocate memory. May return "". +*/ + +static const char *get_client_host(THD *client) +{ + return client->security_ctx->host_or_ip[0] ? + client->security_ctx->host_or_ip : + client->security_ctx->host ? client->security_ctx->host : ""; +} + +extern "C" uchar *get_key_user_stats(USER_STATS *user_stats, size_t *length, + my_bool not_used __attribute__((unused))) +{ + *length= user_stats->user_name_length; + return (uchar*) user_stats->user; +} + +void free_user_stats(USER_STATS* user_stats) +{ + my_free(user_stats); +} + +void init_user_stats(USER_STATS *user_stats, + const char *user, + size_t user_length, + const char *priv_user, + uint total_connections, + uint total_ssl_connections, + uint concurrent_connections, + time_t connected_time, + double busy_time, + double cpu_time, + ulonglong bytes_received, + ulonglong bytes_sent, + ulonglong binlog_bytes_written, + ha_rows rows_sent, + ha_rows rows_read, + ha_rows rows_inserted, + ha_rows rows_deleted, + ha_rows rows_updated, + ulonglong select_commands, + ulonglong update_commands, + ulonglong other_commands, + ulonglong commit_trans, + ulonglong rollback_trans, + ulonglong denied_connections, + ulonglong lost_connections, + ulonglong max_statement_time_exceeded, + ulonglong access_denied_errors, + ulonglong empty_queries) +{ + DBUG_ENTER("init_user_stats"); + DBUG_PRINT("enter", ("user: %s priv_user: %s", user, priv_user)); + + user_length= MY_MIN(user_length, sizeof(user_stats->user)-1); + memcpy(user_stats->user, user, user_length); + user_stats->user[user_length]= 0; + user_stats->user_name_length= (uint)user_length; + strmake_buf(user_stats->priv_user, priv_user); + + user_stats->total_connections= total_connections; + user_stats->total_ssl_connections= total_ssl_connections; + user_stats->concurrent_connections= concurrent_connections; + user_stats->connected_time= connected_time; + user_stats->busy_time= busy_time; + user_stats->cpu_time= cpu_time; + user_stats->bytes_received= bytes_received; + user_stats->bytes_sent= bytes_sent; + user_stats->binlog_bytes_written= binlog_bytes_written; + user_stats->rows_sent= rows_sent; + user_stats->rows_read= rows_read; + user_stats->rows_inserted= rows_inserted; + user_stats->rows_deleted= rows_deleted; + user_stats->rows_updated= rows_updated; + user_stats->select_commands= select_commands; + user_stats->update_commands= update_commands; + user_stats->other_commands= other_commands; + user_stats->commit_trans= commit_trans; + user_stats->rollback_trans= rollback_trans; + user_stats->denied_connections= denied_connections; + user_stats->lost_connections= lost_connections; + user_stats->max_statement_time_exceeded= max_statement_time_exceeded; + user_stats->access_denied_errors= access_denied_errors; + user_stats->empty_queries= empty_queries; + DBUG_VOID_RETURN; +} + + +void init_global_user_stats(void) +{ + my_hash_init(PSI_INSTRUMENT_ME, &global_user_stats, system_charset_info, max_connections, + 0, 0, (my_hash_get_key) get_key_user_stats, + (my_hash_free_key) free_user_stats, 0); +} + +void init_global_client_stats(void) +{ + my_hash_init(PSI_INSTRUMENT_ME, &global_client_stats, system_charset_info, max_connections, + 0, 0, (my_hash_get_key) get_key_user_stats, + (my_hash_free_key) free_user_stats, 0); +} + +extern "C" uchar *get_key_table_stats(TABLE_STATS *table_stats, size_t *length, + my_bool not_used __attribute__((unused))) +{ + *length= table_stats->table_name_length; + return (uchar*) table_stats->table; +} + +extern "C" void free_table_stats(TABLE_STATS* table_stats) +{ + my_free(table_stats); +} + +void init_global_table_stats(void) +{ + my_hash_init(PSI_INSTRUMENT_ME, &global_table_stats, system_charset_info, + max_connections, 0, 0, (my_hash_get_key) get_key_table_stats, + (my_hash_free_key) free_table_stats, 0); +} + +extern "C" uchar *get_key_index_stats(INDEX_STATS *index_stats, size_t *length, + my_bool not_used __attribute__((unused))) +{ + *length= index_stats->index_name_length; + return (uchar*) index_stats->index; +} + +extern "C" void free_index_stats(INDEX_STATS* index_stats) +{ + my_free(index_stats); +} + +void init_global_index_stats(void) +{ + my_hash_init(PSI_INSTRUMENT_ME, &global_index_stats, system_charset_info, + max_connections, 0, 0, (my_hash_get_key) get_key_index_stats, + (my_hash_free_key) free_index_stats, 0); +} + + +void free_global_user_stats(void) +{ + my_hash_free(&global_user_stats); +} + +void free_global_table_stats(void) +{ + my_hash_free(&global_table_stats); +} + +void free_global_index_stats(void) +{ + my_hash_free(&global_index_stats); +} + +void free_global_client_stats(void) +{ + my_hash_free(&global_client_stats); +} + +/* + Increments the global stats connection count for an entry from + global_client_stats or global_user_stats. Returns 0 on success + and 1 on error. +*/ + +static bool increment_count_by_name(const char *name, size_t name_length, + const char *role_name, + HASH *users_or_clients, THD *thd) +{ + USER_STATS *user_stats; + + if (!(user_stats= (USER_STATS*) my_hash_search(users_or_clients, (uchar*) name, + name_length))) + { + /* First connection for this user or client */ + if (!(user_stats= ((USER_STATS*) + my_malloc(PSI_INSTRUMENT_ME, sizeof(USER_STATS), + MYF(MY_WME | MY_ZEROFILL))))) + return TRUE; // Out of memory + + init_user_stats(user_stats, name, name_length, role_name, + 0, 0, 0, // connections + 0, 0, 0, // time + 0, 0, 0, // bytes sent, received and written + 0, 0, // rows sent and read + 0, 0, 0, // rows inserted, deleted and updated + 0, 0, 0, // select, update and other commands + 0, 0, // commit and rollback trans + thd->status_var.access_denied_errors, + 0, // lost connections + 0, // max query timeouts + 0, // access denied errors + 0); // empty queries + + if (my_hash_insert(users_or_clients, (uchar*)user_stats)) + { + my_free(user_stats); + return TRUE; // Out of memory + } + } + user_stats->total_connections++; + if (thd->net.vio && thd->net.vio->type == VIO_TYPE_SSL) + user_stats->total_ssl_connections++; + return FALSE; +} + + +/* + Increments the global user and client stats connection count. + + @param use_lock if true, LOCK_global_user_client_stats will be locked + + @retval 0 ok + @retval 1 error. +*/ + +#ifndef EMBEDDED_LIBRARY +static bool increment_connection_count(THD* thd, bool use_lock) +{ + const char *user_string= get_valid_user_string(thd->main_security_ctx.user); + const char *client_string= get_client_host(thd); + bool return_value= FALSE; + + if (!thd->userstat_running) + return FALSE; + + if (use_lock) + mysql_mutex_lock(&LOCK_global_user_client_stats); + + if (increment_count_by_name(user_string, strlen(user_string), user_string, + &global_user_stats, thd)) + { + return_value= TRUE; + goto end; + } + if (increment_count_by_name(client_string, strlen(client_string), + user_string, &global_client_stats, thd)) + { + return_value= TRUE; + goto end; + } + +end: + if (use_lock) + mysql_mutex_unlock(&LOCK_global_user_client_stats); + return return_value; +} +#endif + +/* + Used to update the global user and client stats +*/ + +static void update_global_user_stats_with_user(THD *thd, + USER_STATS *user_stats, + time_t now) +{ + DBUG_ASSERT(thd->userstat_running); + + user_stats->connected_time+= now - thd->last_global_update_time; + user_stats->busy_time+= (thd->status_var.busy_time - + thd->org_status_var.busy_time); + user_stats->cpu_time+= (thd->status_var.cpu_time - + thd->org_status_var.cpu_time); + /* + This is handle specially as bytes_received is incremented BEFORE + org_status_var is copied. + */ + user_stats->bytes_received+= (thd->org_status_var.bytes_received- + thd->start_bytes_received); + user_stats->bytes_sent+= (thd->status_var.bytes_sent - + thd->org_status_var.bytes_sent); + user_stats->binlog_bytes_written+= + (thd->status_var.binlog_bytes_written - + thd->org_status_var.binlog_bytes_written); + /* We are not counting rows in internal temporary tables here ! */ + user_stats->rows_read+= (thd->status_var.rows_read - + thd->org_status_var.rows_read); + user_stats->rows_sent+= (thd->status_var.rows_sent - + thd->org_status_var.rows_sent); + user_stats->rows_inserted+= (thd->status_var.ha_write_count - + thd->org_status_var.ha_write_count); + user_stats->rows_deleted+= (thd->status_var.ha_delete_count - + thd->org_status_var.ha_delete_count); + user_stats->rows_updated+= (thd->status_var.ha_update_count - + thd->org_status_var.ha_update_count); + user_stats->select_commands+= thd->select_commands; + user_stats->update_commands+= thd->update_commands; + user_stats->other_commands+= thd->other_commands; + user_stats->commit_trans+= (thd->status_var.ha_commit_count - + thd->org_status_var.ha_commit_count); + user_stats->rollback_trans+= (thd->status_var.ha_rollback_count + + thd->status_var.ha_savepoint_rollback_count - + thd->org_status_var.ha_rollback_count - + thd->org_status_var. + ha_savepoint_rollback_count); + user_stats->access_denied_errors+= + (thd->status_var.access_denied_errors - + thd->org_status_var.access_denied_errors); + user_stats->empty_queries+= (thd->status_var.empty_queries - + thd->org_status_var.empty_queries); + + /* The following can only contain 0 or 1 and then connection ends */ + user_stats->denied_connections+= thd->status_var.access_denied_errors; + user_stats->lost_connections+= thd->status_var.lost_connections; + user_stats->max_statement_time_exceeded+= thd->status_var.max_statement_time_exceeded; +} + + +/* Updates the global stats of a user or client */ +void update_global_user_stats(THD *thd, bool create_user, time_t now) +{ + const char *user_string, *client_string; + USER_STATS *user_stats; + size_t user_string_length, client_string_length; + DBUG_ASSERT(thd->userstat_running); + + user_string= get_valid_user_string(thd->main_security_ctx.user); + user_string_length= strlen(user_string); + client_string= get_client_host(thd); + client_string_length= strlen(client_string); + + mysql_mutex_lock(&LOCK_global_user_client_stats); + + // Update by user name + if ((user_stats= (USER_STATS*) my_hash_search(&global_user_stats, + (uchar*) user_string, + user_string_length))) + { + /* Found user. */ + update_global_user_stats_with_user(thd, user_stats, now); + } + else + { + /* Create the entry */ + if (create_user) + { + increment_count_by_name(user_string, user_string_length, user_string, + &global_user_stats, thd); + } + } + + /* Update by client IP */ + if ((user_stats= (USER_STATS*)my_hash_search(&global_client_stats, + (uchar*) client_string, + client_string_length))) + { + // Found by client IP + update_global_user_stats_with_user(thd, user_stats, now); + } + else + { + // Create the entry + if (create_user) + { + increment_count_by_name(client_string, client_string_length, + user_string, &global_client_stats, thd); + } + } + /* Reset variables only used for counting */ + thd->select_commands= thd->update_commands= thd->other_commands= 0; + thd->last_global_update_time= now; + + mysql_mutex_unlock(&LOCK_global_user_client_stats); +} + + +/** + Set thread character set variables from the given ID + + @param thd thread handle + @param cs_number character set and collation ID + + @retval 0 OK; character_set_client, collation_connection and + character_set_results are set to the new value, + or to the default global values. + + @retval 1 error, e.g. the given ID is not supported by parser. + Corresponding SQL error is sent. +*/ + +bool thd_init_client_charset(THD *thd, uint cs_number) +{ + CHARSET_INFO *cs; + /* + Use server character set and collation if + - opt_character_set_client_handshake is not set + - client has not specified a character set + - client character set doesn't exists in server + */ + if (!opt_character_set_client_handshake || + !(cs= get_charset(cs_number, MYF(0)))) + { + thd->update_charset(global_system_variables.character_set_client, + global_system_variables.collation_connection, + global_system_variables.character_set_results); + } + else + { + if (!is_supported_parser_charset(cs)) + { + /* Disallow non-supported parser character sets: UCS2, UTF16, UTF32 */ + my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "character_set_client", + cs->csname); + return true; + } + thd->org_charset= cs; + thd->update_charset(cs,cs,cs); + } + return false; +} + + +/* + Initialize connection threads +*/ + +#ifndef EMBEDDED_LIBRARY +bool init_new_connection_handler_thread() +{ + pthread_detach_this_thread(); + if (my_thread_init()) + { + statistic_increment(aborted_connects,&LOCK_status); + statistic_increment(connection_errors_internal, &LOCK_status); + return 1; + } + DBUG_EXECUTE_IF("simulate_failed_connection_1", return(1); ); + return 0; +} + +/** + Set client address during authentication. + + Initializes THD::main_security_ctx and THD::peer_port. + Optionally does ip to hostname translation. + + @param thd current THD handle + @param addr peer address (can be NULL, if 'ip' is set) + @param ip peer address as string (can be NULL if 'addr' is set) + @param port peer port + @param check_proxy_networks if true, and host is in + 'proxy_protocol_networks' list, skip + "host not privileged" check + @param[out] host_errors - number of connect + errors for this host + + @retval 0 ok, 1 error +*/ +int thd_set_peer_addr(THD *thd, + sockaddr_storage *addr, + const char *ip, + uint port, + bool check_proxy_networks, + uint *host_errors) +{ + *host_errors= 0; + + thd->peer_port= port; + + char ip_string[128]; + if (!ip) + { + void *addr_data; + if (addr->ss_family == AF_UNIX) + { + /* local connection */ + my_free((void *)thd->main_security_ctx.ip); + thd->main_security_ctx.host_or_ip= thd->main_security_ctx.host = my_localhost; + thd->main_security_ctx.ip= 0; + return 0; + } + else if (addr->ss_family == AF_INET) + addr_data= &((struct sockaddr_in *)addr)->sin_addr; + else + addr_data= &((struct sockaddr_in6 *)addr)->sin6_addr; + if (!inet_ntop(addr->ss_family,addr_data, ip_string, sizeof(ip_string))) + { + DBUG_ASSERT(0); + return 1; + } + ip= ip_string; + } + + my_free((void *)thd->main_security_ctx.ip); + if (!(thd->main_security_ctx.ip = my_strdup(PSI_INSTRUMENT_ME, ip, MYF(MY_WME)))) + { + /* + No error accounting per IP in host_cache, + this is treated as a global server OOM error. + TODO: remove the need for my_strdup. + */ + statistic_increment(aborted_connects, &LOCK_status); + statistic_increment(connection_errors_internal, &LOCK_status); + return 1; /* The error is set by my_strdup(). */ + } + thd->main_security_ctx.host_or_ip = thd->main_security_ctx.ip; + if (!(specialflag & SPECIAL_NO_RESOLVE)) + { + int rc; + + rc = ip_to_hostname(addr, + thd->main_security_ctx.ip, + &thd->main_security_ctx.host, + host_errors); + + /* Cut very long hostnames to avoid possible overflows */ + if (thd->main_security_ctx.host) + { + if (thd->main_security_ctx.host != my_localhost) + ((char*)thd->main_security_ctx.host)[MY_MIN(strlen(thd->main_security_ctx.host), + HOSTNAME_LENGTH)] = 0; + thd->main_security_ctx.host_or_ip = thd->main_security_ctx.host; + } + + if (rc == RC_BLOCKED_HOST) + { + /* HOST_CACHE stats updated by ip_to_hostname(). */ + my_error(ER_HOST_IS_BLOCKED, MYF(0), thd->main_security_ctx.host_or_ip); + return 1; + } + } + DBUG_PRINT("info", ("Host: %s ip: %s", + (thd->main_security_ctx.host ? + thd->main_security_ctx.host : "unknown host"), + (thd->main_security_ctx.ip ? + thd->main_security_ctx.ip : "unknown ip"))); + if ((!check_proxy_networks || !is_proxy_protocol_allowed((struct sockaddr *) addr)) + && acl_check_host(thd->main_security_ctx.host, thd->main_security_ctx.ip)) + { + /* HOST_CACHE stats updated by acl_check_host(). */ + my_error(ER_HOST_NOT_PRIVILEGED, MYF(0), + thd->main_security_ctx.host_or_ip); + return 1; + } + return 0; +} + +/* + Perform handshake, authorize client and update thd ACL variables. + + SYNOPSIS + check_connection() + thd thread handle + + RETURN + 0 success, thd is updated. + 1 error +*/ + +static int check_connection(THD *thd) +{ + uint connect_errors= 0; + int auth_rc; + NET *net= &thd->net; + + DBUG_PRINT("info", + ("New connection received on %s", vio_description(net->vio))); + +#ifdef SIGNAL_WITH_VIO_CLOSE + thd->set_active_vio(net->vio); +#endif + + if (!thd->main_security_ctx.host) // If TCP/IP connection + { + my_bool peer_rc; + char ip[NI_MAXHOST]; + uint16 peer_port; + + peer_rc= vio_peer_addr(net->vio, ip, &peer_port, NI_MAXHOST); + + /* + =========================================================================== + DEBUG code only (begin) + Simulate various output from vio_peer_addr(). + =========================================================================== + */ + + DBUG_EXECUTE_IF("vio_peer_addr_error", + { + peer_rc= 1; + } + ); + DBUG_EXECUTE_IF("vio_peer_addr_fake_ipv4", + { + struct sockaddr *sa= (sockaddr *) &net->vio->remote; + sa->sa_family= AF_INET; + struct in_addr *ip4= &((struct sockaddr_in *) sa)->sin_addr; + /* See RFC 5737, 192.0.2.0/24 is reserved. */ + const char* fake= "192.0.2.4"; + inet_pton(AF_INET,fake, ip4); + strcpy(ip, fake); + peer_rc= 0; + } + ); + +#ifdef HAVE_IPV6 + DBUG_EXECUTE_IF("vio_peer_addr_fake_ipv6", + { + struct sockaddr_in6 *sa= (sockaddr_in6 *) &net->vio->remote; + sa->sin6_family= AF_INET6; + struct in6_addr *ip6= & sa->sin6_addr; + /* See RFC 3849, ipv6 2001:DB8::/32 is reserved. */ + const char* fake= "2001:db8::6:6"; + /* inet_pton(AF_INET6, fake, ip6); not available on Windows XP. */ + ip6->s6_addr[ 0] = 0x20; + ip6->s6_addr[ 1] = 0x01; + ip6->s6_addr[ 2] = 0x0d; + ip6->s6_addr[ 3] = 0xb8; + ip6->s6_addr[ 4] = 0x00; + ip6->s6_addr[ 5] = 0x00; + ip6->s6_addr[ 6] = 0x00; + ip6->s6_addr[ 7] = 0x00; + ip6->s6_addr[ 8] = 0x00; + ip6->s6_addr[ 9] = 0x00; + ip6->s6_addr[10] = 0x00; + ip6->s6_addr[11] = 0x00; + ip6->s6_addr[12] = 0x00; + ip6->s6_addr[13] = 0x06; + ip6->s6_addr[14] = 0x00; + ip6->s6_addr[15] = 0x06; + strcpy(ip, fake); + peer_rc= 0; + } + ); +#endif /* HAVE_IPV6 */ + + /* + =========================================================================== + DEBUG code only (end) + =========================================================================== + */ + + if (peer_rc) + { + /* + Since we can not even get the peer IP address, + there is nothing to show in the host_cache, + so increment the global status variable for peer address errors. + */ + statistic_increment(connection_errors_peer_addr, &LOCK_status); + my_error(ER_BAD_HOST_ERROR, MYF(0)); + statistic_increment(aborted_connects_preauth, &LOCK_status); + return 1; + } + + if (thd_set_peer_addr(thd, &net->vio->remote, ip, peer_port, + true, &connect_errors)) + { + statistic_increment(aborted_connects_preauth, &LOCK_status); + return 1; + } + } + else /* Hostname given means that the connection was on a socket */ + { + DBUG_PRINT("info",("Host: %s", thd->main_security_ctx.host)); + thd->main_security_ctx.host_or_ip= thd->main_security_ctx.host; + thd->main_security_ctx.ip= 0; + /* Reset sin_addr */ + bzero((char*) &net->vio->remote, sizeof(net->vio->remote)); + } + vio_keepalive(net->vio, TRUE); + vio_set_keepalive_options(net->vio, &opt_vio_keepalive); + + if (unlikely(thd->packet.alloc(thd->variables.net_buffer_length))) + { + /* + Important note: + net_buffer_length is a SESSION variable, + so it may be tempting to account OOM conditions per IP in the HOST_CACHE, + in case some clients are more demanding than others ... + However, this session variable is *not* initialized with a per client + value during the initial connection, it is initialized from the + GLOBAL net_buffer_length variable from the server. + Hence, there is no reason to account on OOM conditions per client IP, + we count failures in the global server status instead. + */ + statistic_increment(aborted_connects,&LOCK_status); + statistic_increment(connection_errors_internal, &LOCK_status); + statistic_increment(aborted_connects_preauth, &LOCK_status); + return 1; /* The error is set by alloc(). */ + } + + auth_rc= acl_authenticate(thd, 0); + if (auth_rc == 0 && connect_errors != 0) + { + /* + A client connection from this IP was successful, + after some previous failures. + Reset the connection error counter. + */ + reset_host_connect_errors(thd->main_security_ctx.ip); + } + + return auth_rc; +} + + +/* + Setup thread to be used with the current thread + + SYNOPSIS + bool setup_connection_thread_globals() + thd Thread/connection handler + + RETURN + 0 ok + 1 Error (out of memory) + In this case we will close the connection and increment status +*/ + +void setup_connection_thread_globals(THD *thd) +{ + thd->store_globals(); +} + + +/* + Autenticate user, with error reporting + + SYNOPSIS + login_connection() + thd Thread handler + + NOTES + Connection is not closed in case of errors + + RETURN + 0 ok + 1 error +*/ + +bool login_connection(THD *thd) +{ + NET *net= &thd->net; + int error= 0; + DBUG_ENTER("login_connection"); + DBUG_PRINT("info", ("login_connection called by thread %lu", + (ulong) thd->thread_id)); + + /* Use "connect_timeout" value during connection phase */ + my_net_set_read_timeout(net, connect_timeout); + my_net_set_write_timeout(net, connect_timeout); + + error= check_connection(thd); + thd->protocol->end_statement(); + + if (unlikely(error)) + { // Wrong permissions +#ifdef _WIN32 + if (vio_type(net->vio) == VIO_TYPE_NAMEDPIPE) + my_sleep(1000); /* must wait after eof() */ +#endif + statistic_increment(aborted_connects,&LOCK_status); + error=1; + goto exit; + } + /* Connect completed, set read/write timeouts back to default */ + my_net_set_read_timeout(net, thd->variables.net_read_timeout); + my_net_set_write_timeout(net, thd->variables.net_write_timeout); + + /* Updates global user connection stats. */ + if (increment_connection_count(thd, TRUE)) + { + my_error(ER_OUTOFMEMORY, MYF(0), (int) (2*sizeof(USER_STATS))); + error= 1; + goto exit; + } + +exit: + mysql_audit_notify_connection_connect(thd); + DBUG_RETURN(error); +} + + +/* + Close an established connection + + NOTES + This mainly updates status variables +*/ + +void end_connection(THD *thd) +{ + NET *net= &thd->net; + +#ifdef WITH_WSREP + if (thd->wsrep_cs().state() == wsrep::client_state::s_exec) + { + /* Error happened after the thread acquired ownership to wsrep + client state, but before command was processed. Clean up the + state before wsrep_close(). */ + wsrep_after_command_ignore_result(thd); + } + wsrep_close(thd); +#endif /* WITH_WSREP */ + if (thd->user_connect) + { + /* + We decrease this variable early to make it easy to log again quickly. + This code is not critical as we will in any case do this test + again in thd->cleanup() + */ + decrease_user_connections(thd->user_connect); + /* + The thread may returned back to the pool and assigned to a user + that doesn't have a limit. Ensure the user is not using resources + of someone else. + */ + thd->user_connect= NULL; + } + + if (unlikely(thd->killed) || (net->error && net->vio != 0)) + { + statistic_increment(aborted_threads,&LOCK_status); + status_var_increment(thd->status_var.lost_connections); + } + + if (likely(!thd->killed) && (net->error && net->vio != 0)) + thd->print_aborted_warning(1, thd->get_stmt_da()->is_error() + ? thd->get_stmt_da()->message() : ER_THD(thd, ER_UNKNOWN_ERROR)); +} + + +/* + Initialize THD to handle queries +*/ + +void prepare_new_connection_state(THD* thd) +{ + Security_context *sctx= thd->security_ctx; + + if (thd->client_capabilities & CLIENT_COMPRESS) + thd->net.compress=1; // Use compression + + /* + Much of this is duplicated in create_embedded_thd() for the + embedded server library. + TODO: refactor this to avoid code duplication there + */ + thd->proc_info= 0; + thd->set_command(COM_SLEEP); + thd->init_for_queries(); + + if (opt_init_connect.length && + !(sctx->master_access & PRIV_IGNORE_INIT_CONNECT)) + { + execute_init_command(thd, &opt_init_connect, &LOCK_sys_init_connect); + if (unlikely(thd->is_error())) + { + Host_errors errors; + thd->set_killed(KILL_CONNECTION); + thd->print_aborted_warning(0, "init_connect command failed"); + sql_print_warning("%s", thd->get_stmt_da()->message()); + + /* + now let client to send its first command, + to be able to send the error back + */ + NET *net= &thd->net; + thd->lex->current_select= 0; + my_net_set_read_timeout(net, thd->variables.net_wait_timeout); + thd->clear_error(); + net_new_transaction(net); + ulong packet_length= my_net_read(net); + /* + If my_net_read() failed, my_error() has been already called, + and the main Diagnostics Area contains an error condition. + */ + if (packet_length != packet_error) + my_error(ER_NEW_ABORTING_CONNECTION, MYF(0), + thd->thread_id, + thd->db.str ? thd->db.str : "unconnected", + sctx->user ? sctx->user : "unauthenticated", + sctx->host_or_ip, "init_connect command failed"); + thd->server_status&= ~SERVER_STATUS_CLEAR_SET; + thd->protocol->end_statement(); + thd->killed = KILL_CONNECTION; + errors.m_init_connect= 1; + inc_host_errors(thd->main_security_ctx.ip, &errors); + return; + } + + thd->proc_info=0; + } +} + + +/* + Thread handler for a connection + + SYNOPSIS + handle_one_connection() + arg Connection object (THD) + + IMPLEMENTATION + This function (normally) does the following: + - Initialize thread + - Initialize THD to be used with this thread + - Authenticate user + - Execute all queries sent on the connection + - Take connection down + - End thread / Handle next connection using thread from thread cache +*/ + +pthread_handler_t handle_one_connection(void *arg) +{ + CONNECT *connect= (CONNECT*) arg; + + mysql_thread_set_psi_id(connect->thread_id); + + if (init_new_connection_handler_thread()) + connect->close_with_error(0, 0, ER_OUT_OF_RESOURCES); + else + do_handle_one_connection(connect, true); + + DBUG_PRINT("info", ("killing thread")); +#if defined(HAVE_OPENSSL) && !defined(EMBEDDED_LIBRARY) + ERR_remove_state(0); +#endif + my_thread_end(); + return 0; +} + +bool thd_prepare_connection(THD *thd) +{ + bool rc; + lex_start(thd); + rc= login_connection(thd); + if (rc) + return rc; + + MYSQL_CONNECTION_START(thd->thread_id, &thd->security_ctx->priv_user[0], + (char *) thd->security_ctx->host_or_ip); + + prepare_new_connection_state(thd); +#ifdef WITH_WSREP + thd->wsrep_client_thread= true; + wsrep_open(thd); +#endif /* WITH_WSREP */ + return FALSE; +} + +bool thd_is_connection_alive(THD *thd) +{ + NET *net= &thd->net; + if (likely(!net->error && + net->vio != 0 && + thd->killed < KILL_CONNECTION)) + return TRUE; + return FALSE; +} + + +void do_handle_one_connection(CONNECT *connect, bool put_in_cache) +{ + ulonglong thr_create_utime= microsecond_interval_timer(); + THD *thd; + if (!(thd= connect->create_thd(NULL))) + { + connect->close_and_delete(); + return; + } + + DBUG_EXECUTE_IF("CONNECT_wait", + { + extern MYSQL_SOCKET unix_sock; + DBUG_ASSERT(unix_sock.fd >= 0); + while (unix_sock.fd >= 0) + my_sleep(1000); + }); + + /* + If a thread was created to handle this connection: + increment slow_launch_threads counter if it took more than + slow_launch_time seconds to create the thread. + */ + + if (connect->prior_thr_create_utime) + { + ulong launch_time= (ulong) (thr_create_utime - + connect->prior_thr_create_utime); + if (launch_time >= slow_launch_time*1000000L) + statistic_increment(slow_launch_threads, &LOCK_status); + } + + server_threads.insert(thd); // Make THD visible in show processlist + + delete connect; // must be after server_threads.insert, see close_connections() + + thd->thr_create_utime= thr_create_utime; + /* We need to set this because of time_out_user_resource_limits */ + thd->start_utime= thr_create_utime; + + /* + handle_one_connection() is normally the only way a thread would + start and would always be on the very high end of the stack , + therefore, the thread stack always starts at the address of the + first local variable of handle_one_connection, which is thd. We + need to know the start of the stack so that we could check for + stack overruns. + */ + thd->thread_stack= (char*) &thd; + setup_connection_thread_globals(thd); + + for (;;) + { + bool create_user= TRUE; + + mysql_socket_set_thread_owner(thd->net.vio->mysql_socket); + if (thd_prepare_connection(thd)) + { + create_user= FALSE; + goto end_thread; + } + + while (thd_is_connection_alive(thd)) + { + if (mysql_audit_release_required(thd)) + mysql_audit_release(thd); + if (do_command(thd)) + break; + } + end_connection(thd); + +end_thread: + close_connection(thd); + + if (thd->userstat_running) + update_global_user_stats(thd, create_user, time(NULL)); + + unlink_thd(thd); + if (IF_WSREP(thd->wsrep_applier, false) || !put_in_cache || + !(connect= thread_cache.park())) + break; + + /* Create new instrumentation for the new THD job */ + PSI_CALL_set_thread(PSI_CALL_new_thread(key_thread_one_connection, thd, + thd->thread_id)); + + if (!(connect->create_thd(thd))) + { + /* Out of resources. Free thread to get more resources */ + connect->close_and_delete(); + break; + } + delete connect; + + /* + We have to call store_globals to update mysys_var->id and lock_info + with the new thread_id + */ + thd->store_globals(); + + /* reset abort flag for the thread */ + thd->mysys_var->abort= 0; + thd->thr_create_utime= microsecond_interval_timer(); + thd->start_utime= thd->thr_create_utime; + + server_threads.insert(thd); + } + delete thd; +} +#endif /* EMBEDDED_LIBRARY */ + + +/* Handling of CONNECT objects */ + +/* + Close connection without error and delete the connect object + This and close_with_error are only called if we didn't manage to + create a new thd object. +*/ + +void CONNECT::close_and_delete() +{ + DBUG_ENTER("close_and_delete"); + +#if _WIN32 + if (vio_type == VIO_TYPE_NAMEDPIPE) + CloseHandle(pipe); + else +#endif + if (vio_type != VIO_CLOSED) + mysql_socket_close(sock); + vio_type= VIO_CLOSED; + + --*scheduler->connection_count; + statistic_increment(connection_errors_internal, &LOCK_status); + statistic_increment(aborted_connects,&LOCK_status); + + delete this; + DBUG_VOID_RETURN; +} + +/* + Close a connection with a possible error to the end user + Alse deletes the connection object, like close_and_delete() +*/ + +void CONNECT::close_with_error(uint sql_errno, + const char *message, uint close_error) +{ + THD *thd= create_thd(NULL); + if (thd) + { + if (sql_errno) + thd->protocol->net_send_error(thd, sql_errno, message, NULL); + close_connection(thd, close_error); + delete thd; + set_current_thd(0); + } + close_and_delete(); +} + + +/* Reuse or create a THD based on a CONNECT object */ + +THD *CONNECT::create_thd(THD *thd) +{ + bool res, thd_reused= thd != 0; + Vio *vio; + DBUG_ENTER("create_thd"); + + DBUG_EXECUTE_IF("simulate_failed_connection_2", DBUG_RETURN(0); ); + + if (thd) + { + /* reuse old thd */ + thd->reset_for_reuse(); + /* + reset tread_id's, but not thread_dbug_id's as the later isn't allowed + to change as there is already structures in thd marked with the old + value. + */ + thd->thread_id= thd->variables.pseudo_thread_id= thread_id; + } + else if (!(thd= new THD(thread_id))) + DBUG_RETURN(0); + +#if _WIN32 + if (vio_type == VIO_TYPE_NAMEDPIPE) + vio= vio_new_win32pipe(pipe); + else +#endif + vio= mysql_socket_vio_new(sock, vio_type, vio_type == VIO_TYPE_SOCKET ? + VIO_LOCALHOST : 0); + if (!vio) + { + if (!thd_reused) + delete thd; + DBUG_RETURN(0); + } + + set_current_thd(thd); + res= my_net_init(&thd->net, vio, thd, MYF(MY_THREAD_SPECIFIC)); + vio_type= VIO_CLOSED; // Vio now handled by thd + + if (unlikely(res || thd->is_error())) + { + if (!thd_reused) + delete thd; + set_current_thd(0); + DBUG_RETURN(0); + } + + init_net_server_extension(thd); + + thd->security_ctx->host= thd->net.vio->type == VIO_TYPE_NAMEDPIPE || + thd->net.vio->type == VIO_TYPE_SOCKET ? + my_localhost : 0; + + thd->scheduler= scheduler; + thd->real_id= pthread_self(); /* Duplicates THD::store_globals() setting. */ + + /* Attach PSI instrumentation to the new THD */ + + PSI_thread *psi= PSI_CALL_get_thread(); + PSI_CALL_set_thread_os_id(psi); + PSI_CALL_set_thread_THD(psi, thd); + PSI_CALL_set_thread_id(psi, thd->thread_id); + thd->set_psi(psi); + + DBUG_RETURN(thd); +} |