File: sql_auth_cache.cc

package info (click to toggle)
mysql-8.0 8.0.44-1
links: PTS, VCS
area: main
in suites: sid
size: 1,272,892 kB
sloc: cpp: 4,685,345; ansic: 412,712; pascal: 108,395; java: 83,641; perl: 30,221; cs: 27,067; sql: 26,594; python: 21,816; sh: 17,285; yacc: 17,169; php: 11,522; xml: 7,388; javascript: 7,083; makefile: 1,793; lex: 1,075; awk: 670; asm: 520; objc: 183; ruby: 97; lisp: 86
file content (4069 lines) | stat: -rw-r--r-- 140,745 bytes
/* Copyright (c) 2000, 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 "sql/auth/sql_auth_cache.h"

#include <stdarg.h>
#include <boost/graph/properties.hpp>
#include <new>

#include "m_ctype.h"
#include "m_string.h"  // LEX_CSTRING
#include "mutex_lock.h"
#include "my_base.h"
#include "my_compiler.h"
#include "my_dbug.h"
#include "my_loglevel.h"
#include "my_macros.h"
#include "mysql/components/services/bits/psi_bits.h"
#include "mysql/components/services/bits/psi_mutex_bits.h"
#include "mysql/components/services/log_builtins.h"
#include "mysql/plugin.h"
#include "mysql/plugin_audit.h"
#include "mysql/plugin_auth.h"  // st_mysql_auth
#include "mysql/psi/mysql_mutex.h"
#include "mysql/service_mysql_alloc.h"
#include "mysqld_error.h"
#include "prealloced_array.h"
#include "sql/auth/auth_acls.h"
#include "sql/auth/auth_common.h"    // ACL_internal_schema_access
#include "sql/auth/auth_internal.h"  // auth_plugin_is_built_in
#include "sql/auth/auth_utility.h"
#include "sql/auth/dynamic_privilege_table.h"
#include "sql/auth/sql_authentication.h"  // g_cached_authentication_plugins
#include "sql/auth/sql_security_ctx.h"
#include "sql/auth/sql_user_table.h"
#include "sql/auth/user_table.h"  // read_user_table
#include "sql/current_thd.h"      // current_thd
#include "sql/debug_sync.h"
#include "sql/error_handler.h"  // Internal_error_handler
#include "sql/field.h"          // Field
#include "sql/handler.h"
#include "sql/iterators/row_iterator.h"
#include "sql/key.h"
#include "sql/mdl.h"
#include "sql/mysqld.h"          // my_localhost
#include "sql/psi_memory_key.h"  // key_memory_acl_mem
#include "sql/set_var.h"
#include "sql/sql_audit.h"
#include "sql/sql_base.h"   // open_and_lock_tables
#include "sql/sql_class.h"  // THD
#include "sql/sql_const.h"
#include "sql/sql_error.h"
#include "sql/sql_executor.h"  // unique_ptr_destroy_only<RowIterator>
#include "sql/sql_lex.h"
#include "sql/sql_plugin.h"  // my_plugin_lock_by_name
#include "sql/sql_plugin_ref.h"
#include "sql/ssl_acceptor_context_operator.h"
#include "sql/system_variables.h"
#include "sql/table.h"  // TABLE
#include "sql/thd_raii.h"
#include "sql/thr_malloc.h"
#include "sql/tztime.h"  // Time_zone
#include "sql/xa.h"
#include "sql_string.h"
#include "thr_lock.h"
#include "thr_mutex.h"

#define INVALID_DATE "0000-00-00 00:00:00"

#include <algorithm>
#include <functional>
#include <unordered_map>
#include <utility>
#include <vector>

using std::make_unique;
using std::min;
using std::string;
using std::unique_ptr;

PSI_mutex_key key_LOCK_acl_cache_flush;
PSI_mutex_info all_acl_cache_mutexes[] = {
    {&key_LOCK_acl_cache_flush, "LOCK_acl_cache_flush", PSI_FLAG_SINGLETON, 0,
     PSI_DOCUMENT_ME}};
Acl_cache *g_acl_cache = nullptr;
Acl_cache *get_global_acl_cache() { return g_acl_cache; }
ulong get_global_acl_cache_size() { return g_acl_cache->size(); }
void init_acl_cache();
extern Role_index_map *g_authid_to_vertex;
extern Granted_roles_graph *g_granted_roles;
#include <boost/property_map/property_map.hpp>

struct ACL_internal_schema_registry_entry {
  const LEX_CSTRING *m_name;
  const ACL_internal_schema_access *m_access;
};
/**
  Internal schema registered.
  Currently, this is only:
  - performance_schema
  - information_schema,
  This can be reused later for:
  - mysql
*/
static ACL_internal_schema_registry_entry registry_array[2];
static uint m_registry_array_size = 0;

MEM_ROOT global_acl_memory;
MEM_ROOT memex;
Prealloced_array<ACL_USER, ACL_PREALLOC_SIZE> *acl_users = nullptr;
Prealloced_array<ACL_PROXY_USER, ACL_PREALLOC_SIZE> *acl_proxy_users = nullptr;
Prealloced_array<ACL_DB, ACL_PREALLOC_SIZE> *acl_dbs = nullptr;
Prealloced_array<ACL_HOST_AND_IP, ACL_PREALLOC_SIZE> *acl_wild_hosts = nullptr;
Db_access_map acl_db_map;
Default_roles *g_default_roles = nullptr;
std::vector<Role_id> *g_mandatory_roles = nullptr;

unique_ptr<
    malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_TABLE>>>
    column_priv_hash;
unique_ptr<
    malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_NAME>>>
    proc_priv_hash, func_priv_hash;
malloc_unordered_map<std::string, unique_ptr_my_free<acl_entry>> db_cache{
    key_memory_acl_cache};
collation_unordered_map<std::string, ACL_USER *> *acl_check_hosts = nullptr;
unique_ptr<Acl_restrictions> acl_restrictions = nullptr;

/**
  A hashmap on user part of account name for quick lookup.
*/
typedef std::unordered_map<
    std::string, Acl_user_ptr_list, std::hash<std::string>,
    std::equal_to<std::string>,
    Acl_cache_allocator<std::pair<const std::string, Acl_user_ptr_list>>>
    Name_to_userlist;
Name_to_userlist *name_to_userlist = nullptr;

bool initialized = false;
bool skip_grant_tables(void) { return !initialized; }
bool acl_cache_initialized = false;
bool allow_all_hosts = true;
uint grant_version = 0; /* Version of priv tables */
bool validate_user_plugins = true;

#define IP_ADDR_STRLEN (3 + 1 + 3 + 1 + 3 + 1 + 3)
#define ACL_KEY_LENGTH (IP_ADDR_STRLEN + 1 + NAME_LEN + 1 + USERNAME_LENGTH + 1)

/** Helper: Set user name */
static void set_username(char **user, const char *user_arg, MEM_ROOT *mem) {
  assert(user != nullptr);
  *user = (user_arg && *user_arg) ? strdup_root(mem, user_arg) : nullptr;
}

/** Helper: Set host name */
static void set_hostname(ACL_HOST_AND_IP *host, const char *host_arg,
                         MEM_ROOT *mem) {
  assert(host != nullptr);
  host->update_hostname((host_arg && *host_arg) ? strdup_root(mem, host_arg)
                                                : nullptr);
}

/**
  Allocates the memory in the the global_acl_memory MEM_ROOT.
*/
void init_acl_memory() {
  init_sql_alloc(key_memory_acl_mem, &global_acl_memory, ACL_ALLOC_BLOCK_SIZE);
}

/**
  Add an internal schema to the registry.
  @param name the schema name
  @param access the schema ACL specific rules
*/
void ACL_internal_schema_registry::register_schema(
    const LEX_CSTRING &name, const ACL_internal_schema_access *access) {
  assert(m_registry_array_size < array_elements(registry_array));

  /* Not thread safe, and does not need to be. */
  registry_array[m_registry_array_size].m_name = &name;
  registry_array[m_registry_array_size].m_access = access;
  m_registry_array_size++;
}

/**
  Search per internal schema ACL by name.
  @param name a schema name
  @return per schema rules, or NULL
*/
const ACL_internal_schema_access *ACL_internal_schema_registry::lookup(
    const char *name) {
  assert(name != nullptr);

  uint i;

  for (i = 0; i < m_registry_array_size; i++) {
    if (my_strcasecmp(system_charset_info, registry_array[i].m_name->str,
                      name) == 0)
      return registry_array[i].m_access;
  }
  return nullptr;
}

bool ACL_HOST_AND_IP::calc_cidr_mask(const char *ip_arg, long *val) {
  long tmp;
  if (!(ip_arg = str2int(ip_arg, 10, 0, 32, &tmp)) || *ip_arg != '\0')
    return true;

  /* Create IP mask. */
  *val = UINT_MAX32 << (32 - tmp);

  return false;
}

bool ACL_HOST_AND_IP::calc_ip_mask(const char *ip_arg, long *val) {
  long tmp = 0;
  if (!(ip_arg = calc_ip(ip_arg, &tmp)) || *ip_arg != '\0') return true;

  /* Valid IP mask must be continuous bit flags. */
  if (((~tmp & UINT_MAX32) + 1) & ~tmp) return true;

  *val = tmp;

  return false;
}

const char *ACL_HOST_AND_IP::calc_ip(const char *ip_arg, long *val) {
  long ip_val, tmp;
  if (!(ip_arg = str2int(ip_arg, 10, 0, 255, &ip_val)) || *ip_arg != '.')
    return nullptr;
  ip_val <<= 24;
  if (!(ip_arg = str2int(ip_arg + 1, 10, 0, 255, &tmp)) || *ip_arg != '.')
    return nullptr;
  ip_val += tmp << 16;
  if (!(ip_arg = str2int(ip_arg + 1, 10, 0, 255, &tmp)) || *ip_arg != '.')
    return nullptr;
  ip_val += tmp << 8;
  if (!(ip_arg = str2int(ip_arg + 1, 10, 0, 255, &tmp))) return nullptr;
  *val = ip_val + tmp;
  return ip_arg;
}

/**
  @brief Update the hostname. Updates ip and ip_mask accordingly.

  @param host_arg Value to be stored
 */
void ACL_HOST_AND_IP::update_hostname(const char *host_arg) {
  hostname = host_arg;  // This will not be modified!

  ip = ip_mask = 0;

  if (!host_arg) {
    hostname_length = 0;
    return;
  }

  hostname_length = strlen(hostname);

  if ((host_arg = calc_ip(host_arg, &ip))) {
    if (*host_arg == '\0') {
      /* There is only IP part specified. */
      ip_mask_type = ip_mask_type_implicit;
      ip_mask = UINT_MAX32;
    } else if (*host_arg == '/') {
      if (!calc_ip_mask(host_arg + 1, &ip_mask)) {
        ip_mask_type = ip_mask_type_subnet;
      } else if (!calc_cidr_mask(host_arg + 1, &ip_mask)) {
        ip_mask_type = ip_mask_type_cidr;
      } else
        /* Invalid or unsupported IP mask.*/
        ip = 0;
    }
  }
}

/*
   @brief Comparing of hostnames.

   @TODO This function should ideally only
   be called during authentication and not from authorization code. You may
   authenticate with a hostmask, but all authentication should be against a
   specific security context with a specific authentication ID.

   @param  host_arg    Hostname to be compared with
   @param  ip_arg      IP address to be compared with

   @notes
   A hostname may be of type:
   1) hostname   (May include wildcards);   monty.pp.sci.fi
   2) ip     (May include wildcards);   192.168.0.0
   3) ip/netmask                        192.168.0.0/255.255.255.0
   A net mask of 0.0.0.0 is not allowed.

   @return
   true   if matched
   false  if not matched
 */

bool ACL_HOST_AND_IP::compare_hostname(const char *host_arg,
                                       const char *ip_arg) {
  long tmp;
  const char *p;
  if (ip_mask && ip_arg && (p = calc_ip(ip_arg, &tmp)) && *p == '\0') {
    return (tmp & ip_mask) == ip;
  }
  return (!hostname ||
          (host_arg &&
           !wild_case_compare(system_charset_info, host_arg, hostname)) ||
          (ip_arg && !wild_compare(ip_arg, strlen(ip_arg), hostname,
                                   strlen(hostname), false)));
}

ACL_USER::ACL_USER() {
  /* ACL_ACCESS is initialized by its constructor */
  {
    /* USER_RESOURCES */
    user_resource.questions = 0;
    user_resource.updates = 0;
    user_resource.conn_per_hour = 0;
    user_resource.user_conn = 0;
    user_resource.specified_limits = 0;
  }

  user = nullptr;

  {
    /* TLS restrictions */
    ssl_type = SSL_TYPE_NONE;
    ssl_cipher = nullptr;
    x509_issuer = nullptr;
    x509_subject = nullptr;
  }

  plugin = EMPTY_CSTR;
  password_expired = false;
  can_authenticate = false;
  password_last_changed.time_type = MYSQL_TIMESTAMP_ERROR;
  password_lifetime = 0;
  use_default_password_lifetime = false;
  account_locked = false;
  is_role = false;
  password_history_length = 0;
  use_default_password_history = false;
  password_reuse_interval = 0;
  use_default_password_reuse_interval = false;
  password_require_current = Lex_acl_attrib_udyn::DEFAULT;
  m_mfa = nullptr;
  /* Acl_credentials is initialized by its constructor */
}

void ACL_USER::Password_locked_state::set_temporary_lock_state_parameters(
    uint remaining_login_attempts, long daynr_locked) {
  m_remaining_login_attempts = remaining_login_attempts;
  m_daynr_locked = daynr_locked;
}

void ACL_USER::Password_locked_state::set_parameters(
    int password_lock_time_days, uint failed_login_attempts) {
  m_password_lock_time_days = password_lock_time_days;
  m_failed_login_attempts = failed_login_attempts;
  set_temporary_lock_state_parameters(failed_login_attempts, 0);
  assert(is_default() == true);
}

/**
  Updates the password locked state based on the time of day fetched from the
  THD

  @param thd the session to use to calculate time
  @param successful_login true if the login succeeded
  @param[out] ret_days_remaining remaining number of days. Filled only if
  update returns locked account
  @retval false account not locked
  @retval true account locked
*/
bool ACL_USER::Password_locked_state::update(THD *thd, bool successful_login,
                                             long *ret_days_remaining) {
  /* stop if the user is not tracking failed logins */
  if (!is_active()) return false;

  /* reset on a successful login if the account is not locked */
  if (successful_login && m_daynr_locked == 0) {
    m_remaining_login_attempts = m_failed_login_attempts;
    return false;
  }

  /* decreases the remaining login attempts if any */
  if (!successful_login && m_remaining_login_attempts > 0) {
    m_remaining_login_attempts--;
    assert(m_daynr_locked == 0);
  }

  if (m_remaining_login_attempts) return false;

  long now_day;
  /* fetch the current day */
  MYSQL_TIME tm_now;
  thd->time_zone()->gmt_sec_to_TIME(&tm_now, thd->query_start_timeval_trunc(6));
  now_day = calc_daynr(tm_now.year, tm_now.month, tm_now.day);

  DBUG_EXECUTE_IF("account_lock_daynr_add_one", { now_day += 1; });

  DBUG_EXECUTE_IF("account_lock_daynr_add_ten", { now_day += 10; });

  /* last unsuccessful login. lock the account */
  if (m_daynr_locked == 0) {
    assert(!successful_login);
    m_daynr_locked = now_day;
    *ret_days_remaining = m_password_lock_time_days;
    return true;
  };

  /* if the lock should never expire we stop here */
  if (m_daynr_locked > 0 && m_password_lock_time_days < 0) return true;

  /* check if the account is still to be locked */
  if (now_day - m_daynr_locked < (long)m_password_lock_time_days) {
    *ret_days_remaining =
        ((long)m_password_lock_time_days) - (now_day - m_daynr_locked);
    return true;
  }
  /* reset the account lock if the time has expired */
  if (now_day - m_daynr_locked >= (long)m_password_lock_time_days) {
    m_daynr_locked = 0;
    m_remaining_login_attempts = m_failed_login_attempts;
    return false;
  }

  /* it should never get to here */
  assert(false);
  return false;
}

ACL_USER *ACL_USER::copy(MEM_ROOT *root) {
  ACL_USER *dst = (ACL_USER *)root->Alloc(sizeof(ACL_USER));
  if (!dst) return nullptr;
  *dst = *this;
  dst->user = safe_strdup_root(root, user);
  dst->ssl_cipher = safe_strdup_root(root, ssl_cipher);
  dst->x509_issuer = safe_strdup_root(root, x509_issuer);
  dst->x509_subject = safe_strdup_root(root, x509_subject);
  /*
     If the plugin is built in we don't need to reallocate the name of the
     plugin.
   */
  if (auth_plugin_is_built_in(dst->plugin.str))
    dst->plugin = plugin;
  else {
    dst->plugin.str = strmake_root(root, plugin.str, plugin.length);
    dst->plugin.length = plugin.length;
  }
  for (int i = 0; i < NUM_CREDENTIALS; ++i) {
    dst->credentials[i].m_auth_string.str =
        safe_strdup_root(root, credentials[i].m_auth_string.str);
    dst->credentials[i].m_auth_string.length =
        credentials[i].m_auth_string.length;
    dst->credentials[i].m_salt_len = credentials[i].m_salt_len;
    memcpy(dst->credentials[i].m_salt, credentials[i].m_salt,
           credentials[i].m_salt_len);
  }
  dst->host.update_hostname(safe_strdup_root(root, host.get_host()));
  dst->password_require_current = password_require_current;
  dst->password_locked_state = password_locked_state;
  dst->set_mfa(root, m_mfa);
  return dst;
}

void ACL_USER::set_user(MEM_ROOT *mem, const char *user_arg) {
  set_username(&user, user_arg, mem);
}

void ACL_USER::set_host(MEM_ROOT *mem, const char *host_arg) {
  set_hostname(&host, host_arg, mem);
}

void ACL_USER::set_mfa(MEM_ROOT *mem, I_multi_factor_auth *m) {
  if (mem && m) {
    m_mfa = new (mem) Multi_factor_auth_list(mem);
    Multi_factor_auth_list *auth_list = m->get_multi_factor_auth_list();
    /*
      iterate over list of auth factors and make a new copy of each
      individual auth factors
    */
    for (auto m_it : auth_list->get_mfa_list()) {
      Multi_factor_auth_info *af = m_it->get_multi_factor_auth_info();
      m_mfa->add_factor(new (mem)
                            Multi_factor_auth_info(mem, af->get_lex_mfa()));
    }
  } else {
    m_mfa = m;
  }
}

void ACL_PROXY_USER::init(const char *host_arg, const char *user_arg,
                          const char *proxied_host_arg,
                          const char *proxied_user_arg, bool with_grant_arg) {
  user = (user_arg && *user_arg) ? user_arg : nullptr;
  host.update_hostname((host_arg && *host_arg) ? host_arg : nullptr);
  proxied_user =
      (proxied_user_arg && *proxied_user_arg) ? proxied_user_arg : nullptr;
  proxied_host.update_hostname(
      (proxied_host_arg && *proxied_host_arg) ? proxied_host_arg : nullptr);
  with_grant = with_grant_arg;
  sort =
      get_sort(4, host.get_host(), user, proxied_host.get_host(), proxied_user);
}

void ACL_PROXY_USER::init(MEM_ROOT *mem, const char *host_arg,
                          const char *user_arg, const char *proxied_host_arg,
                          const char *proxied_user_arg, bool with_grant_arg) {
  init((host_arg && *host_arg) ? strdup_root(mem, host_arg) : nullptr,
       (user_arg && *user_arg) ? strdup_root(mem, user_arg) : nullptr,
       (proxied_host_arg && *proxied_host_arg)
           ? strdup_root(mem, proxied_host_arg)
           : nullptr,
       (proxied_user_arg && *proxied_user_arg)
           ? strdup_root(mem, proxied_user_arg)
           : nullptr,
       with_grant_arg);
}

void ACL_PROXY_USER::init(TABLE *table, MEM_ROOT *mem) {
  init(get_field(mem, table->field[MYSQL_PROXIES_PRIV_HOST]),
       get_field(mem, table->field[MYSQL_PROXIES_PRIV_USER]),
       get_field(mem, table->field[MYSQL_PROXIES_PRIV_PROXIED_HOST]),
       get_field(mem, table->field[MYSQL_PROXIES_PRIV_PROXIED_USER]),
       table->field[MYSQL_PROXIES_PRIV_WITH_GRANT]->val_int() != 0);
}

bool ACL_PROXY_USER::check_validity(bool check_no_resolve) {
  if (check_no_resolve &&
      (hostname_requires_resolving(host.get_host()) ||
       hostname_requires_resolving(proxied_host.get_host())) &&
      strcmp(host.get_host(), "localhost") != 0) {
    LogErr(WARNING_LEVEL, ER_AUTHCACHE_PROXIES_PRIV_SKIPPED_NEEDS_RESOLVE,
           proxied_user ? proxied_user : "",
           proxied_host.get_host() ? proxied_host.get_host() : "",
           user ? user : "", host.get_host() ? host.get_host() : "");
  }
  return false;
}

bool ACL_PROXY_USER::matches(const char *host_arg, const char *user_arg,
                             const char *ip_arg, const char *proxied_user_arg,
                             bool any_proxy_user) {
  DBUG_TRACE;
  DBUG_PRINT("info",
             ("compare_hostname(%s,%s,%s) &&"
              "compare_hostname(%s,%s,%s) &&"
              "wild_compare (%s,%s) &&"
              "wild_compare (%s,%s)",
              host.get_host() ? host.get_host() : "<NULL>",
              host_arg ? host_arg : "<NULL>", ip_arg ? ip_arg : "<NULL>",
              proxied_host.get_host() ? proxied_host.get_host() : "<NULL>",
              host_arg ? host_arg : "<NULL>", ip_arg ? ip_arg : "<NULL>",
              user_arg ? user_arg : "<NULL>", user ? user : "<NULL>",
              proxied_user_arg ? proxied_user_arg : "<NULL>",
              proxied_user ? proxied_user : "<NULL>"));
  return host.compare_hostname(host_arg, ip_arg) &&
         proxied_host.compare_hostname(host_arg, ip_arg) &&
         (!user || (user_arg && !wild_compare(user_arg, strlen(user_arg), user,
                                              strlen(user), true))) &&
         (any_proxy_user || !proxied_user ||
          (proxied_user &&
           !wild_compare(proxied_user_arg, strlen(proxied_user_arg),
                         proxied_user, strlen(proxied_user), true)));
}

bool ACL_PROXY_USER::pk_equals(ACL_PROXY_USER *grant) {
  DBUG_TRACE;
  DBUG_PRINT("info",
             ("strcmp(%s,%s) &&"
              "strcmp(%s,%s) &&"
              "wild_compare (%s,%s) &&"
              "wild_compare (%s,%s)",
              user ? user : "<NULL>", grant->user ? grant->user : "<NULL>",
              proxied_user ? proxied_user : "<NULL>",
              grant->proxied_user ? grant->proxied_user : "<NULL>",
              host.get_host() ? host.get_host() : "<NULL>",
              grant->host.get_host() ? grant->host.get_host() : "<NULL>",
              proxied_host.get_host() ? proxied_host.get_host() : "<NULL>",
              grant->proxied_host.get_host() ? grant->proxied_host.get_host()
                                             : "<NULL>"));

  return auth_element_equals(user, grant->user) &&
         auth_element_equals(proxied_user, grant->proxied_user) &&
         auth_element_equals(host.get_host(), grant->host.get_host()) &&
         auth_element_equals(proxied_host.get_host(),
                             grant->proxied_host.get_host());
}

void ACL_PROXY_USER::print_grant(THD *thd, String *str) {
  str->append(STRING_WITH_LEN("GRANT PROXY ON "));
  append_auth_id_string(thd, proxied_user, get_proxied_user_length(),
                        proxied_host.get_host(), proxied_host.get_host_len(),
                        str);
  str->append(STRING_WITH_LEN(" TO "));
  append_auth_id_string(thd, user, get_user_length(), host.get_host(),
                        host.get_host_len(), str);
  if (with_grant) str->append(STRING_WITH_LEN(" WITH GRANT OPTION"));
}

int ACL_PROXY_USER::store_pk(TABLE *table, const LEX_CSTRING &hostname,
                             const LEX_CSTRING &user,
                             const LEX_CSTRING &proxied_host,
                             const LEX_CSTRING &proxied_user) {
  DBUG_TRACE;
  DBUG_PRINT("info", ("host=%s, user=%s, proxied_host=%s, proxied_user=%s",
                      hostname.str ? hostname.str : "<NULL>",
                      user.str ? user.str : "<NULL>",
                      proxied_host.str ? proxied_host.str : "<NULL>",
                      proxied_user.str ? proxied_user.str : "<NULL>"));
  if (table->field[MYSQL_PROXIES_PRIV_HOST]->store(
          hostname.str, hostname.length, system_charset_info))
    return true;
  if (table->field[MYSQL_PROXIES_PRIV_USER]->store(user.str, user.length,
                                                   system_charset_info))
    return true;
  if (table->field[MYSQL_PROXIES_PRIV_PROXIED_HOST]->store(
          proxied_host.str, proxied_host.length, system_charset_info))
    return true;
  if (table->field[MYSQL_PROXIES_PRIV_PROXIED_USER]->store(
          proxied_user.str, proxied_user.length, system_charset_info))
    return true;

  return false;
}

int ACL_PROXY_USER::store_with_grant(TABLE *table, bool with_grant) {
  DBUG_TRACE;
  DBUG_PRINT("info", ("with_grant=%s", with_grant ? "TRUE" : "FALSE"));
  if (table->field[MYSQL_PROXIES_PRIV_WITH_GRANT]->store(with_grant ? 1 : 0,
                                                         true))
    return true;

  return false;
}

int ACL_PROXY_USER::store_data_record(TABLE *table, const LEX_CSTRING &hostname,
                                      const LEX_CSTRING &user,
                                      const LEX_CSTRING &proxied_host,
                                      const LEX_CSTRING &proxied_user,
                                      bool with_grant, const char *grantor) {
  DBUG_TRACE;
  if (store_pk(table, hostname, user, proxied_host, proxied_user)) return true;
  if (store_with_grant(table, with_grant)) return true;
  if (table->field[MYSQL_PROXIES_PRIV_GRANTOR]->store(grantor, strlen(grantor),
                                                      system_charset_info))
    return true;

  my_timeval tm = table->in_use->query_start_timeval_trunc(0);
  table->field[MYSQL_PROXIES_PRIV_TIMESTAMP]->store_timestamp(&tm);

  return false;
}

void ACL_PROXY_USER::set_user(MEM_ROOT *mem, const char *user_arg) {
  set_username(const_cast<char **>(&user), user_arg, mem);
}

void ACL_PROXY_USER::set_host(MEM_ROOT *mem, const char *host_arg) {
  set_hostname(&host, host_arg, mem);
}

void ACL_DB::set_user(MEM_ROOT *mem, const char *user_arg) {
  set_username(&user, user_arg, mem);
}

void ACL_DB::set_host(MEM_ROOT *mem, const char *host_arg) {
  set_hostname(&host, host_arg, mem);
}

/**
  Append the authorization id for the user

  @param [in]       thd      The THD to find the SQL mode
  @param [in]       acl_user ACL User to retrieve the user information
  @param [in, out]  str      The string in which authID is suffixed
*/
void append_auth_id(const THD *thd, ACL_USER *acl_user, String *str) {
  assert(thd);
  append_auth_id_string(thd, acl_user->user, acl_user->get_username_length(),
                        acl_user->host.get_host(),
                        acl_user->host.get_host_len(), str);
}

/**
  Append the user\@host to the str.

  @param [in]       thd      The THD to find the SQL mode
  @param [in]       user     Username to append to authID
  @param [in]       user_len Length of Username
  @param [in]       host     hostname to append to authID
  @param [in]       host_len Length of hostname
  @param [in, out]  str      The string in which authID is suffixed
*/
void append_auth_id_string(const THD *thd, const char *user, size_t user_len,
                           const char *host, size_t host_len, String *str) {
  assert(thd);
  append_identifier(thd, str, user, user_len);
  str->append(STRING_WITH_LEN("@"));
  append_identifier(thd, str, host, host_len);
}

/**
  Performs wildcard matching, aka globbing, on the input string with
  the given wildcard pattern, and the specified wildcard characters.
  This method does case insensitive comparisons.

  @param[in] cs character set of the input string and wildcard pattern
  @param[in] str input which should be matched against pattern
  @param[in] str_len length of the input string
  @param[in] wildstr pattern with wildcards
  @param[in] wildstr_len length of the wildcards pattern

  @return 0 if input string match with the pattern
  @return 1 otherwise
*/
int wild_case_compare(CHARSET_INFO *cs, const char *str, size_t str_len,
                      const char *wildstr, size_t wildstr_len) {
  int flag;
  DBUG_TRACE;
  DBUG_PRINT("enter", ("str: '%s'  wildstr: '%s'", str, wildstr));
  const char *wildstr_end = wildstr + wildstr_len;
  const char *str_end = str + str_len;

  /*
    Empty string matches only if there is only a wild_many(%) char
    in the string to be matched with.
  */
  if (str_len == 0) {
    bool ret_value = true;
    if (wildstr_len == 1) {
      ret_value = !(*wildstr == wild_many);
    }
    return ret_value;
  }

  while (wildstr != wildstr_end && str != str_end) {
    while (wildstr != wildstr_end && *wildstr != wild_many &&
           *wildstr != wild_one && str != str_end) {
      if (*wildstr == wild_prefix && wildstr[1]) wildstr++;
      if (my_toupper(cs, *wildstr++) != my_toupper(cs, *str++)) return 1;
    }
    if (wildstr == wildstr_end) {
      return str != str_end;
    }
    if (str == str_end) {
      if (*wildstr == '%' && wildstr + 1 == wildstr_end)
        return 0; /* % match empty string */
      return (wildstr != wildstr_end);
    }
    if (*wildstr++ == wild_one) {
      ++str;
      if (str == str_end) /* One char; skip */
      {
        return wildstr != wildstr_end;
      }
    } else {                                 /* Found wild_many */
      if (wildstr == wildstr_end) return 0;  // empty matches wild_many
      flag = (*wildstr != wild_many && *wildstr != wild_one);
      do {
        if (flag) {
          char cmp;
          if ((cmp = *wildstr) == wild_prefix && wildstr[1]) cmp = wildstr[1];
          cmp = my_toupper(cs, cmp);
          while (str != str_end && my_toupper(cs, *str) != cmp) str++;
          if (str == str_end) return 1;
        }
        if (wild_case_compare(cs, str, str_end - str, wildstr,
                              wildstr_end - wildstr) == 0) {
          return 0;
        }
        ++str;
      } while (str != str_end);
      return 1;
    }
  }
  return str != str_end;
}

int wild_case_compare(CHARSET_INFO *cs, const char *str, const char *wildstr) {
  return wild_case_compare(cs, str, strlen(str), wildstr, strlen(wildstr));
}

/*
  Return a number which, if sorted 'desc', puts strings in this order:
    no wildcards
    strings containing wildcards and non-wildcard characters
    single muilt-wildcard character('%')
    empty string
*/

ulong get_sort(uint count, ...) {
  va_list args;
  va_start(args, count);
  ulong sort = 0;

  /* Should not use this function with more than 4 arguments for compare. */
  assert(count <= 4);

  while (count--) {
    char *start, *str = va_arg(args, char *);
    uint chars = 0;
    uint wild_pos = 0;

    /*
      wild_pos
        0                            if string is empty
        1                            if string is a single muilt-wildcard
                                     character('%')
        first wildcard position + 1  if string containing wildcards and
                                     non-wildcard characters
    */

    if ((start = str)) {
      for (; *str; str++) {
        if (*str == wild_prefix && str[1])
          str++;
        else if (*str == wild_many || *str == wild_one) {
          wild_pos = (uint)(str - start) + 1;
          if (!(wild_pos == 1 && *str == wild_many && *(++str) == '\0'))
            wild_pos++;
          break;
        }
        chars = 128;  // Marker that chars existed
      }
    }
    sort = (sort << 8) + (wild_pos ? min(wild_pos, 127U) : chars);
  }
  va_end(args);
  return sort;
}

/**
  Check if the given host name needs to be resolved or not.
  Host name has to be resolved if it actually contains *name*.

  For example:
    192.168.1.1               --> false
    192.168.1.0/255.255.255.0 --> false
    %                         --> false
    192.168.1.%               --> false
    AB%                       --> false

    AAAAFFFF                  --> true (Hostname)
    AAAA:FFFF:1234:5678       --> false
    ::1                       --> false

  This function does not check if the given string is a valid host name or
  not. It assumes that the argument is a valid host name.

  @param hostname   the string to check.

  @return a flag telling if the argument needs to be resolved or not.
  @retval true the argument is a host name and needs to be resolved.
  @retval false the argument is either an IP address, or a patter and
          should not be resolved.
*/

bool hostname_requires_resolving(const char *hostname) {
  /* called only for --skip-name-resolve */
  assert(specialflag & SPECIAL_NO_RESOLVE);

  if (!hostname) return false;

  /*
    If the string contains any of {':', '%', '_', '/'}, it is definitely
    not a host name:
      - ':' means that the string is an IPv6 address;
      - '%' or '_' means that the string is a pattern;
      - '/' means that the string is an IPv4 network address;
  */

  for (const char *p = hostname; *p; ++p) {
    switch (*p) {
      case ':':
      case '%':
      case '_':
      case '/':
        return false;
    }
  }

  /*
    Now we have to tell a host name (ab.cd, 12.ab) from an IPv4 address
    (12.34.56.78). The assumption is that if the string contains only
    digits and dots, it is an IPv4 address. Otherwise -- a host name.
  */

  for (const char *p = hostname; *p; ++p) {
    if (*p != '.' && !my_isdigit(&my_charset_latin1, *p))
      return true; /* a "letter" has been found. */
  }

  return false; /* all characters are either dots or digits. */
}

GRANT_COLUMN::GRANT_COLUMN(String &c, Access_bitmask y)
    : rights(y), column(c.ptr(), c.length()) {}

void GRANT_NAME::set_user_details(const char *h, const char *d, const char *u,
                                  const char *t, bool is_routine) {
  /* Host given by user */
  set_hostname(&host, h, &memex);
  if (db != d) {
    db = strdup_root(&memex, d);
    if (lower_case_table_names) my_casedn_str(files_charset_info, db);
  }
  user = strdup_root(&memex, u);
  sort = get_sort(3, host.get_host(), db, user);
  if (tname != t) {
    tname = strdup_root(&memex, t);
    if (lower_case_table_names || is_routine)
      my_casedn_str(files_charset_info, tname);
  }

  hash_key = user;
  hash_key.push_back('\0');
  hash_key.append(db);
  hash_key.push_back('\0');
  hash_key.append(tname);
  hash_key.push_back('\0');
}

GRANT_NAME::GRANT_NAME(const char *h, const char *d, const char *u,
                       const char *t, Access_bitmask p, bool is_routine)
    : db(nullptr), tname(nullptr), privs(p) {
  set_user_details(h, d, u, t, is_routine);
}

GRANT_TABLE::GRANT_TABLE(const char *h, const char *d, const char *u,
                         const char *t, Access_bitmask p, Access_bitmask c)
    : GRANT_NAME(h, d, u, t, p, false),
      cols(c),
      hash_columns(system_charset_info, key_memory_acl_memex) {}

GRANT_NAME::GRANT_NAME(TABLE *form, bool is_routine) {
  host.update_hostname(get_field(&memex, form->field[0]));
  db = get_field(&memex, form->field[1]);
  user = get_field(&memex, form->field[2]);
  if (!user) user = "";
  sort = get_sort(3, host.get_host(), db, user);
  tname = get_field(&memex, form->field[3]);
  if (!db || !tname) {
    /* Wrong table row; Ignore it */
    privs = 0;
    return; /* purecov: inspected */
  }
  if (lower_case_table_names) {
    my_casedn_str(files_charset_info, db);
  }
  if (lower_case_table_names || is_routine) {
    my_casedn_str(files_charset_info, tname);
  }

  hash_key = user;
  hash_key.push_back('\0');
  hash_key.append(db);
  hash_key.push_back('\0');
  hash_key.append(tname);
  hash_key.push_back('\0');

  if (form->field[MYSQL_TABLES_PRIV_FIELD_TABLE_PRIV]) {
    privs = (Access_bitmask)form->field[MYSQL_TABLES_PRIV_FIELD_TABLE_PRIV]
                ->val_int();
    privs = fix_rights_for_table(privs);
  }
}

GRANT_TABLE::GRANT_TABLE(TABLE *form)
    : GRANT_NAME(form, false),
      hash_columns(system_charset_info, key_memory_acl_memex) {
  if (!db || !tname) {
    /* Wrong table row; Ignore it */
    cols = 0;
    return;
  }

  if (form->field[MYSQL_TABLES_PRIV_FIELD_COLUMN_PRIV]) {
    cols = (Access_bitmask)form->field[MYSQL_TABLES_PRIV_FIELD_COLUMN_PRIV]
               ->val_int();
    cols = fix_rights_for_column(cols);
  } else
    cols = 0;
}

GRANT_TABLE::~GRANT_TABLE() = default;

bool GRANT_TABLE::init(TABLE *col_privs) {
  int error;

  if (cols) {
    uchar key[MAX_KEY_LENGTH];
    uint key_prefix_len;

    KEY_PART_INFO *key_part = col_privs->key_info->key_part;
    col_privs->field[0]->store(host.get_host(),
                               host.get_host() ? host.get_host_len() : 0,
                               system_charset_info);
    col_privs->field[1]->store(db, strlen(db), system_charset_info);
    col_privs->field[2]->store(user, strlen(user), system_charset_info);
    col_privs->field[3]->store(tname, strlen(tname), system_charset_info);

    key_prefix_len = (key_part[0].store_length + key_part[1].store_length +
                      key_part[2].store_length + key_part[3].store_length);
    key_copy(key, col_privs->record[0], col_privs->key_info, key_prefix_len);
    col_privs->field[4]->store("", 0, &my_charset_latin1);

    error = col_privs->file->ha_index_init(0, true);
    DBUG_EXECUTE_IF("wl7158_grant_table_1", col_privs->file->ha_index_end();
                    error = HA_ERR_LOCK_DEADLOCK;);
    if (error) {
      acl_print_ha_error(error);
      return true;
    }

    error =
        col_privs->file->ha_index_read_map(col_privs->record[0], (uchar *)key,
                                           (key_part_map)15, HA_READ_KEY_EXACT);
    assert(col_privs->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
           error != HA_ERR_LOCK_DEADLOCK);
    assert(col_privs->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
           error != HA_ERR_LOCK_WAIT_TIMEOUT);
    DBUG_EXECUTE_IF("wl7158_grant_table_2", error = HA_ERR_LOCK_DEADLOCK;);
    if (error) {
      bool ret = false;
      cols = 0;
      if (error != HA_ERR_KEY_NOT_FOUND && error != HA_ERR_END_OF_FILE) {
        acl_print_ha_error(error);
        ret = true;
      }
      col_privs->file->ha_index_end();
      return ret;
    }

    do {
      String *res, column_name;
      GRANT_COLUMN *mem_check;
      /* As column name is a string, we don't have to supply a buffer */
      res = col_privs->field[4]->val_str(&column_name);
      const Access_bitmask priv =
          (Access_bitmask)col_privs->field[6]->val_int();
      DBUG_EXECUTE_IF("mysql_grant_table_init_out_of_memory",
                      DBUG_SET("+d,simulate_out_of_memory"););
      if (!(mem_check = new (*THR_MALLOC)
                GRANT_COLUMN(*res, fix_rights_for_column(priv)))) {
        /* Don't use this entry */
        col_privs->file->ha_index_end();

        DBUG_EXECUTE_IF("mysql_grant_table_init_out_of_memory",
                        DBUG_SET("-d,simulate_out_of_memory"););
        return true;
      }
      hash_columns.emplace(mem_check->column,
                           unique_ptr_destroy_only<GRANT_COLUMN>(mem_check));

      error = col_privs->file->ha_index_next(col_privs->record[0]);
      assert(col_privs->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_DEADLOCK);
      assert(col_privs->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_WAIT_TIMEOUT);
      DBUG_EXECUTE_IF("wl7158_grant_table_3", error = HA_ERR_LOCK_DEADLOCK;);
      if (error && error != HA_ERR_END_OF_FILE) {
        acl_print_ha_error(error);
        col_privs->file->ha_index_end();
        return true;
      }
    } while (!error && !key_cmp_if_same(col_privs, key, 0, key_prefix_len));
    col_privs->file->ha_index_end();
  }

  return false;
}

/**
  Build the lists of ACL_USERs which share name or have no name

  All accounts with same name will be chained so that they can be
  retrieved by a single lookup. These entries are sorted using
  ACL_compare to make sure that most specific account is picked up
  first. Anonymous user is added to each chain.
*/

void rebuild_cached_acl_users_for_name(void) {
  DBUG_TRACE;
  DBUG_PRINT("enter", ("acl_users size: %zu", acl_users->size()));

  assert(!current_thd || assert_acl_cache_write_lock(current_thd));

  if (name_to_userlist) {
    name_to_userlist->clear();
  } else {
    size_t size = sizeof(Name_to_userlist);
    myf_t flags = MYF(MY_WME | ME_FATALERROR);
    void *bytes = my_malloc(key_memory_acl_cache, size, flags);
    name_to_userlist = new (bytes) Name_to_userlist();
  }

  std::list<ACL_USER *> anons;

  /* first build each named list */
  for (ACL_USER *acl_user = acl_users->begin(); acl_user != acl_users->end();
       ++acl_user) {
    std::string name = acl_user->user ? acl_user->user : "";
    (*name_to_userlist)[name].push_back(acl_user);

    /* keep track of anonymous acl_users */
    if (!name.compare("")) anons.push_back(acl_user);
  }

  /* add the anonymous acl_users to each non-anon list */
  for (auto it = name_to_userlist->begin(); it != name_to_userlist->end();
       ++it) {
    std::string name = it->first;
    if (!name.compare("")) continue;

    auto *list = &it->second;
    for (auto it2 = anons.begin(); it2 != anons.end(); ++it2) {
      list->push_back(*it2);
    }

    list->sort(ACL_USER_compare());
  }
}

/**
  Fetch the list of ACL_USERs which share name or have no name

  @param [in] name User entry to be searched

  @returns List of users that share same name
*/

Acl_user_ptr_list *cached_acl_users_for_name(const char *name) {
  DBUG_TRACE;
  DBUG_PRINT("enter", ("name: '%s'", name));

  assert(!current_thd || assert_acl_cache_read_lock(current_thd));

  std::string user_name = name ? name : "";

  auto it = name_to_userlist->find(user_name);
  if (it != name_to_userlist->end()) return &it->second;

  it = name_to_userlist->find("");
  if (it != name_to_userlist->end()) return &it->second;

  return nullptr;
}

/*
  Find first entry that matches the current user
*/

ACL_USER *find_acl_user(const char *host, const char *user, bool exact) {
  DBUG_TRACE;
  DBUG_PRINT("enter", ("host: '%s'  user: '%s'", host, user));

  assert(assert_acl_cache_read_lock(current_thd));

  if (likely(acl_users)) {
    Acl_user_ptr_list *list = cached_acl_users_for_name(user);
    if (!list) {
      return nullptr;
    }

    for (auto it = list->begin(); it != list->end(); ++it) {
      ACL_USER *acl_user = (*it);
      DBUG_PRINT("info",
                 ("strcmp('%s','%s'), compare_hostname('%s','%s'),", user,
                  acl_user->user ? acl_user->user : "", host,
                  acl_user->host.get_host() ? acl_user->host.get_host() : ""));
      if (acl_user->user || (user && !user[0])) {
        if (exact ? !my_strcasecmp(system_charset_info, host ? host : "",
                                   acl_user->host.get_host()
                                       ? acl_user->host.get_host()
                                       : "")
                  : acl_user->host.compare_hostname(host, host)) {
          return acl_user;
        }
      }
    }
  }
  return nullptr;
}

/*
  Find user in ACL

  SYNOPSIS
    is_acl_user()
    thd                  Handle to THD
    host                 host name
    user                 user name

  RETURN
   false  user not fond
   true   there are such user
*/

bool is_acl_user(THD *thd, const char *host, const char *user) {
  bool res = true;

  /* --skip-grants */
  if (!initialized) return res;

  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::READ_MODE);
  if (!acl_cache_lock.lock(false)) return res;

  res = find_acl_user(host, user, true) != nullptr;
  return res;
}

/**
  Validate if a user can proxy as another user

  @param user              the logged in user (proxy user)
  @param host              the hostname part of the logged in userid
  @param ip                the ip of the logged in userid
  @param authenticated_as  the effective user a plugin is trying to
                           impersonate as (proxied user)
  @param [out] proxy_used  True if a proxy is found
  @return                  proxy user definition
    @retval NULL           proxy user definition not found or not applicable
    @retval non-null       the proxy user data
*/

ACL_PROXY_USER *acl_find_proxy_user(const char *user, const char *host,
                                    const char *ip, char *authenticated_as,
                                    bool *proxy_used) {
  /* if the proxied and proxy user are the same return OK */
  DBUG_TRACE;
  DBUG_PRINT("info", ("user=%s host=%s ip=%s authenticated_as=%s", user, host,
                      ip, authenticated_as));

  if (!strcmp(authenticated_as, user)) {
    DBUG_PRINT("info", ("user is the same as authenticated_as"));
    return nullptr;
  }

  bool find_any = check_proxy_users && !*authenticated_as;

  if (!find_any) *proxy_used = true;
  for (ACL_PROXY_USER *proxy = acl_proxy_users->begin();
       proxy != acl_proxy_users->end(); ++proxy) {
    if (proxy->matches(host, user, ip, authenticated_as, find_any)) {
      DBUG_PRINT("info", ("proxy matched=%s@%s", proxy->get_proxied_user(),
                          proxy->get_proxied_host()));
      if (!find_any) {
        DBUG_PRINT(
            "info",
            ("returning specific match as authenticated_as was specified"));
        *proxy_used = true;
        return proxy;
      } else {
        // we never use anonymous users when mapping
        // proxy users for internal plugins:
        if (strcmp(proxy->get_proxied_user() ? proxy->get_proxied_user() : "",
                   "")) {
          if (find_acl_user(proxy->get_proxied_host(),
                            proxy->get_proxied_user(), true)) {
            DBUG_PRINT("info",
                       ("setting proxy_used to true, as \
              find_all search matched real user=%s host=%s",
                        proxy->get_proxied_user(), proxy->get_proxied_host()));
            *proxy_used = true;
            strcpy(authenticated_as, proxy->get_proxied_user());
          } else {
            DBUG_PRINT("info", ("skipping match because ACL user \
              does not exist, looking for next match to map"));
          }
          if (*proxy_used) {
            DBUG_PRINT("info", ("returning matching user"));
            return proxy;
          }
        }
      }
    }
  }
  DBUG_PRINT("info", ("No matching users found, returning null"));
  return nullptr;
}

void clear_and_init_db_cache() { db_cache.clear(); }

/**
  Insert a new entry in db_cache

  @param [in] thd    Handle to THD object
  @param [in] entry  Entry to be inserted in db_cache
*/

static void insert_entry_in_db_cache(THD *thd, acl_entry *entry) {
  DBUG_TRACE;
  /* Either have WRITE lock or none at all */
  assert(assert_acl_cache_write_lock(thd) || !assert_acl_cache_read_lock(thd));

  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::WRITE_MODE);

  /*
    In following cases release memory and return
    1. Could not lock cache : This is ok because db_cache
       second level cache anyways.
    2. Someone already inserted a similar entry.
  */
  unique_ptr_my_free<acl_entry> entry_ptr(entry);
  if (!acl_cache_lock.lock(false)) return;
  db_cache.emplace(std::string(entry->key, entry->length),
                   std::move(entry_ptr));
}

/**
  Get privilege for a host, user, and db
  combination.

  @note db_cache is not used if db_is_pattern
  is set.

  @param thd   Thread handler
  @param host  Host name
  @param ip    Ip
  @param user  user name
  @param db    We look for the ACL of this database
  @param db_is_pattern true if @p db can be considered a pattern or false if not

  @return Database ACL
*/

Access_bitmask acl_get(THD *thd, const char *host, const char *ip,
                       const char *user, const char *db, bool db_is_pattern) {
  Access_bitmask host_access = ~(Access_bitmask)0, db_access = 0;
  size_t key_length, copy_length;
  char key[ACL_KEY_LENGTH], *tmp_db, *end;
  acl_entry *entry;
  DBUG_TRACE;

  copy_length =
      (strlen(ip ? ip : "") + strlen(user ? user : "") + strlen(db ? db : "")) +
      2; /* Added 2 at the end to avoid
            buffer overflow at strmov()*/
  /*
    Make sure that my_stpcpy() operations do not result in buffer overflow.
  */
  if (copy_length >= ACL_KEY_LENGTH) return 0;
  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::READ_MODE);
  if (!acl_cache_lock.lock(false)) return db_access;

  end = my_stpcpy(
      (tmp_db = my_stpcpy(my_stpcpy(key, ip ? ip : "") + 1, user) + 1), db);
  if (lower_case_table_names) {
    my_casedn_str(files_charset_info, tmp_db);
    db = tmp_db;
  }
  key_length = (size_t)(end - key);
  if (!db_is_pattern) {
    const auto it = db_cache.find(std::string(key, key_length));
    if (it != db_cache.end()) {
      db_access = it->second->access;
      DBUG_PRINT("exit", ("access: 0x%" PRIx32, db_access));
      return db_access;
    }
  }

  /*
    Check if there are some access rights for database and user
  */
  for (ACL_DB *acl_db = acl_dbs->begin(); acl_db != acl_dbs->end(); ++acl_db) {
    if (!acl_db->user || !strcmp(user, acl_db->user)) {
      if (acl_db->host.compare_hostname(host, ip)) {
        /*
          Do the usual string comparison if partial_revokes is ON,
          otherwise do the wildcard grant comparison
        */
        if (!acl_db->db ||
            (db &&
             (mysqld_partial_revokes()
                  ? (!strcmp(db, acl_db->db))
                  : (!wild_compare(db, strlen(db), acl_db->db,
                                   strlen(acl_db->db), db_is_pattern))))) {
          db_access = acl_db->access;
          if (acl_db->host.get_host()) goto exit;  // Fully specified. Take it
          break;                                   /* purecov: tested */
        }
      }
    }
  }
  if (!db_access) goto exit;  // Can't be better

exit:
  /* Save entry in cache for quick retrieval */
  if (!db_is_pattern &&
      (entry = (acl_entry *)my_malloc(
           key_memory_acl_cache, sizeof(acl_entry) + key_length, MYF(0)))) {
    entry->access = (db_access & host_access);
    entry->length = key_length;
    memcpy((uchar *)entry->key, key, key_length);
    acl_cache_lock.unlock();
    insert_entry_in_db_cache(thd, entry);
  }
  DBUG_PRINT("exit", ("access: 0x%" PRIx32, db_access & host_access));
  return db_access & host_access;
}

/*
  Check if there are any possible matching entries for this host

  NOTES
    All host names without wild cards are stored in a hash table,
    entries with wildcards are stored in a dynamic array
*/

static void init_check_host(void) {
  DBUG_TRACE;
  if (acl_wild_hosts != nullptr)
    acl_wild_hosts->clear();
  else
    acl_wild_hosts = new Prealloced_array<ACL_HOST_AND_IP, ACL_PREALLOC_SIZE>(
        key_memory_acl_mem);

  size_t acl_users_size = acl_users ? acl_users->size() : 0;

  acl_check_hosts = new collation_unordered_map<std::string, ACL_USER *>(
      system_charset_info, key_memory_acl_mem);
  if (acl_users_size && !allow_all_hosts) {
    for (ACL_USER *acl_user = acl_users->begin(); acl_user != acl_users->end();
         ++acl_user) {
      if (acl_user->host.get_host()) {
        if (acl_user->host.has_wildcard()) {  // Has wildcard
          ACL_HOST_AND_IP *acl = nullptr;
          for (acl = acl_wild_hosts->begin(); acl != acl_wild_hosts->end();
               ++acl) {  // Check if host already exists
            if (!my_strcasecmp(system_charset_info, acl_user->host.get_host(),
                               acl->get_host()))
              break;  // already stored
          }
          if (acl == acl_wild_hosts->end())  // If new
            acl_wild_hosts->push_back(acl_user->host);
        } else {
          // Will be ignored if there's already an entry.
          acl_check_hosts->emplace(acl_user->host.get_host(), acl_user);
        }
      }
    }
  }
  acl_wild_hosts->shrink_to_fit();
}

/*
  Rebuild lists used for checking of allowed hosts

  We need to rebuild 'acl_check_hosts' and 'acl_wild_hosts' after adding,
  dropping or renaming user, since they contain pointers to elements of
  'acl_user' array, which are invalidated by drop operation, and use
  ACL_USER::host::hostname as a key, which is changed by rename.
*/
void rebuild_check_host(void) {
  delete acl_wild_hosts;
  acl_wild_hosts = nullptr;
  delete acl_check_hosts;
  acl_check_hosts = nullptr;
  init_check_host();
}

/*
  Gets user credentials without authentication and resource limit checks. This
  function is used to initialized a new Security_context. It's a terrible
  anti-pattern that needs to go.

  SYNOPSIS
    acl_getroot()
      thd                Handle to THD
      sctx               Context which should be initialized
      user               user name
      host               host name
      ip                 IP
      db                 current data base name

  RETURN
    false  OK
    true   Error
*/

bool acl_getroot(THD *thd, Security_context *sctx, const char *user,
                 const char *host, const char *ip, const char *db) {
  int res = 1;
  ACL_USER *acl_user = nullptr;
  DBUG_TRACE;

  DBUG_PRINT("enter", ("Host: '%s', Ip: '%s', User: '%s', db: '%s'",
                       (host ? host : "(NULL)"), (ip ? ip : "(NULL)"), user,
                       (db ? db : "(NULL)")));
  sctx->set_user_ptr(user, user ? strlen(user) : 0);
  sctx->set_host_ptr(host, host ? strlen(host) : 0);
  sctx->set_ip_ptr(ip, ip ? strlen(ip) : 0);
  sctx->set_host_or_ip_ptr();

  if (!initialized) {
    /*
      here if mysqld's been started with --skip-grant-tables option.
    */
    sctx->skip_grants();
    return false;
  }

  sctx->set_master_access(0);
  sctx->cache_current_db_access(0);
  sctx->assign_priv_user("", 0);
  sctx->assign_priv_host("", 0);

  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::READ_MODE);
  if (!acl_cache_lock.lock(false)) return true;

  /*
     Find acl entry in user database.
     This is specially tailored to suit the check we do for CALL of
     a stored procedure; user is set to what is actually a
     priv_user, which can be ''.
  */
  for (ACL_USER *acl_user_tmp = acl_users->begin();
       acl_user_tmp != acl_users->end(); ++acl_user_tmp) {
    if ((!acl_user_tmp->user && !user[0]) ||
        (acl_user_tmp->user && strcmp(user, acl_user_tmp->user) == 0)) {
      if (acl_user_tmp->host.compare_hostname(host, ip)) {
        acl_user = acl_user_tmp;
        res = 0;
        break;
      }
    }
  }

  if (acl_user) {
    sctx->clear_active_roles();
    List_of_auth_id_refs default_roles;
    Auth_id_ref authid = create_authid_from(acl_user);
    /* Needs Acl_cache_lock_guard in read mode */
    get_default_roles(authid, default_roles);
    List_of_auth_id_refs::iterator it = default_roles.begin();
    for (; it != default_roles.end(); ++it) {
      if (sctx->activate_role(it->first, it->second)) {
        sctx->clear_active_roles();
        break;
      }
    }

    if (sctx->get_active_roles()->size() == 0) {
      for (ACL_DB *acl_db = acl_dbs->begin(); acl_db != acl_dbs->end();
           ++acl_db) {
        if (!acl_db->user || (user && user[0] && !strcmp(user, acl_db->user))) {
          if (acl_db->host.compare_hostname(host, ip)) {
            /*
              Do the usual string comparison if partial_revokes is ON,
              otherwise do the wildcard grant comparison
            */
            if (!acl_db->db ||
                (db && (mysqld_partial_revokes()
                            ? (!strcmp(db, acl_db->db))
                            : (!wild_compare(db, strlen(db), acl_db->db,
                                             strlen(acl_db->db), false))))) {
              sctx->cache_current_db_access(acl_db->access);
              break;
            }
          }
        }  // end if
      }    // end for
      sctx->set_master_access(acl_user->access,
                              acl_restrictions->find_restrictions(acl_user));
    }  // end if
    sctx->assign_priv_user(user, user ? strlen(user) : 0);
    sctx->assign_priv_host(
        acl_user->host.get_host(),
        acl_user->host.get_host() ? strlen(acl_user->host.get_host()) : 0);

    sctx->set_password_expired(acl_user->password_expired);
    sctx->lock_account(acl_user->account_locked);
  }  // end if

  if (acl_user && sctx->get_active_roles()->size() > 0) {
    sctx->checkout_access_maps();
    const Access_bitmask db_acl = db ? sctx->db_acl({db, strlen(db)}) : 0;
    sctx->cache_current_db_access(db_acl);
  }
  return res;
}

/**
  Convert scrambled password to binary form, according to scramble type,
  Binary form is stored in user.salt.

  @param acl_user The object where to store the salt

  Despite the name of the function it is used when loading ACLs from disk
  to store the password hash in the ACL_USER object.
  Note that it works only for native and "old" mysql authentication built-in
  plugins.

  Assumption : user's authentication plugin information is available.

  @return Password hash validation
    @retval false Hash is of suitable length
    @retval true Hash is of wrong length or format
*/

bool set_user_salt(ACL_USER *acl_user) {
  bool result = false;
  plugin_ref plugin = nullptr;

  plugin = my_plugin_lock_by_name(nullptr, acl_user->plugin,
                                  MYSQL_AUTHENTICATION_PLUGIN);
  if (plugin) {
    st_mysql_auth *auth = (st_mysql_auth *)plugin_decl(plugin)->info;

    for (int i = 0; i < NUM_CREDENTIALS && !result; ++i) {
      result = auth->set_salt(acl_user->credentials[i].m_auth_string.str,
                              acl_user->credentials[i].m_auth_string.length,
                              acl_user->credentials[i].m_salt,
                              &acl_user->credentials[i].m_salt_len);
    }
    plugin_unlock(nullptr, plugin);
  }
  return result;
}

/**
  Iterate over the user records and check for irregularities.
  Currently this includes :
   - checking if the plugin referenced is present.
   - if there's sha256 users and there's neither SSL nor RSA configured
*/
static void validate_user_plugin_records() {
  DBUG_TRACE;
  if (!validate_user_plugins) return;

  MUTEX_LOCK(plugin_lock, &LOCK_plugin);
  for (ACL_USER *acl_user = acl_users->begin(); acl_user != acl_users->end();
       ++acl_user) {
    struct st_plugin_int *plugin;

    if (acl_user->plugin.length) {
      /* rule 1 : plugin does exit */
      if (!auth_plugin_is_built_in(acl_user->plugin.str)) {
        plugin =
            plugin_find_by_type(acl_user->plugin, MYSQL_AUTHENTICATION_PLUGIN);

        if (!plugin) {
          LogErr(WARNING_LEVEL, ER_AUTHCACHE_PLUGIN_MISSING,
                 (int)acl_user->plugin.length, acl_user->plugin.str,
                 acl_user->user,
                 static_cast<int>(acl_user->host.get_host_len()),
                 acl_user->host.get_host());
        }
      }
      if (Cached_authentication_plugins::compare_plugin(PLUGIN_SHA256_PASSWORD,
                                                        acl_user->plugin) &&
          sha256_rsa_auth_status() && !have_ssl()) {
        LogErr(WARNING_LEVEL, ER_AUTHCACHE_PLUGIN_CONFIG, acl_user->plugin.str,
               acl_user->user, static_cast<int>(acl_user->host.get_host_len()),
               acl_user->host.get_host(), "but neither SSL nor RSA keys are");
      }
    }
  }
}

/**
  Audit notification for flush

  @param [in] thd Handle to THD
*/

void notify_flush_event(THD *thd) {
  mysql_audit_notify(thd, AUDIT_EVENT(MYSQL_AUDIT_AUTHENTICATION_FLUSH), 0,
                     nullptr, nullptr, nullptr, false, nullptr, nullptr);
}

/**
  Initialize roles structures from role tables handle.

  This function is called by acl_reload and may fail to
  initialize role structures if handle to role_edges and/or
  default_roles are NUL

  @param [in] thd      Handle to THD object
  @param [in] tablelst Handle to Roles tables

  @returns status of cache update
    @retval false Success
    @retval true failure
*/
static bool reload_roles_cache(THD *thd, Table_ref *tablelst) {
  DBUG_TRACE;
  assert(tablelst);
  sql_mode_t old_sql_mode = thd->variables.sql_mode;
  thd->variables.sql_mode &= ~MODE_PAD_CHAR_TO_FULL_LENGTH;

  /*
    Attempt to reload the role cache only if the role_edges and
    default_roles tables exist.
  */
  if ((tablelst[0].table) && (tablelst[1].table) &&
      populate_roles_caches(thd, tablelst)) {
    thd->variables.sql_mode = old_sql_mode;
    return true;
  }

  thd->variables.sql_mode = old_sql_mode;
  return false;
}

/*
  Initialize structures responsible for user/db-level privilege checking and
  load privilege information for them from tables in the 'mysql' database.

  SYNOPSIS
    acl_init()
      dont_read_acl_tables  true if we want to skip loading data from
                            privilege tables and disable privilege checking.

  NOTES
    This function is mostly responsible for preparatory steps, main work
    on initialization and grants loading is done in acl_reload().

  RETURN VALUES
    0   ok
    1   Could not initialize grant's
*/

bool acl_init(bool dont_read_acl_tables) {
  THD *thd;
  bool return_val;
  DBUG_TRACE;

  init_acl_cache();

  acl_cache_initialized = true;

  /*
    cache built-in authentication plugins,
    to avoid hash searches and a global mutex lock on every connect
  */
  g_cached_authentication_plugins = new Cached_authentication_plugins();
  unknown_accounts = new Map_with_rw_lock<Auth_id, uint>(0);
  if (!g_cached_authentication_plugins->is_valid()) return true;

  if (dont_read_acl_tables) {
    return false; /* purecov: tested */
  }

  /*
    To be able to run this from boot, we allocate a temporary THD
  */
  if (!(thd = new THD)) return true; /* purecov: inspected */
  thd->thread_stack = (char *)&thd;
  thd->store_globals();

  /*
    Check storage engine type for every ACL table and output warning
    message in case it's different from supported one (InnoDB). We
    still allow server to start-up if tables are in different SE
    since this is necessary to be able to perform privilege table
    upgrade without extra server restarts. Account management
    statements do their own checks and refuse to operate if privilege
    tables are using unsupported SE.
  */
  return_val = check_engine_type_for_acl_table(thd, false);

  /*
    Check all the ACL tables are intact and output warning message in
    case any of the ACL tables are corrupted.
  */
  check_acl_tables_intact(thd, false);

  /*
    It is safe to call acl_reload() since acl_* arrays and hashes which
    will be freed there are global static objects and thus are initialized
    by zeros at startup.
  */
  return_val |= acl_reload(thd, false, false, nullptr);
  notify_flush_event(thd);
  /*
    Turn ON the system variable '@@partial_revokes' during server
    start in case there exist at least one restrictions instance.
    Don't log warning in case partial_revokes is already ON.
  */
  if (mysqld_partial_revokes() == false &&
      check_and_update_partial_revokes_sysvar(thd)) {
    LogErr(WARNING_LEVEL, ER_TURNING_ON_PARTIAL_REVOKES);
  }
  thd->release_resources();
  delete thd;

  return return_val;
}

void clean_user_cache() {
  if (name_to_userlist) name_to_userlist->clear();
  acl_users->clear();
}

/*
  Initialize structures responsible for user/db-level privilege checking
  and load information about grants from open privilege tables.

  SYNOPSIS
    acl_load()
      thd     Current thread
      tables  List containing open "mysql.host", "mysql.user",
              "mysql.db" and "mysql.proxies_priv", "mysql.global_grants"
              tables in that order.

  RETURN VALUES
    false  Success
    true   Error
*/

static bool acl_load(THD *thd, Table_ref *tables) {
  TABLE *table;
  unique_ptr_destroy_only<RowIterator> iterator;
  bool return_val = true;
  bool check_no_resolve = specialflag & SPECIAL_NO_RESOLVE;
  char tmp_name[NAME_LEN + 1];
  sql_mode_t old_sql_mode = thd->variables.sql_mode;
  DBUG_TRACE;

  DBUG_EXECUTE_IF(
      "wl_9262_set_max_length_hostname",
      thd->security_context()->assign_priv_host("oh_my_gosh_this_is_a_long_"
                                                "hostname_look_at_it_it_has_60"
                                                "_char",
                                                60);
      thd->security_context()->assign_host("oh_my_gosh_this_is_a_long_"
                                           "hostname_look_at_it_it_has_60"
                                           "_char",
                                           60);
      thd->security_context()->set_host_or_ip_ptr(););

  thd->variables.sql_mode &= ~MODE_PAD_CHAR_TO_FULL_LENGTH;

  grant_version++; /* Privileges updated */

  clear_and_init_db_cache();  // Clear locked hostname cache
  init_acl_memory();          // Allocate the memory blocks in the MEM_ROOT

  if (read_user_table(thd, tables[0].table)) goto end;

  /*
    Prepare reading from the mysql.db table
  */
  iterator = init_table_iterator(thd, table = tables[1].table,
                                 /*ignore_not_found_rows=*/false,
                                 /*count_examined_rows=*/false);
  if (iterator == nullptr) goto end;
  table->use_all_columns();
  acl_dbs->clear();
  int read_rec_errcode;
  while (!(read_rec_errcode = iterator->Read())) {
    /* Reading record in mysql.db */
    ACL_DB db;
    db.host.update_hostname(
        get_field(&global_acl_memory, table->field[MYSQL_DB_FIELD_HOST]));
    db.db = get_field(&global_acl_memory, table->field[MYSQL_DB_FIELD_DB]);
    if (!db.db) {
      LogErr(WARNING_LEVEL, ER_AUTHCACHE_DB_IGNORED_EMPTY_NAME);
      continue;
    }
    db.user = get_field(&global_acl_memory, table->field[MYSQL_DB_FIELD_USER]);
    if (check_no_resolve && hostname_requires_resolving(db.host.get_host()) &&
        strcmp(db.host.get_host(), "localhost") != 0) {
      LogErr(WARNING_LEVEL, ER_AUTHCACHE_DB_SKIPPED_NEEDS_RESOLVE, db.db,
             db.user ? db.user : "",
             db.host.get_host() ? db.host.get_host() : "");
    }
    db.access = get_access(table, 3, nullptr);
    db.access = fix_rights_for_db(db.access);
    if (lower_case_table_names) {
      /*
        convert db to lower case and give a warning if the db wasn't
        already in lower case
      */
      (void)my_stpcpy(tmp_name, db.db);
      my_casedn_str(files_charset_info, db.db);
      if (strcmp(db.db, tmp_name) != 0) {
        LogErr(WARNING_LEVEL, ER_AUTHCACHE_DB_ENTRY_LOWERCASED_REVOKE_WILL_FAIL,
               db.db, db.user ? db.user : "",
               db.host.get_host() ? db.host.get_host() : "");
      }
    }
    db.sort = get_sort(3, db.host.get_host(), db.db, db.user);
    if (table->s->fields <= 9) {  // Without grant
      if (db.access & CREATE_ACL)
        db.access |= REFERENCES_ACL | INDEX_ACL | ALTER_ACL;
    }
    acl_dbs->push_back(db);
  }  // END reading records from mysql.db tables

  iterator.reset();
  if (read_rec_errcode > 0) goto end;

  std::sort(acl_dbs->begin(), acl_dbs->end(), ACL_compare());
  acl_dbs->shrink_to_fit();

  /* Prepare to read records from the mysql.proxies_priv table */
  acl_proxy_users->clear();

  if (tables[2].table) {
    iterator = init_table_iterator(thd, table = tables[2].table,
                                   /*ignore_not_found_rows=*/false,
                                   /*count_examined_rows=*/false);
    if (iterator == nullptr) goto end;
    table->use_all_columns();
    while (!(read_rec_errcode = iterator->Read())) {
      /* Reading record in mysql.proxies_priv */
      ACL_PROXY_USER proxy;
      proxy.init(table, &global_acl_memory);
      if (proxy.check_validity(check_no_resolve)) continue;
      if (acl_proxy_users->push_back(proxy)) {
        goto end;
      }
    }  // END reading records from the mysql.proxies_priv table

    iterator.reset();
    if (read_rec_errcode > 0) goto end;

    std::sort(acl_proxy_users->begin(), acl_proxy_users->end(), ACL_compare());
  } else {
    LogErr(WARNING_LEVEL, ER_AUTHCACHE_TABLE_PROXIES_PRIV_MISSING);
  }
  acl_proxy_users->shrink_to_fit();
  validate_user_plugin_records();
  init_check_host();

  /* Load dynamic privileges */
  if (tables[3].table) {
    if (populate_dynamic_privilege_caches(thd, &tables[3])) {
      return_val = true;
      goto end;
    }
  } else {
    LogErr(WARNING_LEVEL, ER_MISSING_GRANT_SYSTEM_TABLE);
  }

  return_val = false;

end:
  thd->variables.sql_mode = old_sql_mode;
  DBUG_EXECUTE_IF("induce_acl_load_failure", return_val = true;);
  return return_val;
}

/**
  Clear second level cache on account names.
*/

void free_name_to_userlist() {
  if (!name_to_userlist) return;

  name_to_userlist->~unordered_map();
  my_free(name_to_userlist);
  name_to_userlist = nullptr;
}

void acl_free(bool end /*= false*/) {
  free_name_to_userlist();
  delete acl_users;
  acl_users = nullptr;
  delete acl_dbs;
  acl_dbs = nullptr;
  delete acl_wild_hosts;
  acl_wild_hosts = nullptr;
  delete acl_proxy_users;
  acl_proxy_users = nullptr;
  delete acl_check_hosts;
  acl_check_hosts = nullptr;
  if (!end)
    clear_and_init_db_cache();
  else {
    shutdown_acl_cache();
    if (acl_cache_initialized == true) {
      db_cache.clear();
      delete g_cached_authentication_plugins;
      g_cached_authentication_plugins = nullptr;
      delete unknown_accounts;
      unknown_accounts = nullptr;
      acl_cache_initialized = false;
    }
  }
  global_acl_memory.Clear();
}

bool check_engine_type_for_acl_table(THD *thd, bool mdl_locked) {
  Table_ref tables[ACL_TABLES::LAST_ENTRY];
  uint flags = mdl_locked
                   ? MYSQL_OPEN_HAS_MDL_LOCK | MYSQL_LOCK_IGNORE_TIMEOUT |
                         MYSQL_OPEN_IGNORE_FLUSH
                   : MYSQL_LOCK_IGNORE_TIMEOUT;

  /*
    Open the following ACL tables to check their consistency.
    Although we don't read here from the tables being opened we still
    request a lock type MDL_SHARED_READ_ONLY for the sake of consistency
    with other code.
  */

  acl_tables_setup_for_read(tables);
  bool result = open_and_lock_tables(thd, tables, flags);

  if (!result) {
    check_engine_type_for_acl_table(tables, false);
    if (!mdl_locked)
      commit_and_close_mysql_tables(thd);
    else
      close_thread_tables(thd);
  }

  return result;
}

/*
  This internal handler implements downgrade from SL_ERROR to SL_WARNING
  for acl_init()/handle_reload_request().
*/
class Acl_ignore_error_handler : public Internal_error_handler {
 public:
  bool handle_condition(THD *, uint sql_errno, const char *,
                        Sql_condition::enum_severity_level *level,
                        const char *) override {
    switch (sql_errno) {
      case ER_CANNOT_LOAD_FROM_TABLE_V2:
      case ER_COL_COUNT_DOESNT_MATCH_CORRUPTED_V2:
      case ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE_V2:
        (*level) = Sql_condition::SL_WARNING;
        break;
      default:
        break;
    }
    return false;
  }
};

/**
  Helper function that checks the sanity of tables object present in
  the Table_ref object. it logs a warning message when a table is
  missing

  @param thd        Handle of current thread.
  @param tables     A valid table list pointer

  @retval
    false       OK.
    true        Error.
*/
bool check_acl_tables_intact(THD *thd, Table_ref *tables) {
  Acl_table_intact table_intact(thd, WARNING_LEVEL);
  bool result_acl = false;

  assert(tables);
  for (auto idx = 0; idx < ACL_TABLES::LAST_ENTRY; idx++) {
    if (tables[idx].table) {
      result_acl |= table_intact.check(tables[idx].table, (ACL_TABLES)idx);
    } else {
      LogErr(WARNING_LEVEL, ER_MISSING_ACL_SYSTEM_TABLE,
             tables[idx].table_name_length, tables[idx].table_name);
      result_acl |= true;
    }
  }
  /* say that we're still gonna give reading a try */
  if (result_acl)
    LogErr(INFORMATION_LEVEL, ER_ACL_WRONG_OR_MISSING_ACL_TABLES_LOG);

  return result_acl;
}

/**
  Opens the ACL tables and checks their sanity. This method reports error
  only if it is unable to open or lock tables. It is called in situations
  when server has to continue even if a corrupt table was found -
  For example - acl_init()

  @param thd        Handle of current thread.
  @param mdl_locked MDL is locked

  @retval
    false       OK.
    true        Unable to open the table(s).
*/
bool check_acl_tables_intact(THD *thd, bool mdl_locked) {
  Table_ref tables[ACL_TABLES::LAST_ENTRY];
  Acl_ignore_error_handler acl_ignore_handler;
  uint flags = mdl_locked
                   ? MYSQL_OPEN_HAS_MDL_LOCK | MYSQL_LOCK_IGNORE_TIMEOUT |
                         MYSQL_OPEN_IGNORE_FLUSH
                   : MYSQL_LOCK_IGNORE_TIMEOUT;

  acl_tables_setup_for_read(tables);
  bool result_acl = open_and_lock_tables(thd, tables, flags);

  thd->push_internal_handler(&acl_ignore_handler);
  if (!result_acl) {
    check_acl_tables_intact(thd, tables);
    if (!mdl_locked)
      commit_and_close_mysql_tables(thd);
    else
      close_thread_tables(thd);
  }
  thd->pop_internal_handler();

  return result_acl;
}

/**
  Small helper function which allows to determine if error which caused
  failure to open and lock privilege tables should not be reported to error
  log (because this is expected or temporary condition).
*/

bool is_expected_or_transient_error(THD *thd) {
  return !thd->get_stmt_da()->is_error() ||  // Interrupted/no error condition.
         thd->get_stmt_da()->mysql_errno() == ER_TABLE_NOT_LOCKED ||
         thd->get_stmt_da()->mysql_errno() == ER_LOCK_DEADLOCK;
}

ACL_temporary_lock_state::ACL_temporary_lock_state(
    const char *host, const char *user, uint remaining_login_attempts,
    long daynr_locked)
    : m_host(host ? host : ""),
      m_user(user ? user : ""),
      m_remaining_login_attempts(remaining_login_attempts),
      m_daynr_locked(daynr_locked) {}

bool ACL_temporary_lock_state::is_modified(ACL_USER *acl_user) {
  assert(acl_user != nullptr);
  assert(assert_acl_cache_write_lock(current_thd));
  return (!acl_user->password_locked_state.is_default() &&
          !acl_user->account_locked);
}

/**
  Enables preserving temporary account locking attributes of a user during ACL
  DDL.

  @ref ACL_temporary_lock_state

  @param [in]  host      Account's hostname.
  @param [in]  user      Account's username.
  @param [out] user_list List in which a user's lock state is preserved.

  @returns find_acl_user(host, user, true)
*/
ACL_USER *ACL_temporary_lock_state::preserve_user_lock_state(
    const char *host, const char *user, Lock_state_list &user_list) {
  assert(assert_acl_cache_write_lock(current_thd));
  ACL_USER *acl_user;
  if ((acl_user = find_acl_user(host, user, true))) {
    if (ACL_temporary_lock_state::is_modified(acl_user)) {
      user_list.emplace_back(
          host, user,
          acl_user->password_locked_state.get_remaining_login_attempts(),
          acl_user->password_locked_state.get_daynr_locked());
    }
  }
  return acl_user;
}

/**
  Enables restoring temporary account locking attributes of a user after ACL
  reload.

  @ref ACL_temporary_lock_state

  @param [in] host       Account's hostname.
  @param [in] user       Account's username.
  @param [in] remaining_login_attempts
                         Account's remaining login attempts.
  @param [in] daynr_locked
                         Account's locked day.
*/
void ACL_temporary_lock_state::restore_user_lock_state(
    const char *host, const char *user, uint remaining_login_attempts,
    long daynr_locked) {
  assert(assert_acl_cache_write_lock(current_thd));
  ACL_USER *acl_user;
  if ((acl_user = find_acl_user(host, user, true))) {
    acl_user->password_locked_state.set_temporary_lock_state_parameters(
        remaining_login_attempts, daynr_locked);
  }
}

/**
  Enables restoring temporary account locking attributes of all users after ACL
  reload.

  @ref ACL_temporary_lock_state

  @param [in] old_acl_users    List of users in the old ACL cache.
  @param [in] modified_user_lock_state_list
                               List of users whose temporary account
                               locking attributes are likely modified.
*/
void ACL_temporary_lock_state::restore_temporary_account_locking(
    Prealloced_array<ACL_USER, ACL_PREALLOC_SIZE> *old_acl_users,
    Lock_state_list *modified_user_lock_state_list) {
  assert(assert_acl_cache_write_lock(current_thd));
  if (old_acl_users) {
    for (ACL_USER *old_acl_user = old_acl_users->begin();
         old_acl_user != old_acl_users->end(); ++old_acl_user) {
      if (ACL_temporary_lock_state::is_modified(old_acl_user)) {
        ACL_temporary_lock_state::restore_user_lock_state(
            old_acl_user->host.get_host(),
            old_acl_user->user ? old_acl_user->user : "",
            old_acl_user->password_locked_state.get_remaining_login_attempts(),
            old_acl_user->password_locked_state.get_daynr_locked());
      }
    }
  }
  if (modified_user_lock_state_list) {
    for (auto element = modified_user_lock_state_list->rbegin();
         element != modified_user_lock_state_list->rend(); ++element) {
      ACL_temporary_lock_state::restore_user_lock_state(
          element->m_host, element->m_user, element->m_remaining_login_attempts,
          element->m_daynr_locked);
    }
  }
}

/*
  Forget current user/db-level privileges and read new privileges
  from the privilege tables.

  SYNOPSIS
    acl_reload()
      thd  Current thread

  NOTE
    All tables of calling thread which were open and locked by LOCK TABLES
    statement will be unlocked and closed.
    This function is also used for initialization of structures responsible
    for user/db-level privilege checking.

  RETURN VALUE
    false  Success
    true   Failure
*/

bool acl_reload(THD *thd, bool mdl_locked,
                bool preserve_temporary_account_locking,
                Lock_state_list *modified_user_lock_state_list) {
  assert(preserve_temporary_account_locking || !modified_user_lock_state_list);
  MEM_ROOT old_mem;
  bool return_val = true;
  uint flags = mdl_locked
                   ? MYSQL_OPEN_HAS_MDL_LOCK | MYSQL_LOCK_IGNORE_TIMEOUT |
                         MYSQL_OPEN_IGNORE_FLUSH
                   : MYSQL_LOCK_IGNORE_TIMEOUT;
  Prealloced_array<ACL_USER, ACL_PREALLOC_SIZE> *old_acl_users = nullptr;
  Prealloced_array<ACL_DB, ACL_PREALLOC_SIZE> *old_acl_dbs = nullptr;
  Prealloced_array<ACL_PROXY_USER, ACL_PREALLOC_SIZE> *old_acl_proxy_users =
      nullptr;
  Granted_roles_graph *old_granted_roles = nullptr;
  Default_roles *old_default_roles = nullptr;
  Role_index_map *old_authid_to_vertex = nullptr;
  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::WRITE_MODE);
  User_to_dynamic_privileges_map *old_dyn_priv_map;
  unique_ptr<Acl_restrictions> old_acl_restrictions = nullptr;
  DBUG_TRACE;

  // Interchange the global role cache ptrs with the local role cache ptrs.
  auto swap_role_cache = [&]() {
    std::swap(old_granted_roles, g_granted_roles);
    std::swap(old_default_roles, g_default_roles);
    std::swap(old_authid_to_vertex, g_authid_to_vertex);
  };
  // Delete the memory pointed by the local role cache ptrs.
  auto delete_old_role_cache = [&]() {
    delete old_granted_roles;
    delete old_default_roles;
    delete old_authid_to_vertex;
  };

  /*
    To avoid deadlocks we should obtain table locks before obtaining
    acl_cache->lock mutex.
  */
  Table_ref tables[6] = {
      Table_ref("mysql", "user", TL_READ, MDL_SHARED_READ_ONLY),
      /*
        For a Table_ref element that is inited with a lock type TL_READ
        the type MDL_SHARED_READ_ONLY of MDL is requested for.
        Acquiring strong MDL lock allows to avoid deadlock and timeout errors
        from SE level.
      */
      Table_ref("mysql", "db", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "proxies_priv", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "global_grants", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "role_edges", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "default_roles", TL_READ, MDL_SHARED_READ_ONLY)};

  tables[0].next_local = tables[0].next_global = tables + 1;
  tables[1].next_local = tables[1].next_global = tables + 2;
  tables[2].next_local = tables[2].next_global = tables + 3;
  tables[3].next_local = tables[3].next_global = tables + 4;
  tables[4].next_local = tables[4].next_global = tables + 5;

  tables[0].open_type = tables[1].open_type = tables[2].open_type =
      tables[3].open_type = tables[4].open_type = tables[5].open_type =
          OT_BASE_ONLY;
  tables[3].open_strategy = tables[4].open_strategy = tables[5].open_strategy =
      Table_ref::OPEN_IF_EXISTS;

  if (open_and_lock_tables(thd, tables, flags)) {
    /*
      Execution might have been interrupted; only print the error message
      if a user error condition has been raised. Also do not print expected/
      transient errors about tables not being locked (occurs when user does
      FLUSH PRIVILEGES under LOCK TABLES) and MDL deadlocks. These errors
      can't occur at start-up and will be reported to user anyway.
    */
    if (!is_expected_or_transient_error(thd)) {
      LogErr(ERROR_LEVEL, ER_AUTHCACHE_CANT_OPEN_AND_LOCK_PRIVILEGE_TABLES,
             thd->get_stmt_da()->message_text());
    }
    goto end;
  }

  if (!acl_cache_lock.lock()) goto end;

  old_acl_users = acl_users;
  old_acl_dbs = acl_dbs;
  old_acl_proxy_users = acl_proxy_users;
  old_acl_restrictions = std::move(acl_restrictions);
  swap_role_cache();
  roles_init();

  acl_users =
      new Prealloced_array<ACL_USER, ACL_PREALLOC_SIZE>(key_memory_acl_mem);
  acl_dbs = new Prealloced_array<ACL_DB, ACL_PREALLOC_SIZE>(key_memory_acl_mem);
  acl_proxy_users = new Prealloced_array<ACL_PROXY_USER, ACL_PREALLOC_SIZE>(
      key_memory_acl_mem);
  acl_restrictions = make_unique<Acl_restrictions>();

  // acl_load() overwrites global_acl_memory, so we need to free it.
  // However, we can't do that immediately, because acl_load() might fail,
  // and then we'd need to keep it.
  old_mem = std::move(global_acl_memory);
  delete acl_wild_hosts;
  acl_wild_hosts = nullptr;
  delete acl_check_hosts;
  acl_check_hosts = nullptr;
  old_dyn_priv_map =
      swap_dynamic_privileges_map(new User_to_dynamic_privileges_map());

  /*
    Revert to the old acl caches, if either loading of acl cache or role
    cache failed. We do this because roles caches maintain the shallow
    copies of the ACL_USER(s).
  */
  if ((return_val = acl_load(thd, tables)) ||
      (return_val = reload_roles_cache(
           thd, (tables + 4)))) {  // Error. Revert to old list
    DBUG_PRINT("error", ("Reverting to old privileges"));
    acl_free(); /* purecov: inspected */
    acl_users = old_acl_users;
    acl_dbs = old_acl_dbs;
    acl_proxy_users = old_acl_proxy_users;
    global_acl_memory = std::move(old_mem);
    acl_restrictions = std::move(old_acl_restrictions);
    // Revert to the old role caches
    swap_role_cache();
    // Old caches must be pointing to the global role caches right now
    delete_old_role_cache();

    init_check_host();
    delete swap_dynamic_privileges_map(old_dyn_priv_map);
    if (!old_dyn_priv_map) dynamic_privileges_init();
    if (acl_users) {
      rebuild_cached_acl_users_for_name();
      if (preserve_temporary_account_locking) {
        ACL_temporary_lock_state::restore_temporary_account_locking(
            nullptr, modified_user_lock_state_list);
      }
    }
  } else {
    if (acl_users && preserve_temporary_account_locking) {
      ACL_temporary_lock_state::restore_temporary_account_locking(
          old_acl_users, modified_user_lock_state_list);
    }
    delete old_acl_users;
    delete old_acl_dbs;
    delete old_acl_proxy_users;
    delete old_dyn_priv_map;
    // Delete the old role caches
    delete_old_role_cache();
    old_mem.Clear();
  }

end:
  if (!mdl_locked)
    commit_and_close_mysql_tables(thd);
  else
    close_thread_tables(thd);
  get_global_acl_cache()->increase_version();
  DEBUG_SYNC(thd, "after_acl_reload");
  return return_val;
}

void acl_insert_proxy_user(ACL_PROXY_USER *new_value) {
  DBUG_TRACE;
  assert(assert_acl_cache_write_lock(current_thd));
  auto upper_bound =
      std::upper_bound(acl_proxy_users->begin(), acl_proxy_users->end(),
                       *new_value, ACL_compare());
  acl_proxy_users->insert(upper_bound, *new_value);
}

struct Free_grant_table {
  void operator()(GRANT_TABLE *grant_table) const {
    grant_table->~GRANT_TABLE();
  }
};

/* Free grant array if possible */

void grant_free(void) {
  DBUG_TRACE;
  column_priv_hash.reset();
  proc_priv_hash.reset();
  func_priv_hash.reset();
  memex.Clear();
}

/**
  @brief Initialize structures responsible for table/column-level privilege
   checking and load information for them from tables in the 'mysql' database.

  @param skip_grant_tables  true if the command line option
    --skip-grant-tables is specified, else false.

  @return Error status
    @retval false OK
    @retval true  Could not initialize grant subsystem.
*/

bool grant_init(bool skip_grant_tables) {
  THD *thd;
  bool return_val;
  DBUG_TRACE;

  if (skip_grant_tables) return false;

  if (!(thd = new THD)) return true; /* purecov: deadcode */
  thd->thread_stack = (char *)&thd;
  thd->store_globals();

  return_val = grant_reload(thd, false);

  if (return_val && thd->get_stmt_da()->is_error())
    LogErr(ERROR_LEVEL, ER_AUTHCACHE_CANT_INIT_GRANT_SUBSYSTEM,
           thd->get_stmt_da()->message_text());

  if (opt_mandatory_roles.length > 0) {
    return_val |= check_authorization_id_string(thd, opt_mandatory_roles);
  }

  thd->release_resources();
  delete thd;

  return return_val;
}

/**
  @brief Helper function to grant_reload_procs_priv

  Reads the procs_priv table into memory hash.

  @param p_table A pointer to the procs_priv table structure.

  @see grant_reload
  @see grant_reload_procs_priv

  @return Error state
    @retval true An error occurred
    @retval false Success
*/

static bool grant_load_procs_priv(TABLE *p_table) {
  MEM_ROOT *memex_ptr;
  bool return_val = true;
  int error;
  bool check_no_resolve = specialflag & SPECIAL_NO_RESOLVE;
  MEM_ROOT **save_mem_root_ptr = THR_MALLOC;
  DBUG_TRACE;
  proc_priv_hash.reset(
      new malloc_unordered_multimap<string,
                                    unique_ptr_destroy_only<GRANT_NAME>>(
          key_memory_acl_memex));
  func_priv_hash.reset(
      new malloc_unordered_multimap<string,
                                    unique_ptr_destroy_only<GRANT_NAME>>(
          key_memory_acl_memex));
  error = p_table->file->ha_index_init(0, true);
  DBUG_EXECUTE_IF("wl7158_grant_load_proc_1", p_table->file->ha_index_end();
                  error = HA_ERR_LOCK_DEADLOCK;);
  if (error) {
    acl_print_ha_error(error);
    return true;
  }
  p_table->use_all_columns();

  error = p_table->file->ha_index_first(p_table->record[0]);
  assert(p_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
         error != HA_ERR_LOCK_DEADLOCK);
  assert(p_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
         error != HA_ERR_LOCK_WAIT_TIMEOUT);
  DBUG_EXECUTE_IF("wl7158_grant_load_proc_2", error = HA_ERR_LOCK_DEADLOCK;);

  if (error) {
    if (error == HA_ERR_END_OF_FILE)
      return_val = false;  // Return Ok.
    else
      acl_print_ha_error(error);
  } else {
    memex_ptr = &memex;
    THR_MALLOC = &memex_ptr;
    do {
      GRANT_NAME *mem_check;
      malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_NAME>>
          *hash;
      if (!(mem_check = new (memex_ptr) GRANT_NAME(p_table, true))) {
        /* This could only happen if we are out memory */
        goto end_unlock;
      }

      if (check_no_resolve) {
        if (hostname_requires_resolving(mem_check->host.get_host())) {
          LogErr(WARNING_LEVEL, ER_AUTHCACHE_PROCS_PRIV_SKIPPED_NEEDS_RESOLVE,
                 mem_check->tname, mem_check->user,
                 mem_check->host.get_host() ? mem_check->host.get_host() : "");
        }
      }
      const enum_sp_type sp_type = to_sp_type(p_table->field[4]->val_int());
      if (sp_type == enum_sp_type::PROCEDURE) {
        hash = proc_priv_hash.get();
      } else if (sp_type == enum_sp_type::FUNCTION) {
        hash = func_priv_hash.get();
      } else {
        LogErr(WARNING_LEVEL,
               ER_AUTHCACHE_PROCS_PRIV_ENTRY_IGNORED_BAD_ROUTINE_TYPE,
               mem_check->tname);
        destroy(mem_check);
        goto next_record;
      }

      mem_check->privs = fix_rights_for_procedure(mem_check->privs);
      if (!mem_check->ok()) {
        destroy(mem_check);
      } else {
        hash->emplace(mem_check->hash_key,
                      unique_ptr_destroy_only<GRANT_NAME>(mem_check));
      }
    next_record:
      error = p_table->file->ha_index_next(p_table->record[0]);
      assert(p_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_DEADLOCK);
      assert(p_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_WAIT_TIMEOUT);
      DBUG_EXECUTE_IF("wl7158_grant_load_proc_3",
                      error = HA_ERR_LOCK_DEADLOCK;);
      if (error) {
        if (error == HA_ERR_END_OF_FILE)
          return_val = false;
        else
          acl_print_ha_error(error);
        goto end_unlock;
      }
    } while (true);
  }

end_unlock:
  p_table->file->ha_index_end();
  THR_MALLOC = save_mem_root_ptr;
  return return_val;
}

/**
  @brief Initialize structures responsible for table/column-level privilege
    checking and load information about grants from open privilege tables.

  @param thd Current thread
  @param tables List containing open "mysql.tables_priv" and
    "mysql.columns_priv" tables.

  @see grant_reload

  @return Error state
    @retval false Success
    @retval true Error
*/

static bool grant_load(THD *thd, Table_ref *tables) {
  bool return_val = true;
  int error;
  TABLE *t_table = nullptr, *c_table = nullptr;
  bool check_no_resolve = specialflag & SPECIAL_NO_RESOLVE;
  sql_mode_t old_sql_mode = thd->variables.sql_mode;
  DBUG_TRACE;

  thd->variables.sql_mode &= ~MODE_PAD_CHAR_TO_FULL_LENGTH;

  column_priv_hash.reset(
      new malloc_unordered_multimap<string,
                                    unique_ptr_destroy_only<GRANT_TABLE>>(
          key_memory_acl_memex));

  t_table = tables[0].table;
  c_table = tables[1].table;
  error = t_table->file->ha_index_init(0, true);
  DBUG_EXECUTE_IF("wl7158_grant_load_1", t_table->file->ha_index_end();
                  error = HA_ERR_LOCK_DEADLOCK;);
  if (error) {
    acl_print_ha_error(error);
    goto end_index_init;
  }
  t_table->use_all_columns();
  c_table->use_all_columns();

  error = t_table->file->ha_index_first(t_table->record[0]);
  assert(t_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
         error != HA_ERR_LOCK_DEADLOCK);
  assert(t_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
         error != HA_ERR_LOCK_WAIT_TIMEOUT);
  DBUG_EXECUTE_IF("wl7158_grant_load_2", error = HA_ERR_LOCK_DEADLOCK;);
  if (error) {
    if (error == HA_ERR_END_OF_FILE)
      return_val = false;  // Return Ok.
    else
      acl_print_ha_error(error);
  } else {
    Swap_mem_root_guard guard(thd, &memex);
    do {
      GRANT_TABLE *mem_check = new (thd->mem_root) GRANT_TABLE(t_table);

      if (!mem_check) {
        /* This could only happen if we are out memory */
        goto end_unlock;
      }

      if (mem_check->init(c_table)) {
        destroy(mem_check);
        goto end_unlock;
      }

      if (check_no_resolve) {
        if (hostname_requires_resolving(mem_check->host.get_host()) &&
            strcmp(mem_check->host.get_host(), "localhost") != 0) {
          LogErr(WARNING_LEVEL, ER_AUTHCACHE_TABLES_PRIV_SKIPPED_NEEDS_RESOLVE,
                 mem_check->tname, mem_check->user ? mem_check->user : "",
                 mem_check->host.get_host() ? mem_check->host.get_host() : "");
        }
      }

      if (!mem_check->ok()) {
        destroy(mem_check);
      } else {
        column_priv_hash->emplace(
            mem_check->hash_key,
            unique_ptr_destroy_only<GRANT_TABLE>(mem_check));
      }
      error = t_table->file->ha_index_next(t_table->record[0]);
      assert(t_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_DEADLOCK);
      assert(t_table->file->ht->db_type == DB_TYPE_NDBCLUSTER ||
             error != HA_ERR_LOCK_WAIT_TIMEOUT);
      DBUG_EXECUTE_IF("wl7158_grant_load_3", error = HA_ERR_LOCK_DEADLOCK;);
      if (error) {
        if (error != HA_ERR_END_OF_FILE)
          acl_print_ha_error(error);
        else
          return_val = false;
        goto end_unlock;
      }

    } while (true);
  }

end_unlock:
  t_table->file->ha_index_end();
end_index_init:
  thd->variables.sql_mode = old_sql_mode;
  if (!return_val) initialized = true;
  return return_val;
}

/**
  @brief Helper function to grant_reload. Reloads procs_priv table is it
    exists.

  @param table A pointer to the table list.

  @see grant_reload

  @return Error state
    @retval false Success
    @retval true An error has occurred.
*/

static bool grant_reload_procs_priv(Table_ref *table) {
  DBUG_TRACE;

  /* Save a copy of the current hash if we need to undo the grant load */
  unique_ptr<
      malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_NAME>>>
      old_proc_priv_hash(std::move(proc_priv_hash));
  unique_ptr<
      malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_NAME>>>
      old_func_priv_hash(std::move(func_priv_hash));
  bool return_val = false;

  if ((return_val = grant_load_procs_priv(table->table))) {
    /* Error; Reverting to old hash */
    DBUG_PRINT("error", ("Reverting to old privileges"));
    proc_priv_hash = std::move(old_proc_priv_hash);
    func_priv_hash = std::move(old_func_priv_hash);
  }

  return return_val;
}

/**
  @brief Reload information about table and column level privileges if
  possible

  @param thd        Current thread
  @param mdl_locked MDL lock status - affects open/close table operations

  Locked tables are checked by acl_reload() and doesn't have to be checked
  in this call.
  This function is also used for initialization of structures responsible
  for table/column-level privilege checking.

  @return Error state
    @retval false Success
    @retval true  Error
*/

bool grant_reload(THD *thd, bool mdl_locked) {
  DBUG_EXECUTE_IF("enable_debug_sync_before_grant_reload",
                  { DEBUG_SYNC(thd, "before_grant_reload"); };);
  MEM_ROOT old_mem;
  bool return_val = true;
  uint flags = mdl_locked
                   ? MYSQL_OPEN_HAS_MDL_LOCK | MYSQL_LOCK_IGNORE_TIMEOUT |
                         MYSQL_OPEN_IGNORE_FLUSH
                   : MYSQL_LOCK_IGNORE_TIMEOUT;
  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::WRITE_MODE);

  DBUG_TRACE;

  Table_ref tables[3] = {

      /*
        Acquiring strong MDL lock allows to avoid deadlock and timeout errors
        from SE level.
      */
      Table_ref("mysql", "tables_priv", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "columns_priv", TL_READ, MDL_SHARED_READ_ONLY),

      Table_ref("mysql", "procs_priv", TL_READ, MDL_SHARED_READ_ONLY)};

  tables[0].next_local = tables[0].next_global = tables + 1;
  tables[1].next_local = tables[1].next_global = tables + 2;
  tables[0].open_type = tables[1].open_type = tables[2].open_type =
      OT_BASE_ONLY;

  if (open_and_lock_tables(thd, tables, flags)) {
    if (!is_expected_or_transient_error(thd)) {
      LogErr(ERROR_LEVEL, ER_AUTHCACHE_CANT_OPEN_AND_LOCK_PRIVILEGE_TABLES,
             thd->get_stmt_da()->message_text());
    }
    goto end;
  }

  if (!acl_cache_lock.lock()) goto end;

  {
    unique_ptr<
        malloc_unordered_multimap<string, unique_ptr_destroy_only<GRANT_TABLE>>>
        old_column_priv_hash(std::move(column_priv_hash));

    /*
      Create a new memory pool but save the current memory pool to make an
      undo operation possible in case of failure.
    */
    old_mem = std::move(memex);
    init_sql_alloc(key_memory_acl_memex, &memex, ACL_ALLOC_BLOCK_SIZE);
    /*
      tables[2].table i.e. procs_priv can be null if we are working with
      pre 4.1 privilege tables
    */
    if ((return_val = (grant_load(thd, tables) ||
                       grant_reload_procs_priv(
                           &tables[2])))) {  // Error. Revert to old hash
      DBUG_PRINT("error", ("Reverting to old privileges"));
      column_priv_hash =
          std::move(old_column_priv_hash); /* purecov: deadcode */
      memex.Clear();
      memex = std::move(old_mem); /* purecov: deadcode */
    } else {                      // Reload successful
      old_column_priv_hash.reset();
      old_mem.Clear();
      grant_version++;
      get_global_acl_cache()->increase_version();
    }
  }

end:
  if (!mdl_locked)
    commit_and_close_mysql_tables(thd);
  else
    close_thread_tables(thd);
  return return_val;
}

void acl_update_user(const char *user, const char *host, enum SSL_type ssl_type,
                     const char *ssl_cipher, const char *x509_issuer,
                     const char *x509_subject, USER_RESOURCES *mqh,
                     Access_bitmask privileges, const LEX_CSTRING &plugin,
                     const LEX_CSTRING &auth, const std::string &second_auth,
                     const MYSQL_TIME &password_change_time,
                     const LEX_ALTER &password_life, Restrictions &restrictions,
                     acl_table::Pod_user_what_to_update &what_to_update,
                     uint failed_login_attempts, int password_lock_time,
                     const I_multi_factor_auth *mfa) {
  DBUG_TRACE;
  assert(assert_acl_cache_write_lock(current_thd));
  for (ACL_USER *acl_user = acl_users->begin(); acl_user != acl_users->end();
       ++acl_user) {
    if ((!acl_user->user && !user[0]) ||
        (acl_user->user && !strcmp(user, acl_user->user))) {
      if ((!acl_user->host.get_host() && !host[0]) ||
          (acl_user->host.get_host() &&
           !my_strcasecmp(system_charset_info, host,
                          acl_user->host.get_host()))) {
        if (plugin.length > 0) {
          acl_user->plugin.str = plugin.str;
          acl_user->plugin.length = plugin.length;
          optimize_plugin_compare_by_pointer(&acl_user->plugin);
          if (!auth_plugin_is_built_in(acl_user->plugin.str))
            acl_user->plugin.str =
                strmake_root(&global_acl_memory, plugin.str, plugin.length);
          /* Update auth string only when specified in ALTER/GRANT */
          if (auth.str) {
            if (auth.length == 0)
              acl_user->credentials[PRIMARY_CRED].m_auth_string = EMPTY_CSTR;
            else
              acl_user->credentials[PRIMARY_CRED].m_auth_string.str =
                  strmake_root(&global_acl_memory, auth.str, auth.length);
            acl_user->credentials[PRIMARY_CRED].m_auth_string.length =
                auth.length;
            set_user_salt(acl_user);
            if (password_change_time.time_type != MYSQL_TIMESTAMP_ERROR)
              acl_user->password_last_changed = password_change_time;
          }

          if (what_to_update.m_what & USER_ATTRIBUTES) {
            if (what_to_update.m_user_attributes &
                acl_table::USER_ATTRIBUTE_RETAIN_PASSWORD) {
              acl_user->credentials[SECOND_CRED].m_auth_string.str =
                  strmake_root(&global_acl_memory, second_auth.c_str(),
                               second_auth.length());
              acl_user->credentials[SECOND_CRED].m_auth_string.length =
                  second_auth.length();
            }
            if (what_to_update.m_user_attributes &
                acl_table::USER_ATTRIBUTE_DISCARD_PASSWORD) {
              acl_user->credentials[SECOND_CRED].m_auth_string = EMPTY_CSTR;
            }
            if (what_to_update.m_user_attributes &
                (acl_table::USER_ATTRIBUTE_FAILED_LOGIN_ATTEMPTS |
                 acl_table::USER_ATTRIBUTE_PASSWORD_LOCK_TIME)) {
              acl_user->password_locked_state.set_parameters(
                  password_lock_time, failed_login_attempts);
            }
            set_user_salt(acl_user);
          }
        }
        DBUG_PRINT("info",
                   ("Updates global privilege for %s@%s to %" PRIu32,
                    acl_user->user, acl_user->host.get_host(), privileges));
        acl_user->access = privileges;
        if (what_to_update.m_what & USER_ATTRIBUTES &&
            (what_to_update.m_user_attributes &
             acl_table::USER_ATTRIBUTE_RESTRICTIONS))
          acl_restrictions->upsert_restrictions(acl_user, restrictions);

        if (mqh->specified_limits & USER_RESOURCES::QUERIES_PER_HOUR)
          acl_user->user_resource.questions = mqh->questions;
        if (mqh->specified_limits & USER_RESOURCES::UPDATES_PER_HOUR)
          acl_user->user_resource.updates = mqh->updates;
        if (mqh->specified_limits & USER_RESOURCES::CONNECTIONS_PER_HOUR)
          acl_user->user_resource.conn_per_hour = mqh->conn_per_hour;
        if (mqh->specified_limits & USER_RESOURCES::USER_CONNECTIONS)
          acl_user->user_resource.user_conn = mqh->user_conn;
        if (ssl_type != SSL_TYPE_NOT_SPECIFIED) {
          acl_user->ssl_type = ssl_type;
          acl_user->ssl_cipher =
              (ssl_cipher ? strdup_root(&global_acl_memory, ssl_cipher)
                          : nullptr);
          acl_user->x509_issuer =
              (x509_issuer ? strdup_root(&global_acl_memory, x509_issuer)
                           : nullptr);
          acl_user->x509_subject =
              (x509_subject ? strdup_root(&global_acl_memory, x509_subject)
                            : nullptr);
        }
        /* update details related to password lifetime, password expiry */
        if (password_life.update_password_expired_column ||
            what_to_update.m_what & PLUGIN_ATTR)
          acl_user->password_expired =
              password_life.update_password_expired_column;
        if (!password_life.update_password_expired_column &&
            password_life.update_password_expired_fields) {
          if (!password_life.use_default_password_lifetime) {
            acl_user->password_lifetime = password_life.expire_after_days;
            acl_user->use_default_password_lifetime = false;
          } else
            acl_user->use_default_password_lifetime = true;
        }

        if (password_life.update_account_locked_column) {
          acl_user->account_locked = password_life.account_locked;

          /* reset the runtime locked state if there is account locking */
          if (!acl_user->account_locked &&
              acl_user->password_locked_state.is_active())
            acl_user->password_locked_state.set_parameters(
                acl_user->password_locked_state.get_password_lock_time_days(),
                acl_user->password_locked_state.get_failed_login_attempts());
        }

        /* Update role graph  */
        string authid_role = create_authid_str_from(acl_user);
        Role_index_map::iterator it = g_authid_to_vertex->find(authid_role);
        if (it != g_authid_to_vertex->end()) {
          boost::property_map<Granted_roles_graph,
                              boost::vertex_acl_user_t>::type user_pacl_user;
          user_pacl_user =
              boost::get(boost::vertex_acl_user_t(), *g_granted_roles);
          boost::put(user_pacl_user, it->second, *acl_user);
        }

        /* update password history */
        if (password_life.update_password_history) {
          acl_user->use_default_password_history =
              password_life.use_default_password_history;
          acl_user->password_history_length =
              password_life.use_default_password_history
                  ? 0
                  : password_life.password_history_length;
        }
        /* update password history */
        if (password_life.update_password_reuse_interval) {
          acl_user->use_default_password_reuse_interval =
              password_life.use_default_password_reuse_interval;
          acl_user->password_reuse_interval =
              password_life.use_default_password_reuse_interval
                  ? 0
                  : password_life.password_reuse_interval;
        }
        /* update current password field value */
        if (password_life.update_password_require_current !=
            Lex_acl_attrib_udyn::UNCHANGED) {
          acl_user->password_require_current =
              password_life.update_password_require_current;
        }
        /* get details of Multi factor authentication */
        acl_user->set_mfa(nullptr, const_cast<I_multi_factor_auth *>(mfa));
        /* search complete: */
        break;
      }
    }
  }
}

void acl_users_add_one(const char *user, const char *host,
                       enum SSL_type ssl_type, const char *ssl_cipher,
                       const char *x509_issuer, const char *x509_subject,
                       USER_RESOURCES *mqh, Access_bitmask privileges,
                       const LEX_CSTRING &plugin, const LEX_CSTRING &auth,
                       const LEX_CSTRING &second_auth,
                       const MYSQL_TIME &password_change_time,
                       const LEX_ALTER &password_life, bool add_role_vertex,
                       Restrictions &restrictions, uint failed_login_attempts,
                       int password_lock_time, const I_multi_factor_auth *mfa,
                       THD *thd [[maybe_unused]]) {
  DBUG_TRACE;
  ACL_USER acl_user;

  assert(assert_acl_cache_write_lock(thd));
  /*
  All accounts can authenticate per default. This will change when
  we add a new field to the user table.

  Currently this flag is only set to false when authentication is attempted
  using an unknown user name.
  */
  acl_user.can_authenticate = true;

  acl_user.set_user(&global_acl_memory, user);
  acl_user.set_host(&global_acl_memory, host);
  assert(plugin.str);
  if (plugin.str[0]) {
    acl_user.plugin = plugin;
    optimize_plugin_compare_by_pointer(&acl_user.plugin);
    if (!auth_plugin_is_built_in(acl_user.plugin.str))
      acl_user.plugin.str =
          strmake_root(&global_acl_memory, plugin.str, plugin.length);
    acl_user.credentials[PRIMARY_CRED].m_auth_string.str =
        auth.str ? strmake_root(&global_acl_memory, auth.str, auth.length) : "";
    acl_user.credentials[PRIMARY_CRED].m_auth_string.length =
        auth.str ? auth.length : 0;
    if (second_auth.length) {
      acl_user.credentials[SECOND_CRED].m_auth_string.str =
          strmake_root(&global_acl_memory, second_auth.str, second_auth.length);
      acl_user.credentials[SECOND_CRED].m_auth_string.length =
          second_auth.length;
    } else {
      acl_user.credentials[SECOND_CRED].m_auth_string = EMPTY_CSTR;
    }
    optimize_plugin_compare_by_pointer(&acl_user.plugin);
  }

  acl_user.access = privileges;
  acl_user.user_resource = *mqh;
  acl_user.sort = get_sort(2, acl_user.host.get_host(), acl_user.user);
  // acl_user.hostname_length=(uint) strlen(host);
  acl_user.ssl_type =
      (ssl_type != SSL_TYPE_NOT_SPECIFIED ? ssl_type : SSL_TYPE_NONE);
  acl_user.ssl_cipher =
      ssl_cipher ? strdup_root(&global_acl_memory, ssl_cipher) : nullptr;
  acl_user.x509_issuer =
      x509_issuer ? strdup_root(&global_acl_memory, x509_issuer) : nullptr;
  acl_user.x509_subject =
      x509_subject ? strdup_root(&global_acl_memory, x509_subject) : nullptr;
  /* update details related to password lifetime, password expiry, history */
  acl_user.password_expired = password_life.update_password_expired_column;
  acl_user.password_lifetime = password_life.expire_after_days;
  acl_user.use_default_password_lifetime =
      password_life.use_default_password_lifetime;
  acl_user.password_last_changed = password_change_time;
  acl_user.account_locked = password_life.account_locked;
  acl_user.password_history_length =
      password_life.use_default_password_history
          ? 0
          : password_life.password_history_length;
  acl_user.use_default_password_history =
      password_life.use_default_password_history;
  acl_user.password_reuse_interval =
      password_life.use_default_password_reuse_interval
          ? 0
          : password_life.password_reuse_interval;
  acl_user.use_default_password_reuse_interval =
      password_life.use_default_password_reuse_interval;

  /*
  Assign the password_require_current field value to the ACL USER.
  if it was not specified then assign the default value
  */
  if (password_life.update_password_require_current ==
      Lex_acl_attrib_udyn::UNCHANGED) {
    acl_user.password_require_current = Lex_acl_attrib_udyn::DEFAULT;
  } else {
    acl_user.password_require_current =
        password_life.update_password_require_current;
  }
  acl_restrictions->upsert_restrictions(&acl_user, restrictions);
  set_user_salt(&acl_user);
  /* New user is not a role by default. */
  acl_user.is_role = false;

  acl_user.password_locked_state.set_parameters(password_lock_time,
                                                failed_login_attempts);
  /* get details of Multi factor authentication */
  acl_user.set_mfa(nullptr, const_cast<I_multi_factor_auth *>(mfa));

  acl_users->push_back(acl_user);
  if (acl_user.host.check_allow_all_hosts())
    allow_all_hosts = true;  // Anyone can connect /* purecov: tested */

  if (add_role_vertex) {
    /*
      Add vertex to role graph. ACL_USER object is copied with a shallow copy
    */
    create_role_vertex(&acl_user);
  }
}

void acl_insert_user(THD *thd [[maybe_unused]], const char *user,
                     const char *host, enum SSL_type ssl_type,
                     const char *ssl_cipher, const char *x509_issuer,
                     const char *x509_subject, USER_RESOURCES *mqh,
                     Access_bitmask privileges, const LEX_CSTRING &plugin,
                     const LEX_CSTRING &auth,
                     const MYSQL_TIME &password_change_time,
                     const LEX_ALTER &password_life, Restrictions &restrictions,
                     uint failed_login_attempts, int password_lock_time,
                     const I_multi_factor_auth *mfa) {
  DBUG_TRACE;
  acl_users_add_one(user, host, ssl_type, ssl_cipher, x509_issuer, x509_subject,
                    mqh, privileges, plugin, auth, EMPTY_CSTR,
                    password_change_time, password_life, true, restrictions,
                    failed_login_attempts, password_lock_time, mfa, thd);
  /*
    acl_users_add_one() has added new entry as the last element in array,
    let us move it to the position which is correct according to sort order.
  */
  auto upper_bound = std::upper_bound(acl_users->begin(), acl_users->end() - 1,
                                      acl_users->back(), ACL_USER_compare());
  std::rotate(upper_bound, acl_users->end() - 1, acl_users->end());
  rebuild_cached_acl_users_for_name();
  /* Rebuild 'acl_check_hosts' since 'acl_users' has been modified */
  rebuild_check_host();
  /* reparse mandatory roles variable */
  opt_mandatory_roles_cache = false;
}

void acl_update_proxy_user(ACL_PROXY_USER *new_value, bool is_revoke) {
  DBUG_TRACE;
  assert(assert_acl_cache_write_lock(current_thd));
  for (ACL_PROXY_USER *acl_user = acl_proxy_users->begin();
       acl_user != acl_proxy_users->end(); ++acl_user) {
    if (acl_user->pk_equals(new_value)) {
      if (is_revoke) {
        DBUG_PRINT("info", ("delting ACL_PROXY_USER"));
        acl_proxy_users->erase(acl_user);
      } else {
        DBUG_PRINT("info", ("updating ACL_PROXY_USER"));
        acl_user->set_data(new_value);
      }
      break;
    }
  }
}

void acl_update_db(const char *user, const char *host, const char *db,
                   Access_bitmask privileges) {
  assert(assert_acl_cache_write_lock(current_thd));

  for (ACL_DB *acl_db = acl_dbs->begin(); acl_db < acl_dbs->end();) {
    if ((!acl_db->user && !user[0]) ||
        (acl_db->user && !strcmp(user, acl_db->user))) {
      if ((!acl_db->host.get_host() && !host[0]) ||
          (acl_db->host.get_host() && !strcmp(host, acl_db->host.get_host()))) {
        if ((!acl_db->db && !db[0]) ||
            (acl_db->db && !strcmp(db, acl_db->db))) {
          if (privileges)
            acl_db->access = privileges;
          else {
            acl_db = acl_dbs->erase(acl_db);
            // Don't increment loop variable.
            continue;
          }
        }
      }
    }
    ++acl_db;
  }
}

/*
  Insert a user/db/host combination into the global acl_cache

  SYNOPSIS
    acl_insert_db()
    user                User name
    host                Host name
    db                  Database name
    privileges          Bitmap of privileges

  NOTES
    Acl caches must be locked
*/

void acl_insert_db(const char *user, const char *host, const char *db,
                   Access_bitmask privileges) {
  ACL_DB acl_db;
  assert(assert_acl_cache_write_lock(current_thd));
  acl_db.set_user(&global_acl_memory, user);
  acl_db.set_host(&global_acl_memory, host);
  acl_db.db = strdup_root(&global_acl_memory, db);
  acl_db.access = privileges;
  acl_db.sort = get_sort(3, acl_db.host.get_host(), acl_db.db, acl_db.user);
  auto upper_bound =
      std::upper_bound(acl_dbs->begin(), acl_dbs->end(), acl_db, ACL_compare());
  acl_dbs->insert(upper_bound, acl_db);
}

void get_mqh(THD *thd, const char *user, const char *host, USER_CONN *uc) {
  ACL_USER *acl_user;
  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::READ_MODE);

  if (initialized && acl_cache_lock.lock(false) &&
      (acl_user = find_acl_user(host, user, false)))
    uc->user_resources = acl_user->user_resource;
  else
    memset(&uc->user_resources, 0, sizeof(uc->user_resources));
}

/**
  Update the security context when updating the user

  Helper function.
  Update only if the security context is pointing to the same user and
  the user is not a proxied user for a different proxy user.
  And return true if the update happens (i.e. we're operating on the
  user account of the current user).
  Normalize the names for a safe compare.

  @param sctx           The security context to update
  @param acl_user_ptr   User account being updated
  @param expired        new value of the expiration flag
  @return               did the update happen ?
 */
bool update_sctx_cache(Security_context *sctx, ACL_USER *acl_user_ptr,
                       bool expired) {
  const char *acl_host = acl_user_ptr->host.get_host();
  const char *acl_user = acl_user_ptr->user;
  const char *sctx_user = sctx->priv_user().str;
  const char *sctx_host = sctx->priv_host().str;

  /* If the user is connected as a proxied user, verify against proxy user */
  if (sctx->proxy_user().str && *sctx->proxy_user().str != '\0') {
    sctx_user = sctx->user().str;
  }

  if (!acl_host) acl_host = "";
  if (!acl_user) acl_user = "";
  if (!sctx_host) sctx_host = "";
  if (!sctx_user) sctx_user = "";

  if (!strcmp(acl_user, sctx_user) && !strcmp(acl_host, sctx_host)) {
    sctx->set_password_expired(expired);
    return true;
  }

  return false;
}

struct Acl_hash_entry {
  uint64 version;
  uchar *key;
  unsigned key_length;
  Acl_map *map;
};

const uchar *hash_key(const uchar *el, size_t *length) {
  const Acl_hash_entry *entry = reinterpret_cast<const Acl_hash_entry *>(el);
  *length = entry->key_length;
  return entry->key;
}

/**
  Allocate a new cache key based on active roles, current user and
  global cache version

  @param [out] out_key The resulting key
  @param [out] key_len Key length
  @param version Global Acl_cache version
  @param uid The authorization ID of the current user
  @param active_roles The active roles of the current user

  @return Success state
    @retval true OK
    @retval false Fatal error occurred.
*/

bool create_acl_cache_hash_key(uchar **out_key, unsigned *key_len,
                               uint64 version, const Auth_id_ref &uid,
                               const List_of_auth_id_refs &active_roles) {
  List_of_auth_id_refs::const_iterator it = active_roles.begin();
  uint32 active_roles_size = 0;
  for (; it != active_roles.end(); ++it) {
    active_roles_size += it->first.length + it->second.length + 2;
  }
  auto auth_id(uid);
  *key_len = auth_id.first.length + auth_id.second.length + 2 + sizeof(uint64) +
             active_roles_size;
  *out_key =
      (uchar *)my_malloc(key_memory_acl_map_cache, *key_len, MYF(MY_WME));
  if (out_key == nullptr) return false;
  auth_id.first.str = auth_id.first.str ? auth_id.first.str : "";
  auth_id.second.str = auth_id.second.str ? auth_id.second.str : "";
  memcpy(*out_key, auth_id.first.str, auth_id.first.length);
  *(*out_key + uid.first.length) = '@';
  memcpy(*out_key + auth_id.first.length + 1, auth_id.second.str,
         auth_id.second.length);
  uint offset = auth_id.first.length + auth_id.second.length + 1;
  /* Separator between version and role */
  *(*out_key + offset) = '`';
  ++offset;
  memcpy(*out_key + offset, reinterpret_cast<void *>(&version), sizeof(uint64));
  it = active_roles.begin();
  offset += sizeof(uint64);

  for (; it != active_roles.end(); ++it) {
    memcpy(*out_key + offset, it->first.str, it->first.length);
    *(*out_key + offset + it->first.length) = '@';
    memcpy(*out_key + offset + it->first.length + 1, it->second.str,
           it->second.length);
    offset += it->first.length + it->second.length + 1;
    /* Separator between roles */
    *(*out_key + offset) = '`';
    ++offset;
  }
  assert(((offset - *key_len) == 0));
  return true;
}

Acl_cache::Acl_cache() : m_role_graph_version(0L) {
  const char *category = "sql";
  int count;
  count = static_cast<int>(array_elements(all_acl_cache_mutexes));
  mysql_mutex_register(category, all_acl_cache_mutexes, count);
  lf_hash_init(&m_cache, sizeof(Acl_hash_entry), LF_HASH_UNIQUE,
               0, /* key offset */
               0, /* key length not used */
               hash_key, &my_charset_bin);
  mysql_mutex_init(key_LOCK_acl_cache_flush, &m_cache_flush_mutex,
                   MY_MUTEX_INIT_SLOW);
}

Acl_cache::~Acl_cache() {
  mysql_mutex_destroy(&m_cache_flush_mutex);
  lf_hash_destroy(&m_cache);
}

Acl_map::Acl_map(Security_context *sctx, uint64 ver)
    : m_reference_count(0), m_version(ver), m_restrictions() {
  DBUG_TRACE;
  Acl_cache_lock_guard acl_cache_lock(current_thd,
                                      Acl_cache_lock_mode::READ_MODE);
  if (!acl_cache_lock.lock(false)) {
    DBUG_PRINT("error", ("Acl_map could not be constructed for user %s@%s => "
                         "Could not lock Acl caches.",
                         sctx->priv_user().str, sctx->priv_host().str));
    return;
  }
  m_global_acl = 0;
  ACL_USER *acl_user =
      find_acl_user(sctx->priv_host().str, sctx->priv_user().str, true);
  if (acl_user == nullptr) {
    DBUG_PRINT("error", ("Acl_map could not be constructed for user %s@%s => "
                         "No such user",
                         sctx->priv_user().str, sctx->priv_host().str));
    return;
  }
  List_of_granted_roles granted_roles;
  get_privilege_access_maps(
      acl_user, sctx->get_active_roles(), &m_global_acl, &m_db_acls,
      &m_db_wild_acls, &m_table_acls, &m_sp_acls, &m_func_acls, &granted_roles,
      &m_with_admin_acls, &m_dynamic_privileges, m_restrictions);
}

Acl_map::~Acl_map() {
  // Db_access_map is automatically destroyed and cleaned up.
  assert(reference_count() == 0);
}

Acl_map::Acl_map(const Acl_map &&map) { operator=(map); }

Acl_map &Acl_map::operator=(Acl_map &&map) {
  m_db_acls = std::move(map.m_db_acls);
  m_global_acl = map.m_global_acl;
  m_reference_count = map.m_reference_count.load();
  m_table_acls = std::move(map.m_table_acls);
  m_sp_acls = std::move(map.m_sp_acls);
  m_func_acls = std::move(map.m_func_acls);
  m_with_admin_acls = std::move(map.m_with_admin_acls);
  m_version = map.m_version;
  m_restrictions = map.m_restrictions;
  map.m_reference_count = 0;
  return *this;
}

Acl_map &Acl_map::operator=(const Acl_map &) { return *this; }

Access_bitmask Acl_map::global_acl() { return m_global_acl; }

Db_access_map *Acl_map::db_acls() { return &m_db_acls; }

Db_access_map *Acl_map::db_wild_acls() { return &m_db_wild_acls; }

Table_access_map *Acl_map::table_acls() { return &m_table_acls; }

Grant_acl_set *Acl_map::grant_acls() { return &m_with_admin_acls; }

SP_access_map *Acl_map::sp_acls() { return &m_sp_acls; }

SP_access_map *Acl_map::func_acls() { return &m_func_acls; }

Dynamic_privileges *Acl_map::dynamic_privileges() {
  return &m_dynamic_privileges;
}

Restrictions &Acl_map::restrictions() { return m_restrictions; }

void Acl_map::increase_reference_count() { ++m_reference_count; }

void Acl_map::decrease_reference_count() { --m_reference_count; }

void Acl_cache::increase_version() {
  DBUG_TRACE;
  mysql_mutex_lock(&m_cache_flush_mutex);
  /**
    We increase the m_role_graph_version twice here i.e., once before the
    flush_cache() and once after to indicate that flush has started and stopped.
    With this, the LF Hash will now contains the records with:
    a. version equal to the global version => current records
    b. version+1 equal to global version => almost old (could be still in use)
    c. version+2 less or equal to global version => old records (safe to delete)
    We check the reference count along with the map version in cache_flusher()
    to match the old records that can be deleted and we delete only those
    objects from the hash.
   */
  ++m_role_graph_version;
  flush_cache();
  ++m_role_graph_version;
  mysql_mutex_unlock(&m_cache_flush_mutex);
}

uint64 Acl_cache::version() { return m_role_graph_version.load(); }

int32 Acl_cache::size() { return m_cache.count.load(); }

Acl_map *Acl_cache::checkout_acl_map(Security_context *sctx, Auth_id_ref &uid,
                                     List_of_auth_id_refs &active_roles) {
  DBUG_TRACE;
  // CREATE KEY
  uchar *key;  // allocated by create_hash_key and released by
               // Acl_cache::flush_cache
  unsigned key_len;
  uint64 version = m_role_graph_version.load();
  if (!create_acl_cache_hash_key(&key, &key_len, version, uid, active_roles)) {
    /* OOM happened */
    active_roles.clear();
    return nullptr;
  }
  LF_PINS *pins = lf_hash_get_pins(&m_cache);
  Acl_hash_entry *entry =
      (Acl_hash_entry *)lf_hash_search(&m_cache, pins, key, key_len);
  if (entry == nullptr || entry == MY_LF_ERRPTR) {
    lf_hash_search_unpin(pins);
    Acl_map *map = create_acl_map(version, sctx);  // deleted in cache_flusher
    map->increase_reference_count();
    Acl_hash_entry new_entry;
    new_entry.version = version;
    new_entry.map = map;
    new_entry.key = key;
    new_entry.key_length = key_len;
    int rc =
        lf_hash_insert(&m_cache, pins, &new_entry);  // shallow copy of entry
    if (rc != 0) {
      /* There was a duplicate; throw away the allocated memory */
      lf_hash_put_pins(pins);
      map->decrease_reference_count();
      my_free(key);
      delete map;
      DBUG_PRINT("info", ("Someone else checked out the cache key"));
      /* Potentially dangerous to dive here? */
      return checkout_acl_map(sctx, uid, active_roles);
    }
    DBUG_EXECUTE_IF("test_acl_race_condition", {
      if (sctx->get_thd()->get_command() == COM_CONNECT) {
        const char act[] = "now SIGNAL map_inserted WAIT_FOR map_removed";
        assert(!debug_sync_set_action(current_thd, STRING_WITH_LEN(act)));
      }
    });
    lf_hash_put_pins(pins);
    DBUG_PRINT("info", ("Checked out new privilege map. Key= %s", key));
    return map;
  }
  Acl_map *map = entry->map;
  map->increase_reference_count();
  lf_hash_search_unpin(pins);
  lf_hash_put_pins(pins);
  my_free(key);
  DBUG_PRINT("info", ("Checked out old privilege map. Key= %s", key));
  return map;
}

void Acl_cache::return_acl_map(Acl_map *map) {
  map->decrease_reference_count();
}

/**
  This global is protected by the Acl_cache::m_cache_flush_mutex and used when
  iterating the Acl_map hash in Acl_cache::flush_cache
  @see Acl_cache::flush_cache
*/
uint64 l_cache_flusher_global_version;

/**
  Utility function for removing items from the hash.
  @param ptr A pointer to a Acl_hash_entry
  @param arg not used
  @return Always 0 with the intention that this causes the hash_search
  function to iterate every single element in the hash.
*/
static int cache_flusher(const uchar *ptr, void *arg [[maybe_unused]]) {
  DBUG_TRACE;
  const Acl_hash_entry *entry = reinterpret_cast<const Acl_hash_entry *>(ptr);
  if (entry != nullptr) {
    if (entry->map->reference_count() == 0 &&
        ((entry->map->version() + 2) <= l_cache_flusher_global_version))
      return 1;
  }
  return 0;
}

void Acl_cache::flush_cache() {
  DBUG_TRACE;
  DEBUG_SYNC(current_thd, "flush_cache_begin");
  if (m_cache.size == 0) return;
  Acl_hash_entry *entry = nullptr;
  l_cache_flusher_global_version = version();
  do {
    LF_PINS *pins = lf_hash_get_pins(&m_cache);
    entry = static_cast<Acl_hash_entry *>(
        lf_hash_random_match(&m_cache, pins, &cache_flusher, 0, nullptr));
    if (entry &&
        !lf_hash_delete(&m_cache, pins, entry->key, entry->key_length)) {
      // Hash element is removed from cache; safe to delete
      my_free(entry->key);
      delete entry->map;
    }
    lf_hash_search_unpin(pins);
    lf_hash_put_pins(pins);
  } while (entry != nullptr);
  DEBUG_SYNC(current_thd, "flush_cache_end");
}

/**
  Helper function for Acl_cache::clear_acl_cache

  @returns 1 to indicate that entry is a match
*/
int match_all_entries(const uchar *, void *) { return 1; }

/**
  This method is called from the shutdown_acl_cache() to remove the remaining
  entries, if any present, from the Acl_cache irrespective of the reference
  count or the map version.
*/
void Acl_cache::clear_acl_cache() {
  Acl_hash_entry *entry = nullptr;
  if (m_cache.size == 0) return;
  do {
    LF_PINS *pins = lf_hash_get_pins(&m_cache);
    entry = static_cast<Acl_hash_entry *>(
        lf_hash_random_match(&m_cache, pins, match_all_entries, 0, nullptr));
    if (entry != nullptr && entry != MY_LF_ERRPTR &&
        !lf_hash_delete(&m_cache, pins, entry->key, entry->key_length)) {
      // Hash element is removed from cache; safe to delete
      my_free(entry->key);
      delete entry->map;
    }
    lf_hash_search_unpin(pins);
    lf_hash_put_pins(pins);
  } while (entry != nullptr);
}

Acl_map *Acl_cache::create_acl_map(uint64 version, Security_context *sctx) {
  Acl_map *map = new Acl_map(sctx, version);
  return map;
}

void *Acl_map::operator new(size_t size) {
  return my_malloc(key_memory_acl_map_cache, size, MYF(0));
}

void Acl_map::operator delete(void *p) { my_free(p); }

void init_acl_cache() {
  g_acl_cache = new Acl_cache();
  g_mandatory_roles = new std::vector<Role_id>;
  opt_mandatory_roles_cache = false;
}

/**
  Shutdown the global Acl_cache system which was only initialized if the
  rwlocks were initialized.
  @see acl_init()
*/

void shutdown_acl_cache() {
  if (!acl_cache_initialized) return;

  /* This should clean up all remaining Acl_cache items */
  g_acl_cache->clear_acl_cache();
  assert(g_acl_cache->size() == 0);
  delete g_acl_cache;
  g_acl_cache = nullptr;
  roles_delete();
  dynamic_privileges_delete();
  delete g_mandatory_roles;
}

/* Constants used by Acl_cache_lock_guard */
static const ulong ACL_CACHE_LOCK_TIMEOUT = 3600UL;
static const MDL_key ACL_CACHE_KEY(MDL_key::ACL_CACHE, "", "");

/**
  Internal_error_handler subclass to suppress ER_LOCK_DEADLOCK,
  ER_LOCK_WAIT_TIMEOUT, ER_QUERY_INTERRUPTED and ER_QUERY_TIMEOUT.
  Instead, we will use Acl_cache_lock_guard::lock()
  to raise ER_CANNOT_LOCK_USER_MANAGEMENT_CACHES error.
*/
class Acl_cache_error_handler : public Internal_error_handler {
 public:
  /**
    Handle an error condition

    @param [in] thd           THD handle
    @param [in] sql_errno     Error raised by MDL subsystem
    @param [in] sqlstate      SQL state. Unused.
    @param [in] level         Severity level. Unused.
    @param [in] msg           Message string. Unused.
  */

  bool handle_condition(THD *thd [[maybe_unused]], uint sql_errno,
                        const char *sqlstate [[maybe_unused]],
                        Sql_condition::enum_severity_level *level
                        [[maybe_unused]],
                        const char *msg [[maybe_unused]]) override {
    return (sql_errno == ER_LOCK_DEADLOCK ||
            sql_errno == ER_LOCK_WAIT_TIMEOUT ||
            sql_errno == ER_QUERY_INTERRUPTED || sql_errno == ER_QUERY_TIMEOUT);
  }
};

/*
  MDL_release_locks_visitor subclass to release MDL for ACL_CACHE.
*/
class Release_acl_cache_locks : public MDL_release_locks_visitor {
 public:
  /**
    Lock releaser.
    Check details of given key and see it is of type ACL_CACHE
    and if key name it matches with m_partition. If so, release it.

    @param [in] ticket      MDL Ticket returned by MDL subsystem

    @returns Whether ticket matches our criteria or not
      @retval true  Ticket matches
      @retval false Ticket does not match
  */

  bool release(MDL_ticket *ticket) override {
    return ticket->get_key()->mdl_namespace() == MDL_key::ACL_CACHE;
  }
};

/**
  Acl_cache_lock_guard constructor.

  @param [in] thd             THD Handle.
  @param [in] mode            Lock mode

*/

Acl_cache_lock_guard::Acl_cache_lock_guard(THD *thd, Acl_cache_lock_mode mode)
    : m_thd(thd), m_mode(mode), m_locked(false) {
  assert(thd);
}

/**
  Explicitly take lock on Acl_cache_lock_cache object.
  If cache was already locked, just return.

  @param [in] raise_error  Whether to raise error if we fail to acquire lock

  @returns status of lock
    @retval true Lock was acquired/already acquired.
    @retval false There was some problem acquiring lock.
*/

bool Acl_cache_lock_guard::lock(bool raise_error) /* = true */
{
  assert(!m_locked);

  if (already_locked()) return true;

  /* If we do not have MDL, we should not be holding LOCK_open */
  mysql_mutex_assert_not_owner(&LOCK_open);

  MDL_request lock_request;
  MDL_REQUEST_INIT_BY_KEY(
      &lock_request, &ACL_CACHE_KEY,
      m_mode == Acl_cache_lock_mode::READ_MODE ? MDL_SHARED : MDL_EXCLUSIVE,
      MDL_EXPLICIT);
  Acl_cache_error_handler handler;
  m_thd->push_internal_handler(&handler);
  m_locked =
      !m_thd->mdl_context.acquire_lock(&lock_request, ACL_CACHE_LOCK_TIMEOUT);
  m_thd->pop_internal_handler();

  if (!m_locked && raise_error)
    my_error(ER_CANNOT_LOCK_USER_MANAGEMENT_CACHES, MYF(0));

  return m_locked;
}

/**
  Explicitly unlock all acquired locks.
*/

void Acl_cache_lock_guard::unlock() {
  /*
    It is possible that we did not take any lock. E.g.
    1. Function 1 : Take shared lock on ACL caches
    2. Function 1 : Take Lock_open
    3. Call Function 2
    4. Function 2 : Try to acquire shared lock on ACL
    caches again

    In such cases, at 4, we do not actually take
    MDL because it will be in conflict with LOCK_open

    If unlock() is called as part of destructor or
    directly in Function 2 above, we should
    not release any locks because we never acquired
    any in the first place.
  */
  if (m_locked) {
    Release_acl_cache_locks lock_visitor;
    m_thd->mdl_context.release_locks(&lock_visitor);
    m_locked = false;
  }
}

/**
  Check whether lock is already obtained or not.

  @returns lock status
    @retval true  Lock has already been taken.
    @retval false No lock is taken with this Acl_cache_lock_guard object
*/

bool Acl_cache_lock_guard::already_locked() {
  return m_thd->mdl_context.owns_equal_or_stronger_lock(
      MDL_key::ACL_CACHE, "", "",
      m_mode == Acl_cache_lock_mode::READ_MODE ? MDL_SHARED : MDL_EXCLUSIVE);
}

/**
  Assert that thread owns MDL_SHARED on partition specific to the thread

  @param [in] thd    Thread for which lock is to be checked

  @returns thread owns required lock or not
    @retval true    Thread owns lock
    @retval false   Thread does not own lock
*/

bool assert_acl_cache_read_lock(THD *thd) {
  return thd->mdl_context.owns_equal_or_stronger_lock(MDL_key::ACL_CACHE, "",
                                                      "", MDL_SHARED);
}

/**
  Assert that thread owns MDL_EXCLUSIVE on all partitions

  @param [in] thd    Thread for which lock is to be checked

  @returns thread owns required lock or not
    @retval true    Thread owns lock
    @retval false   Thread does not own lock
*/

bool assert_acl_cache_write_lock(THD *thd) {
  return thd->mdl_context.owns_equal_or_stronger_lock(MDL_key::ACL_CACHE, "",
                                                      "", MDL_EXCLUSIVE);
}

/** Global sysvar: the number of old passwords to check in the history. */
uint32 global_password_history = 0;
/** Global sysvar: the number of days before a password can be reused. */
uint32 global_password_reuse_interval = 0;

/**
  Reload all ACL caches

  We call this in two cases:
  1. FLUSH PRIVILEGES
  2. ACL DDL rollback

  @param [in] thd              THD handle
  @param [in] mdl_locked       MDL locks are taken
  @param [in] preserve_temporary_account_locking
                               Preserve temporary account locking
                               attributes of all users.
  @param [in] modified_user_lock_state_list
                               List of users whose temporary account
                               locking attributes are likely modified.

  @returns Status of reloading ACL caches
    @retval false Success
    @retval true Error
*/

bool reload_acl_caches(THD *thd, bool mdl_locked,
                       bool preserve_temporary_account_locking,
                       Lock_state_list *modified_user_lock_state_list) {
  DBUG_TRACE;
  bool retval = true;
  bool save_mdl_locked = mdl_locked;
  uint flags = MYSQL_LOCK_IGNORE_TIMEOUT;

  DBUG_EXECUTE_IF("wl14084_simulate_flush_privileges_timeout", flags = 0;);
  DEBUG_SYNC(thd, "wl14084_flush_privileges_before_table_locks");
  /*
    If mdl_locked is false, it implies we are here as a part
    of FLUSH PRIVILEGES. In such situation, it is useful to
    acquire and keep MDLs till the end of the function so that
    a concurrent ACL DDL does not interfere with cache reload
    operation.

    If mdl_locked is true, it would mean we are in process of
    reverting ACL DDL. This means we already have required
    MDLs and thus, no need to acquire MDLs.
  */
  if (!mdl_locked) {
    Table_ref tables[ACL_TABLES::LAST_ENTRY];
    acl_tables_setup_for_read(tables);
    /*
      Ideally, we can just call lock_table_names() here because all we want
      is to obtain MDL on ACL tables. However, if FLUSH PRIVILEGES is called
      under LOCK TABLES, we will hit an assertion that checks that
      lock_table_names() is not called under LOCK TABLES. The only exception
      to this rule RENAME TABLE. Hence, we call open_and_lock_tables()
      instead and make sure that proper cleanup is done in case we encounter
      an error.
    */
    bool result = open_and_lock_tables(thd, tables, flags);
    if (!result)
      close_thread_tables(thd);
    else {
      commit_and_close_mysql_tables(thd);
      return result;
    }
    mdl_locked = true;
    DEBUG_SYNC(thd, "wl14084_flush_privileges_after_table_locks");
  }

  if (check_engine_type_for_acl_table(thd, mdl_locked) ||
      check_acl_tables_intact(thd, mdl_locked) ||
      acl_reload(thd, mdl_locked, preserve_temporary_account_locking,
                 modified_user_lock_state_list) ||
      grant_reload(thd, mdl_locked)) {
    goto end;
  }
  // If there exists at least a partial revoke then update the sys variable
  check_and_update_partial_revokes_sysvar(thd);
  retval = false;
  DBUG_EXECUTE_IF("wl14084_trigger_acl_ddl_timeout", sleep(2););

end:
  /*
    When reload_acl_caches() is called with mdl_locked = true,
    it implies that caller has all required MDLs in place and
    thus, reload_acl_caches() need not go through process of
    obtained any MDLs.

    If mdl_locked = false when function is called, it means
    we would have acquired required MDLs to make sure that
    reload operations do not have compete for this resources
    with other threads. In such situation, release MDLs before
    returning to caller.

    In addition, since we pass mdl_locked = true to functions that
    reload caches, we would also have to commit the transaction.
  */
  if (!save_mdl_locked) {
    close_thread_tables(thd);
    commit_and_close_mysql_tables(thd);
  }
  return retval;
}

bool ACL_compare::operator()(const ACL_ACCESS &a, const ACL_ACCESS &b) {
  if (a.host.ip != 0) {
    if (b.host.ip != 0) {
      /* Both elements have specified IPs. The one with the greater mask goes
       * first. */
      if (a.host.ip_mask_type != b.host.ip_mask_type)
        return a.host.ip_mask_type < b.host.ip_mask_type;

      /* if masks are not equal compare these */
      if (a.host.ip_mask != b.host.ip_mask)
        return a.host.ip_mask > b.host.ip_mask;

      /* otherwise stick with the sort value */
      return a.sort > b.sort;
    }
    /* The element with the IP goes first. */
    return true;
  }

  /* The element with the IP goes first. */
  if (b.host.ip != 0) return false;

  /* None of the elements has IP defined. Use default comparison. */
  return a.sort > b.sort;
}

bool ACL_compare::operator()(const ACL_ACCESS *a, const ACL_ACCESS *b) {
  return this->operator()(*a, *b);
}

bool ACL_USER_compare::operator()(const ACL_USER &a, const ACL_USER &b) {
  if (a.host.ip != 0) {
    if (b.host.ip != 0) {
      /* Both elements have specified IPs. The one with the greater mask goes
       * first. */
      if (a.host.ip_mask_type != b.host.ip_mask_type)
        return a.host.ip_mask_type < b.host.ip_mask_type;

      if (a.host.ip_mask == b.host.ip_mask) return a.user > b.user;

      return a.host.ip_mask > b.host.ip_mask;
    }
    /* The element with the IP goes first. */
    return true;
  }

  /* The element with the IP goes first. */
  if (b.host.ip != 0) return false;

  /* None of the elements has IP defined. Use default comparison. */
  return a.sort > b.sort;
}

bool ACL_USER_compare::operator()(const ACL_USER *a, const ACL_USER *b) {
  return this->operator()(*a, *b);
}

/**
  Construstor
*/
Acl_restrictions::Acl_restrictions() : m_restrictions_map(key_memory_acl_mem) {}

/**
  Remove the Restrictions of the ACL_USER.

  @param [in] acl_user The ACL_USER for whom to remove the Restrictions
*/
void Acl_restrictions::remove_restrictions(const ACL_USER *acl_user) {
  assert(assert_acl_cache_write_lock(current_thd));
  const Auth_id auth_id(acl_user);
  auto itr = m_restrictions_map.find(auth_id.auth_str());
  if (itr != m_restrictions_map.end()) m_restrictions_map.erase(itr);
}

/**
  Update, insert or remove the Restrictions for the ACL_USER.

  If ACL_USER has a Restrictions
   - If specified Restrictions is not empty then update ACL_USER's
  Restrictions
   - Otherwise clear the ACL_USER's restriction
  Else if there no Restrictions for the ACL_USER then insert the specified
    Restrictions.

  @param [in] acl_user The ACL_USER for whom to alter the Restrictions
  @param [in] restrictions Restrictions to be inserted for the ACL_USER
*/
void Acl_restrictions::upsert_restrictions(const ACL_USER *acl_user,
                                           const Restrictions &restrictions) {
  assert(assert_acl_cache_write_lock(current_thd));
  const Auth_id auth_id(acl_user);
  const std::string auth_str = auth_id.auth_str();
  auto restrictions_itr = m_restrictions_map.find(auth_str);
  if (restrictions_itr != m_restrictions_map.end()) {
    if (restrictions.is_empty()) {
      /* Empty restrictions means we want to remove that from global cache */
      m_restrictions_map.erase(restrictions_itr);
    } else {
      /* If there exists restrictions then update that in the global cache */
      restrictions_itr->second = restrictions;
    }
  } else if (!restrictions.is_empty()) {
    /* Insert non-empty restrictions object in the global cache */
    m_restrictions_map.emplace(auth_str, restrictions);
  }
}

/**
  Find the Restrictions of the ACL_USER.

  @param [in]  acl_user The ACL_USER for whom to find the Restrictions

  @returns valid Restrictions if found otherwise empty Restrictions
*/
Restrictions Acl_restrictions::find_restrictions(
    const ACL_USER *acl_user) const {
  assert(assert_acl_cache_read_lock(current_thd));
  const Auth_id auth_id(acl_user);
  auto restrictions_itr = m_restrictions_map.find(auth_id.auth_str());
  if (restrictions_itr != m_restrictions_map.end())
    return restrictions_itr->second;
  else
    return Restrictions{};
}

/**
  @returns the number of Restrictions present. It is not thread safe method.
*/
size_t Acl_restrictions::size() const { return m_restrictions_map.size(); }

/**
  Method to check if there exists at least one partial revokes in the cache.
  If the cache is not initialized at the time of the method call then it
  returns no partial revokes exists.

  @param [in] thd THD handle

  @retval  true  Partial revokes exists
  @retval  false Otherwise
*/
bool is_partial_revoke_exists(THD *thd) {
  bool partial_revoke = false;
  assert(thd);
  Acl_cache_lock_guard acl_cache_lock(thd, Acl_cache_lock_mode::READ_MODE);
  // In case of failure assume that partial revokes exists to be on safe side.
  if (!acl_cache_lock.lock(false)) {
    return true;
  }
  /*
    Check the restrictions only if server has initialized the acl caches
    (i.e. Server is not started with --skip-grant-tables=1 option).
  */
  if (acl_restrictions) partial_revoke = (acl_restrictions->size() > 0);
  return partial_revoke;
}

bool is_acl_inited() { return acl_cache_initialized; }