/* Copyright (C) 2013, 2015, Alexey Botchkov and SkySQL Ab
   Copyright (c) 2019, 2020, MariaDB Corporation.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */


#define PLUGIN_VERSION 0x108
#define PLUGIN_STR_VERSION "1.8.0"

#include <my_config.h>
#include "my_counter.h"
#include <assert.h>
#include "sql_command.h"

#ifndef _WIN32
#define DO_SYSLOG
#include <syslog.h>

static const char out_type_desc[]= "Desired output type. Possible values - 'syslog', 'file'"
                                   " or 'null' as no output";
#else
static const char out_type_desc[]= "Desired output type. Possible values - 'file'"
                                   " or 'null' as no output";
#define syslog(PRIORITY, FORMAT, INFO, MESSAGE_LEN, MESSAGE) do {}while(0)
static void closelog() {}
#define openlog(IDENT, LOG_NOWAIT, LOG_USER)  do {}while(0)

/* priorities */
#define LOG_EMERG       0       /* system is unusable */
#define LOG_ALERT       1       /* action must be taken immediately */
#define LOG_CRIT        2       /* critical conditions */
#define LOG_ERR         3       /* error conditions */
#define LOG_WARNING     4       /* warning conditions */
#define LOG_NOTICE      5       /* normal but significant condition */
#define LOG_INFO        6       /* informational */
#define LOG_DEBUG       7       /* debug-level messages */

#define LOG_MAKEPRI(fac, pri)   (((fac) << 3) | (pri))

/* facility codes */
#define LOG_KERN        (0<<3)  /* kernel messages */
#define LOG_USER        (1<<3)  /* random user-level messages */
#define LOG_MAIL        (2<<3)  /* mail system */
#define LOG_DAEMON      (3<<3)  /* system daemons */
#define LOG_AUTH        (4<<3)  /* security/authorization messages */
#define LOG_SYSLOG      (5<<3)  /* messages generated internally by syslogd */
#define LOG_LPR         (6<<3)  /* line printer subsystem */
#define LOG_NEWS        (7<<3)  /* network news subsystem */
#define LOG_UUCP        (8<<3)  /* UUCP subsystem */
#define LOG_CRON        (9<<3)  /* clock daemon */
#define LOG_AUTHPRIV    (10<<3) /* security/authorization messages (private) */
#define LOG_FTP         (11<<3) /* ftp daemon */
#define LOG_LOCAL0      (16<<3) /* reserved for local use */
#define LOG_LOCAL1      (17<<3) /* reserved for local use */
#define LOG_LOCAL2      (18<<3) /* reserved for local use */
#define LOG_LOCAL3      (19<<3) /* reserved for local use */
#define LOG_LOCAL4      (20<<3) /* reserved for local use */
#define LOG_LOCAL5      (21<<3) /* reserved for local use */
#define LOG_LOCAL6      (22<<3) /* reserved for local use */
#define LOG_LOCAL7      (23<<3) /* reserved for local use */

#endif /*!_WIN32*/

#include <my_global.h>
#include <my_base.h>
#include <my_sys.h>
#include <typelib.h>
#include <mysql/plugin.h>
#include <mysql/plugin_audit.h>
#include <string.h>
#include <mysql/service_logger.h>
#include "../../mysys/mysys_priv.h"
#ifndef RTLD_DEFAULT
#define RTLD_DEFAULT NULL
#endif

#ifndef HOSTNAME_LENGTH
#define HOSTNAME_LENGTH 255
#endif
#ifndef USERNAME_CHAR_LENGTH
#define USERNAME_CHAR_LENGTH 128
#endif
#define TIMESTAMP_FORMAT_LENGTH 128
#define TIMESTAMP_OUTPUT_LENGTH 256
#define MAX_AUDIT_PAYLOAD_LENGTH 1024
#define MAX_AUDIT_QUERY_LENGTH 2048

#ifndef DBUG_OFF
#define PLUGIN_DEBUG_VERSION "-debug"
#else
#define PLUGIN_DEBUG_VERSION ""
#endif /*DBUG_OFF*/
#ifdef _WIN32
#define localtime_r(a, b) localtime_s(b, a)
#endif /*WIN32*/


const char *(*thd_priv_host_ptr)(MYSQL_THD thd, size_t *length);
static char *incl_users, *excl_users,
            *file_path, *syslog_info, *timestamp_format;
static char path_buffer[FN_REFLEN];
static char timestamp_format_buffer[TIMESTAMP_FORMAT_LENGTH];
static unsigned int mode;
static ulong output_type;
static ulong syslog_facility, syslog_priority;

static ulonglong events; /* mask for events to log */
static unsigned long long file_rotate_size;
static unsigned int rotations;
static my_bool rotate= TRUE;
static my_bool sync_file= TRUE;
static unsigned int file_buffer_size;
static char logging;
static Atomic_counter<int> internal_stop_logging;
static char incl_user_buffer[1024];
static char excl_user_buffer[1024];
static unsigned int query_log_limit= 0;

static char servhost[HOSTNAME_LENGTH+1];
static uint servhost_len;
static char syslog_ident[128]= "mysql-server_auditing";
static char *syslog_ident_ptr= syslog_ident;

struct connection_info
{
  int header;
  unsigned long thread_id;
  unsigned long long query_id;
  char db[256];
  int db_length;
  char user[USERNAME_CHAR_LENGTH+1];
  int user_length;
  char host[HOSTNAME_LENGTH+1];
  int host_length;
  char ip[64];
  int ip_length;
  char tls_version[64];
  int tls_version_length;
  const char *query;
  int query_length;
  char query_buffer[1024];
  int log_always;
  unsigned int port;
  char proxy[USERNAME_CHAR_LENGTH+1];
  int proxy_length;
  char proxy_host[HOSTNAME_LENGTH+1];
  int proxy_host_length;
  /*
    This is used to have the SET GLOBAL ...buffer_sync=1 statement itself
    written into the file with this flush.
  */
  int sync_statement;
};

#define DEFAULT_FILENAME_LEN 16
static char default_file_name[DEFAULT_FILENAME_LEN+1]= "server_audit.log";

static void update_file_path(MYSQL_THD thd, struct st_mysql_sys_var *var,
                             void *var_ptr, const void *save);
static void update_file_rotate_size(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                    void *var_ptr, const void *save);
static void update_file_rotations(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                  void *var_ptr, const void *save);
static void update_file_buffer_size(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                    void *var_ptr, const void *save);
static void update_incl_users(MYSQL_THD thd, struct st_mysql_sys_var *var,
                              void *var_ptr, const void *save);
static int check_incl_users(MYSQL_THD thd, struct st_mysql_sys_var *var, void *save,
                            struct st_mysql_value *value);
static int check_excl_users(MYSQL_THD thd, struct st_mysql_sys_var *var, void *save,
                            struct st_mysql_value *value);
static void update_excl_users(MYSQL_THD thd, struct st_mysql_sys_var *var,
                              void *var_ptr, const void *save);
static void update_output_type(MYSQL_THD thd, struct st_mysql_sys_var *var,
                               void *var_ptr, const void *save);
static void update_syslog_facility(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                   void *var_ptr, const void *save);
static void update_syslog_priority(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                   void *var_ptr, const void *save);
static void update_mode(MYSQL_THD thd, struct st_mysql_sys_var *var,
                        void *var_ptr, const void *save);
static void update_logging(MYSQL_THD thd, struct st_mysql_sys_var *var,
                           void *var_ptr, const void *save);
static void update_syslog_ident(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                void *var_ptr, const void *save);
static void rotate_log(MYSQL_THD thd, struct st_mysql_sys_var *var,
                       void *var_ptr, const void *save);
static int check_timestamp_format(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                  void *save, struct st_mysql_value *value);
static void update_timestamp_format(MYSQL_THD thd, struct st_mysql_sys_var *var,
                                    void *var_ptr, const void *save);
static void sync_log(MYSQL_THD thd, struct st_mysql_sys_var *var,
                     void *var_ptr, const void *save);

static MYSQL_SYSVAR_STR(incl_users, incl_users, PLUGIN_VAR_RQCMDARG,
       "Comma separated list of users to monitor",
       check_incl_users, update_incl_users, NULL);
static MYSQL_SYSVAR_STR(excl_users, excl_users, PLUGIN_VAR_RQCMDARG,
       "Comma separated list of users to exclude from auditing",
       check_excl_users, update_excl_users, NULL);
/* bits in the event filter. */
#define EVENT_CONNECT 1
#define EVENT_QUERY_ALL 2
#define EVENT_QUERY 122
#define EVENT_TABLE 4
#define EVENT_QUERY_DDL 8
#define EVENT_QUERY_DML 16
#define EVENT_QUERY_DCL 32
#define EVENT_QUERY_DML_NO_SELECT 64

static const char *event_names[]=
{
  "CONNECT", "QUERY", "TABLE", "QUERY_DDL", "QUERY_DML", "QUERY_DCL",
  "QUERY_DML_NO_SELECT", NULL
};
static TYPELIB events_typelib= CREATE_TYPELIB_FOR(event_names);
static MYSQL_SYSVAR_SET(events, events, PLUGIN_VAR_RQCMDARG,
       "Specifies the set of events to monitor. Can be CONNECT, QUERY, TABLE,"
           " QUERY_DDL, QUERY_DML, QUERY_DML_NO_SELECT, QUERY_DCL",
       NULL, NULL, 0, &events_typelib);
#ifdef DO_SYSLOG
#define OUTPUT_SYSLOG 0
#define OUTPUT_FILE 1
#else
#define OUTPUT_SYSLOG 0xFFFF
#define OUTPUT_FILE 0
#endif /*DO_SYSLOG*/

#define OUTPUT_NO 0xFFFF
static const char *output_type_names[]= {
#ifdef DO_SYSLOG
  "syslog",
#endif
  "file", 0 };
static TYPELIB output_typelib=CREATE_TYPELIB_FOR(output_type_names);
static MYSQL_SYSVAR_ENUM(output_type, output_type, PLUGIN_VAR_RQCMDARG,
       out_type_desc,
       0, update_output_type, OUTPUT_FILE,
       &output_typelib);
static MYSQL_SYSVAR_STR(file_path, file_path, PLUGIN_VAR_RQCMDARG,
       "Path to the log file", NULL, update_file_path, default_file_name);
static MYSQL_SYSVAR_ULONGLONG(file_rotate_size, file_rotate_size,
       PLUGIN_VAR_RQCMDARG, "Rotate the log when it grows larger than that",
       NULL, update_file_rotate_size,
       1000000, 100, ((long long) 0x7FFFFFFFFFFFFFFFLL), 1);
static MYSQL_SYSVAR_UINT(file_rotations, rotations,
       PLUGIN_VAR_RQCMDARG, "Number of rotations before log is removed",
       NULL, update_file_rotations, 9, 0, 999, 1);
static MYSQL_SYSVAR_UINT(file_buffer_size, file_buffer_size,
       PLUGIN_VAR_RQCMDARG, "Size of file buffer to make logging faster",
       NULL, update_file_buffer_size, 0, 0, 65536, 8192);
static MYSQL_SYSVAR_BOOL(file_rotate_now, rotate, PLUGIN_VAR_OPCMDARG,
       "Force log rotation now", NULL, rotate_log, FALSE);
static MYSQL_SYSVAR_BOOL(sync_log_file, sync_file, PLUGIN_VAR_OPCMDARG,
       "Force sync log file", NULL, sync_log, FALSE);
static MYSQL_SYSVAR_BOOL(logging, logging,
       PLUGIN_VAR_OPCMDARG, "Turn on/off the logging", NULL,
       update_logging, 0);
static MYSQL_SYSVAR_UINT(mode, mode,
       PLUGIN_VAR_OPCMDARG, "Auditing mode", NULL, update_mode, 0, 0, 1, 1);
static MYSQL_SYSVAR_STR(syslog_ident, syslog_ident_ptr, PLUGIN_VAR_RQCMDARG,
       "The SYSLOG identifier - the beginning of each SYSLOG record",
       NULL, update_syslog_ident, syslog_ident);
static MYSQL_SYSVAR_STR(syslog_info, syslog_info,
       PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_MEMALLOC,
       "The <info> string to be added to the SYSLOG record", NULL, NULL, "");
static MYSQL_SYSVAR_UINT(query_log_limit, query_log_limit,
       PLUGIN_VAR_OPCMDARG, "Limit on the length of the query string in a record",
       NULL, NULL, 1024, 0, 0x7FFFFFFF, 1);
static MYSQL_SYSVAR_STR(timestamp_format, timestamp_format, 
       PLUGIN_VAR_RQCMDARG,
       "A format string used to print the timestamp into the audit log messages "
       "The format used is the same as DATE_FORMAT",
       check_timestamp_format, update_timestamp_format, "%Y%m%d %H:%i:%s");

char locinfo_ini_value[sizeof(struct connection_info)+4];

static MYSQL_THDVAR_STR(loc_info,
                        PLUGIN_VAR_NOSYSVAR | PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC,
                        "Internal info", NULL, NULL, locinfo_ini_value);

static const char *syslog_facility_names[]=
{
  "LOG_USER", "LOG_MAIL", "LOG_DAEMON", "LOG_AUTH",
  "LOG_SYSLOG", "LOG_LPR", "LOG_NEWS", "LOG_UUCP",
  "LOG_CRON",
#ifdef LOG_AUTHPRIV
 "LOG_AUTHPRIV",
#endif
#ifdef LOG_FTP
 "LOG_FTP",
#endif
  "LOG_LOCAL0", "LOG_LOCAL1", "LOG_LOCAL2", "LOG_LOCAL3",
  "LOG_LOCAL4", "LOG_LOCAL5", "LOG_LOCAL6", "LOG_LOCAL7",
  0
};
#ifndef _WIN32
static unsigned int syslog_facility_codes[]=
{
  LOG_USER, LOG_MAIL, LOG_DAEMON, LOG_AUTH,
  LOG_SYSLOG, LOG_LPR, LOG_NEWS, LOG_UUCP,
  LOG_CRON,
#ifdef LOG_AUTHPRIV
 LOG_AUTHPRIV,
#endif
#ifdef LOG_FTP
  LOG_FTP,
#endif
  LOG_LOCAL0, LOG_LOCAL1, LOG_LOCAL2, LOG_LOCAL3,
  LOG_LOCAL4, LOG_LOCAL5, LOG_LOCAL6, LOG_LOCAL7,
};
#endif
static TYPELIB syslog_facility_typelib=CREATE_TYPELIB_FOR(syslog_facility_names);
static MYSQL_SYSVAR_ENUM(syslog_facility, syslog_facility, PLUGIN_VAR_RQCMDARG,
       "The 'facility' parameter of the SYSLOG record."
       " The default is LOG_USER", 0, update_syslog_facility, 0/*LOG_USER*/,
       &syslog_facility_typelib);

static const char *syslog_priority_names[]=
{
  "LOG_EMERG", "LOG_ALERT", "LOG_CRIT", "LOG_ERR",
  "LOG_WARNING", "LOG_NOTICE", "LOG_INFO", "LOG_DEBUG",
  0
};

#ifndef _WIN32
static unsigned int syslog_priority_codes[]=
{
  LOG_EMERG, LOG_ALERT, LOG_CRIT, LOG_ERR,
  LOG_WARNING, LOG_NOTICE, LOG_INFO, LOG_DEBUG,
};
#endif

static TYPELIB syslog_priority_typelib=CREATE_TYPELIB_FOR(syslog_priority_names);
static MYSQL_SYSVAR_ENUM(syslog_priority, syslog_priority, PLUGIN_VAR_RQCMDARG,
       "The 'priority' parameter of the SYSLOG record."
       " The default is LOG_INFO", 0, update_syslog_priority, 6/*LOG_INFO*/,
       &syslog_priority_typelib);


static struct st_mysql_sys_var* vars[] = {
    MYSQL_SYSVAR(incl_users),
    MYSQL_SYSVAR(excl_users),
    MYSQL_SYSVAR(events),
    MYSQL_SYSVAR(output_type),
    MYSQL_SYSVAR(file_buffer_size),
    MYSQL_SYSVAR(file_path),
    MYSQL_SYSVAR(file_rotate_size),
    MYSQL_SYSVAR(file_rotations),
    MYSQL_SYSVAR(file_rotate_now),
    MYSQL_SYSVAR(logging),
    MYSQL_SYSVAR(mode),
    MYSQL_SYSVAR(sync_log_file),
    MYSQL_SYSVAR(syslog_info),
    MYSQL_SYSVAR(syslog_ident),
    MYSQL_SYSVAR(syslog_facility),
    MYSQL_SYSVAR(syslog_priority),
    MYSQL_SYSVAR(query_log_limit),
    MYSQL_SYSVAR(timestamp_format),
    MYSQL_SYSVAR(loc_info),
    NULL
};


/* Status variables for SHOW STATUS */
static int is_active= 0;
static Atomic_counter<long> log_write_failures;
static char current_log_buf[FN_REFLEN]= "";
static char last_error_buf[512]= "";

static struct st_mysql_show_var audit_status[]=
{
  {"server_audit_active", (char *)&is_active, SHOW_BOOL},
  {"server_audit_current_log", current_log_buf, SHOW_CHAR},
  {"server_audit_writes_failed", (char *)&log_write_failures, SHOW_LONG},
  {"server_audit_last_error", last_error_buf, SHOW_CHAR},
  {0,0,SHOW_UNDEF}
};

#include "../../storage/innobase/include/srw_lock.h"

static srw_lock_low lock_operations;

#define CLIENT_ERROR my_printf_error


static uchar *getkey_user(const char *entry, size_t *length, my_bool)
{
  const char *e= entry;
  while (*e && *e != ' ' && *e != ',')
    ++e;
  *length= e - entry;
  return (uchar *) entry;
}


static void blank_user(char *user)
{
  for (; *user && *user != ','; user++)
    *user= ' ';
}


static void remove_user(char *user)
{
  char *start_user= user;
  while (*user != ',')
  {
    if (*user == 0)
    {
      *start_user= 0;
      return;
    }
    user++;
  }
  user++;
  while (*user == ' ')
    user++;

  do {
    *(start_user++)= *user;
  } while (*(user++));
}


static void remove_blanks(char *user)
{
  char *user_orig= user;
  char *user_to= user;
  char *start_tok;
  int blank_name;

  while (*user != 0)
  {
    start_tok= user;
    blank_name= 1;
    while (*user !=0 && *user != ',')
    {
      if (*user != ' ')
        blank_name= 0;
      user++;
    }
    if (!blank_name)
    {
      while (start_tok <= user)
        *(user_to++)= *(start_tok++);
    }
    if (*user == ',')
      user++;
  }
  if (user_to > user_orig && user_to[-1] == ',')
    user_to--;
  *user_to= 0;
}


struct user_name
{
  size_t name_len;
  char *name;
};


struct user_coll
{
  int n_users;
  struct user_name *users;
  int n_alloced;
};


static void coll_init(struct user_coll *c)
{
  c->n_users= 0;
  c->users= 0;
  c->n_alloced= 0;
}


static void coll_free(struct user_coll *c)
{
  if (c->users)
  {
    free(c->users);
    coll_init(c);
  }
}


static int cmp_users(const void *ia, const void *ib)
{
  const struct user_name *a= (const struct user_name *) ia;
  const struct user_name *b= (const struct user_name *) ib;
  int dl= (int)(a->name_len - b->name_len);
  if (dl != 0)
    return dl;

  return strncmp(a->name, b->name, a->name_len);
}


static char *coll_search(struct user_coll *c, const char *n, size_t len)
{
  struct user_name un;
  struct user_name *found;
  if (!c->n_users)
    return 0;
  un.name_len= len;
  un.name= (char *) n;
  found= (struct user_name*)  bsearch(&un, c->users, c->n_users,
                                      sizeof(c->users[0]), cmp_users);
  return found ? found->name : 0;
}


static int coll_insert(struct user_coll *c, char *n, size_t len)
{
  if (c->n_users >= c->n_alloced)
  {
    c->n_alloced+= 128;
    const size_t size{c->n_alloced * sizeof *c->users};
    c->users= static_cast<user_name*>
      (c->users ? realloc(c->users, size) : malloc(size));

    if (c->users == NULL)
      return 1;
  }
  c->users[c->n_users].name= n;
  c->users[c->n_users].name_len= len;
  c->n_users++;
  return 0;
}


static void coll_sort(struct user_coll *c)
{
  if (c->n_users)
    qsort(c->users, c->n_users, sizeof(c->users[0]), cmp_users);
}


static int user_coll_fill(struct user_coll *c, char *users,
                          struct user_coll *cmp_c, int take_over_cmp)
{
  char *orig_users= users;
  char *cmp_user= 0;
  size_t cmp_length;
  int refill_cmp_coll= 0;

  c->n_users= 0;

  while (*users)
  {
    while (*users == ' ')
      users++;
    if (!*users)
      return 0;

    (void) getkey_user(users, &cmp_length, FALSE);
    if (cmp_c)
    {
      cmp_user= coll_search(cmp_c, users, cmp_length);

      if (cmp_user && take_over_cmp)
      {
        internal_stop_logging++;
        CLIENT_ERROR(1, "User '%.*sB' was removed from the"
            " server_audit_excl_users.",
            MYF(ME_WARNING), (int) cmp_length, users);
        internal_stop_logging--;
        blank_user(cmp_user);
        refill_cmp_coll= 1;
      }
      else if (cmp_user)
      {
        internal_stop_logging++;
        CLIENT_ERROR(1, "User '%.*sB' is in the server_audit_incl_users, "
            "so wasn't added.", MYF(ME_WARNING), (int) cmp_length, users);
        internal_stop_logging--;
        remove_user(users);
        continue;
      }
    }
    if (coll_insert(c, users, cmp_length))
      return 1;
    while (*users && *users != ',')
      users++;
    if (!*users)
      break;
    users++;
  }

  if (refill_cmp_coll)
  {
    remove_blanks(excl_users);
    return user_coll_fill(cmp_c, excl_users, 0, 0);
  }

  if (users > orig_users && users[-1] == ',')
    users[-1]= 0;

  coll_sort(c);

  return 0;
}


static void error_header()
{
  struct tm tm_time;
  time_t curtime;

  (void) time(&curtime);
  (void) localtime_r(&curtime, &tm_time);

  (void) fprintf(stderr,"%02d%02d%02d %2d:%02d:%02d server_audit: ",
    tm_time.tm_year % 100, tm_time.tm_mon + 1,
    tm_time.tm_mday, tm_time.tm_hour, tm_time.tm_min, tm_time.tm_sec);
}


static LOGGER_HANDLE *logfile;
static struct user_coll incl_user_coll, excl_user_coll;
static unsigned long long query_counter= 1;


static struct connection_info *get_loc_info(MYSQL_THD thd)
{
  /*
    This is the original code and supposed to be returned
    bach to this as the MENT-1438 is finally understood/resolved.
  return (struct connection_info *) THDVAR(thd, loc_info);
  */
  struct connection_info *ci= (struct connection_info *) THDVAR(thd, loc_info);
  /* THDVAR can return NULL if the plugin was previously uninstalled (MDEV-34074). */
  if (ci && (size_t) ci->user_length > sizeof(ci->user))
  {
    ci->user_length= 0;
    ci->host_length= 0;
    ci->ip_length= 0;
    ci->tls_version_length= 0;
  }
  return ci;
}


static int ci_needs_setup(const struct connection_info *ci)
{
  return ci->header != 0;
}


static void get_str_n(char *dest, int *dest_len, size_t dest_size,
                      const char *src, size_t src_len)
{
  if (src_len >= dest_size)
    src_len= dest_size - 1;

  if (src_len)
    memcpy(dest, src, src_len);
  dest[src_len]= 0;
  *dest_len= (int)src_len;
}


static int get_user_host(const char *uh_line, unsigned int uh_len,
                         char *buffer, size_t buf_len,
                         size_t *user_len, size_t *host_len, size_t *ip_len)
{
  const char *buf_end= buffer + buf_len - 1;
  const char *buf_start;
  const char *uh_end= uh_line + uh_len;

  while (uh_line < uh_end && *uh_line != '[')
    ++uh_line;

  if (uh_line == uh_end)
    return 1;
  ++uh_line;

  buf_start= buffer;
  while (uh_line < uh_end && *uh_line != ']')
  {
    if (buffer == buf_end)
      return 1;
    *(buffer++)= *(uh_line++);
  }
  if (uh_line == uh_end)
    return 1;
  *user_len= buffer - buf_start;
  *(buffer++)= 0;

  while (uh_line < uh_end && *uh_line != '@')
    ++uh_line;
  if (uh_line == uh_end || *(++uh_line) == 0)
    return 1;
  ++uh_line;

  buf_start= buffer;
  while (uh_line < uh_end && *uh_line != ' ' && *uh_line != '[')
  {
    if (buffer == buf_end)
      break;
    *(buffer++)= *(uh_line++);
  }
  *host_len= buffer - buf_start;
  *(buffer++)= 0;

  while (uh_line < uh_end && *uh_line != '[')
    ++uh_line;

  buf_start= buffer;
  if (*uh_line == '[')
  {
    ++uh_line;
    while (uh_line < uh_end && *uh_line != ']')
      *(buffer++)= *(uh_line++);
  }
  *ip_len= buffer - buf_start;
  return 0;
}


static int sql_command_to_cmdtype(int sql_command)
{
  switch (sql_command)
  {
    case SQLCOM_ALTER_DB:
    case SQLCOM_ALTER_DB_UPGRADE:
    case SQLCOM_ALTER_FUNCTION:
    case SQLCOM_ALTER_PROCEDURE:
    case SQLCOM_ALTER_SEQUENCE:
    case SQLCOM_ALTER_TABLE:
    case SQLCOM_CREATE_DB:
    case SQLCOM_CREATE_INDEX:
    case SQLCOM_CREATE_PACKAGE:
    case SQLCOM_CREATE_PACKAGE_BODY:
    case SQLCOM_CREATE_PROCEDURE:
    case SQLCOM_CREATE_SEQUENCE:
    case SQLCOM_CREATE_SPFUNCTION:
    case SQLCOM_CREATE_TABLE:
    case SQLCOM_CREATE_TRIGGER:
    case SQLCOM_CREATE_VIEW:
    case SQLCOM_DROP_DB:
    case SQLCOM_DROP_FUNCTION:
    case SQLCOM_DROP_INDEX:
    case SQLCOM_DROP_PACKAGE:
    case SQLCOM_DROP_PACKAGE_BODY:
    case SQLCOM_DROP_PROCEDURE:
    case SQLCOM_DROP_SEQUENCE:
    case SQLCOM_DROP_TABLE:
    case SQLCOM_DROP_TRIGGER:
    case SQLCOM_DROP_VIEW:
    case SQLCOM_RENAME_TABLE:
    case SQLCOM_TRUNCATE:
      return EVENT_QUERY_DDL;

    case SQLCOM_DO:
    case SQLCOM_CALL:
    case SQLCOM_LOAD:
    case SQLCOM_DELETE:
    case SQLCOM_DELETE_MULTI:
    case SQLCOM_INSERT:
    case SQLCOM_INSERT_SELECT:
    case SQLCOM_UPDATE:
    case SQLCOM_UPDATE_MULTI:
    case SQLCOM_HA_OPEN:
    case SQLCOM_HA_READ:
    case SQLCOM_HA_CLOSE:
    case SQLCOM_REPLACE:
    case SQLCOM_REPLACE_SELECT:
      return EVENT_QUERY_DML | EVENT_QUERY_DML_NO_SELECT;

    case SQLCOM_SELECT:
      return EVENT_QUERY_DML;

    case SQLCOM_CREATE_USER:
    case SQLCOM_CREATE_ROLE:
    case SQLCOM_DROP_ROLE:
    case SQLCOM_DROP_USER:
    case SQLCOM_RENAME_USER:
    case SQLCOM_GRANT:
    case SQLCOM_GRANT_ROLE:
    case SQLCOM_REVOKE:
    case SQLCOM_REVOKE_ROLE:
    case SQLCOM_REVOKE_ALL:
      return EVENT_QUERY_DCL;

    default:
      return EVENT_QUERY_ALL;
  };
}


#if defined(_WIN32) && !defined(S_ISDIR)
#define S_ISDIR(x) ((x) & _S_IFDIR)
#endif /*_WIN32 && !S_ISDIR*/

static int start_logging()
{
  last_error_buf[0]= 0;
  log_write_failures= 0;
  if (output_type == OUTPUT_FILE)
  {
    char alt_path_buffer[FN_REFLEN+1+DEFAULT_FILENAME_LEN];
    struct stat *f_stat= (struct stat *)alt_path_buffer;
    const char *alt_fname= file_path;

    while (*alt_fname == ' ')
      alt_fname++;

    if (*alt_fname == 0)
    {
      /* Empty string means the default file name. */
      alt_fname= default_file_name;
    }
    else
    {
      /* See if the directory exists with the name of file_path.    */
      /* Log file name should be [file_path]/server_audit.log then. */
      if (stat(file_path, (struct stat *)alt_path_buffer) == 0 &&
          S_ISDIR(f_stat->st_mode))
      {
        size_t p_len= strlen(file_path);
        memcpy(alt_path_buffer, file_path, p_len);
        if (alt_path_buffer[p_len-1] != FN_LIBCHAR)
        {
          alt_path_buffer[p_len]= FN_LIBCHAR;
          p_len++;
        }
        memcpy(alt_path_buffer+p_len, default_file_name, DEFAULT_FILENAME_LEN);
        alt_path_buffer[p_len+DEFAULT_FILENAME_LEN]= 0;
        alt_fname= alt_path_buffer;
      }
    }

    logfile= logger_open(alt_fname, file_rotate_size,
                         rotations, file_buffer_size);

    if (logfile == NULL)
    {
      error_header();
      fprintf(stderr, "Could not create file '%s'.\n",
              alt_fname);
      logging= 0;
      my_snprintf(last_error_buf, sizeof(last_error_buf),
                  "Could not create file '%s'.", alt_fname);
      is_active= 0;
      CLIENT_ERROR(1, "SERVER AUDIT plugin can't create file '%s'.",
          MYF(ME_WARNING), alt_fname);
      return 1;
    }
    error_header();
    fprintf(stderr, "logging started to the file %s.\n", alt_fname);
    strncpy(current_log_buf, alt_fname, sizeof(current_log_buf)-1);
    current_log_buf[sizeof(current_log_buf)-1]= 0;
  }
  else if (output_type == OUTPUT_SYSLOG)
  {
    openlog(syslog_ident, LOG_NOWAIT, syslog_facility_codes[syslog_facility]);
    error_header();
    fprintf(stderr, "logging started to the syslog.\n");
    strncpy(current_log_buf, "[SYSLOG]", sizeof(current_log_buf)-1);
    compile_time_assert(sizeof current_log_buf > sizeof "[SYSLOG]");
  }
  is_active= 1;
  return 0;
}


static int stop_logging()
{
  last_error_buf[0]= 0;
  if (output_type == OUTPUT_FILE && logfile)
  {
    logger_close(logfile);
    logfile= NULL;
  }
  else if (output_type == OUTPUT_SYSLOG)
  {
    closelog();
  }
  error_header();
  fprintf(stderr, "logging was stopped.\n");
  is_active= 0;
  return 0;
}


static void setup_connection_simple(struct connection_info *ci)
{
  ci->db_length= 0;
  ci->user_length= 0;
  ci->host_length= 0;
  ci->ip_length= 0;
  ci->tls_version_length= 0;
  ci->query_length= 0;
  ci->header= 0;
  ci->proxy_length= 0;
  ci->port= 0;
}


#define MAX_HOSTNAME  (HOSTNAME_LENGTH + 1) /* len+1 in mysql.user */

static void setup_connection_connect(MYSQL_THD thd,struct connection_info *cn,
    const struct mysql_event_connection *event)
{
  cn->query_id= 0;
  cn->query_length= 0;
  cn->log_always= 0;
  cn->sync_statement= 0;
  cn->thread_id= event->thread_id;
  get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
            event->database.str, event->database.length);
  get_str_n(cn->user, &cn->user_length, sizeof(cn->db),
            event->user, event->user_length);
  get_str_n(cn->host, &cn->host_length, sizeof(cn->host),
            event->host, event->host_length);
  get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
            event->ip, event->ip_length);
  get_str_n(cn->tls_version, &cn->tls_version_length, sizeof(cn->tls_version),
          event->tls_version, event->tls_version_length);
  cn->header= 0;
  if (event->proxy_user && event->proxy_user[0])
  {
    const char *priv_host;
    size_t priv_host_length;

    if (thd_priv_host_ptr)
    {
      priv_host= (*thd_priv_host_ptr)(thd, &priv_host_length);
    }
    else
    {
      // 5 is "'" around host and user and "@"
      priv_host= event->proxy_user +
            sizeof(char[MAX_HOSTNAME + USERNAME_LENGTH + 5]);
      priv_host_length= strlen(priv_host);
    }


    get_str_n(cn->proxy, &cn->proxy_length, sizeof(cn->proxy),
              event->priv_user, event->priv_user_length);
    get_str_n(cn->proxy_host, &cn->proxy_host_length,
              sizeof(cn->proxy_host),
              priv_host, priv_host_length);
  }
  else
    cn->proxy_length= 0;
}


#define SAFE_STRLEN(s) (s ? strlen(s) : 0)
#define SAFE_STRLEN_UI(s) ((unsigned int) (s ? strlen(s) : 0))
static char empty_str[1]= { 0 };


static int is_space(char c)
{
  return c == ' ' || c == '\r' || c == '\n' || c == '\t';
}


#define SKIP_SPACES(str) \
do { \
  while (is_space(*str)) \
    ++str; \
} while(0)


#define ESC_MAP_SIZE 0x60
static const char esc_map[ESC_MAP_SIZE]=
{
  0, 0, 0, 0, 0, 0, 0, 0, 'b', 't', 'n', 0, 'f', 'r', 0, 0,
  0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
  0, 0, 0, 0, 0, 0, 0, '\'', 0, 0, 0, 0, 0, 0, 0, 0,
  0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
  0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
  0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '\\', 0, 0, 0
};

static char escaped_char(char c)
{
  return ((unsigned char ) c) >= ESC_MAP_SIZE ? 0 : esc_map[(unsigned char) c];
}


static void setup_connection_initdb(struct connection_info *cn,
    const struct mysql_event_general *event)
{
  size_t user_len, host_len, ip_len;
  char uh_buffer[512];

  cn->thread_id= event->general_thread_id;
  cn->query_id= 0;
  cn->sync_statement= 0;
  cn->query_length= 0;
  cn->log_always= 0;
  get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
            event->general_query, event->general_query_length);

  if (get_user_host(event->general_user, event->general_user_length,
                    uh_buffer, sizeof(uh_buffer),
                    &user_len, &host_len, &ip_len))
  {
    /* The user@host line is incorrect. */
    cn->user_length= 0;
    cn->host_length= 0;
    cn->ip_length= 0;
  }
  else
  {
    get_str_n(cn->user, &cn->user_length, sizeof(cn->user),
              uh_buffer, user_len);
    get_str_n(cn->host, &cn->host_length, sizeof(cn->host),
              uh_buffer+user_len+1, host_len);
    get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
              uh_buffer+user_len+1+host_len+1, ip_len);
  }
  cn->header= 0;
}


static void setup_connection_table(struct connection_info *cn,
    const struct mysql_event_table *event)
{
  cn->thread_id= event->thread_id;
  cn->query_id= query_counter++;
  cn->sync_statement= 0;
  cn->log_always= 0;
  cn->query_length= 0;
  get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
            event->database.str, event->database.length);
  get_str_n(cn->user, &cn->user_length, sizeof(cn->db),
            event->user, SAFE_STRLEN(event->user));
  get_str_n(cn->host, &cn->host_length, sizeof(cn->host),
            event->host, SAFE_STRLEN(event->host));
  get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
            event->ip, SAFE_STRLEN(event->ip));
  cn->header= 0;
}


static void setup_connection_query(struct connection_info *cn,
    const struct mysql_event_general *event)
{
  size_t user_len, host_len, ip_len;
  char uh_buffer[512];

  cn->thread_id= event->general_thread_id;
  cn->sync_statement= 0;
  cn->query_id= query_counter++;
  cn->log_always= 0;
  cn->query_length= 0;
  get_str_n(cn->db, &cn->db_length, sizeof(cn->db), "", 0);

  if (get_user_host(event->general_user, event->general_user_length,
                    uh_buffer, sizeof(uh_buffer),
                    &user_len, &host_len, &ip_len))
  {
    /* The user@host line is incorrect. */
    cn->user_length= 0;
    cn->host_length= 0;
    cn->ip_length= 0;
  }
  else
  {
    get_str_n(cn->user, &cn->user_length, sizeof(cn->user),
              uh_buffer, user_len);
    get_str_n(cn->host, &cn->host_length, sizeof(cn->host),
              uh_buffer+user_len+1, host_len);
    get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
              uh_buffer+user_len+1+host_len+1, ip_len);
  }
  cn->header= 0;
}


static void change_connection(struct connection_info *cn,
    const struct mysql_event_connection *event)
{
  get_str_n(cn->user, &cn->user_length, sizeof(cn->user),
            event->user, event->user_length);
  get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
            event->ip, event->ip_length);
  get_str_n(cn->tls_version, &cn->tls_version_length, sizeof(cn->tls_version),
          event->tls_version, event->tls_version_length);
}

/**
  Write to the log
*/

static int write_log(const char *message, size_t len, unsigned long long ts_us)
{
#if defined _WIN32 || !defined SUX_LOCK_GENERIC
  DBUG_ASSERT(lock_operations.is_locked_or_waiting());
#endif
  int result= 0;

  if (output_type == OUTPUT_FILE)
  {
    if (logfile)
    {
      MYSQL_TIME ltime;
      thd_gmt_sec_to_TIME(NULL, <ime, (time_t) (ts_us / 1000000));
      ltime.second_part= (ulong) (ts_us % 1000000);

      size_t ts_len= 0;
      char *ts_start= (char *) message - TIMESTAMP_OUTPUT_LENGTH;

      /*
        Format the timestamp into the start of the gap. We use
        TIMESTAMP_OUTPUT_LENGTH - 1 as the limit to ensure there is always at
        least one byte available at the end of the gap for the trailing comma,
        even if the formatted timestamp takes up the maximum allowed space
        (including its null terminator).
      */
      thd_TIME_to_str(NULL, <ime, timestamp_format_buffer,
                      ts_start, TIMESTAMP_OUTPUT_LENGTH - 1);
      ts_len= strlen(ts_start);

      /* Append a comma as the CSV field separator. */
      ts_start[ts_len++]= ',';
      ts_start[ts_len]= 0;

      if (ts_len > 0 && ts_len < TIMESTAMP_OUTPUT_LENGTH)
        memmove((char *) message - ts_len, ts_start, ts_len);

      char *final_message= (char *) message - ts_len;
      size_t final_len= len + ts_len;

      if (!(is_active= (logger_write(logfile, final_message, final_len) == (int) final_len)))
      {
        ++log_write_failures;
        result= 1;
      }
    }
  }
  else if (output_type == OUTPUT_SYSLOG)
  {
    syslog(syslog_facility_codes[syslog_facility] |
           syslog_priority_codes[syslog_priority],
           "%s %.*s", syslog_info, (int) len, message);
  }

  return result;
}

/**
  Write to the log, acquiring the lock.
*/

static int write_log_and_lock(const char *message, size_t len, unsigned long long ts_us)
{
  lock_operations.rd_lock();
  int result= write_log(message, len, ts_us);
  lock_operations.rd_unlock();
  return result;
}


/**
  Write to the log

  @param lock  whether the caller did not acquire lock_operations
*/
static int write_log_maybe_lock(const char *message, size_t len, bool lock,
                                unsigned long long ts_us)
{
  if (unlikely(!lock))
    return write_log(message, len, ts_us);
  else
    return write_log_and_lock(message, len, ts_us);
}


static size_t log_header(char *message, size_t message_len,
                      const char *serverhost, size_t serverhost_len,
                      const char *username, unsigned int username_len,
                      const char *host, unsigned int host_len,
                      const char *userip, unsigned int userip_len,
                      unsigned int connection_id, unsigned int port,
                      long long query_id, const char *operation)
{
  char port_str[16];
  if (host_len == 0 && userip_len != 0)
  {
    host_len= userip_len;
    host= userip;
  }
  if (port == 0) {
    port_str[0] = '\0';
  } else {
    my_snprintf(port_str, sizeof(port_str), ":%u", port);
  }

  /*
    That was added to find the possible cause of the MENT-1438.
    Supposed to be removed after that.
  */
  if (username_len > MAX_AUDIT_PAYLOAD_LENGTH)
  {
    username= "unknown_user";
    username_len= (unsigned int) strlen(username);
  }

  return my_snprintf(message, message_len,
      "%.*s,%.*s,%.*s%s,%d,%lld,%s",
      (int) serverhost_len, serverhost,
      username_len, username,
      host_len, host, port_str,
      connection_id, query_id, operation);
}

static size_t create_tls_obj(const struct mysql_event_connection *ev, char *obj_str, size_t len) {
  size_t obj_len;

  obj_len= 0;
  memset(obj_str, 0, len);
  if (ev->tls_version_length > 0) {
    obj_len= my_snprintf(obj_str, len,
      "%.*s", ev->tls_version_length, ev->tls_version);
  }
  return obj_len;
}

static int log_proxy(const struct connection_info *cn,
                     const struct mysql_event_connection *event)
{
  size_t csize;
  char raw_message[MAX_AUDIT_PAYLOAD_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message= raw_message + TIMESTAMP_OUTPUT_LENGTH;

  csize= log_header(message, MAX_AUDIT_PAYLOAD_LENGTH - 1,
                    servhost, servhost_len,
                    cn->user, cn->user_length,
                    cn->host, cn->host_length,
                    cn->ip, cn->ip_length,
                    event->thread_id, event->port,
                    0, "PROXY_CONNECT");
  csize+= my_snprintf(message+csize, MAX_AUDIT_PAYLOAD_LENGTH - 1 - csize,
    ",%.*s,`%.*s`@`%.*s`,%d", cn->db_length, cn->db,
                     cn->proxy_length, cn->proxy,
                     cn->proxy_host_length, cn->proxy_host,
                     event->status);
  message[csize]= '\n';
  return write_log_and_lock(message, csize + 1, my_hrtime().val);
}


static int log_connection(const struct connection_info *cn,
                          const struct mysql_event_connection *event,
                          const char *type)
{
  size_t csize;
  char raw_message[MAX_AUDIT_PAYLOAD_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message= raw_message + TIMESTAMP_OUTPUT_LENGTH;
  char tls_obj[32];
  size_t obj_len;

  csize= log_header(message, MAX_AUDIT_PAYLOAD_LENGTH - 1,
                    servhost, servhost_len,
                    cn->user, cn->user_length,
                    cn->host, cn->host_length,
                    cn->ip, cn->ip_length,
                    event->thread_id, event->port, 0, type);

  obj_len= create_tls_obj(event, tls_obj, sizeof(tls_obj));
  csize+= my_snprintf(message+csize, MAX_AUDIT_PAYLOAD_LENGTH - 1 - csize,
    ",%.*s,%.*s,%d", cn->db_length, cn->db, (int) obj_len, tls_obj,
    event->status);
  message[csize]= '\n';
  return write_log_and_lock(message, csize + 1, my_hrtime().val);
}


static int log_connection_event(const struct mysql_event_connection *event,
                                const char *type)
{
  size_t csize;
  char raw_message[MAX_AUDIT_PAYLOAD_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message= raw_message + TIMESTAMP_OUTPUT_LENGTH;
  char tls_obj[32];
  size_t obj_len;

  csize= log_header(message, MAX_AUDIT_PAYLOAD_LENGTH - 1,
                    servhost, servhost_len,
                    event->user, event->user_length,
                    event->host, event->host_length,
                    event->ip, event->ip_length,
                    event->thread_id, event->port, 0, type);
  obj_len= create_tls_obj(event, tls_obj, sizeof(tls_obj));
  csize+= my_snprintf(message+csize, MAX_AUDIT_PAYLOAD_LENGTH - 1 - csize,
    ",%.*s,%.*s,%d", (int) event->database.length,event->database.str,
    (int) obj_len, tls_obj, event->status);
  message[csize]= '\n';
  return write_log_and_lock(message, csize + 1, my_hrtime().val);
}


static size_t escape_string(const char *str, unsigned int len,
                          char *result, size_t result_len)
{
  const char *res_start= result;
  const char *res_end= result + result_len - 2;
  while (len)
  {
    char esc_c;

    if (result >= res_end)
      break;
    if ((esc_c= escaped_char(*str)))
    {
      if (result+1 >= res_end)
        break;
      *(result++)= '\\';
      *(result++)= esc_c;
    }
    else if (is_space(*str))
      *(result++)= ' ';
    else
      *(result++)= *str;
    str++;
    len--;
  }
  *result= 0;
  return result - res_start;
}

/*
  Replace "password" with "*****" in

  <word1> <maybe spaces> <word2> <maybe spaces> "password"

  if <word2> is 0

  <word1> <maybe spaces> "password"

  or

  <word0> <maybe spaces> <chr0> <maybe any characters> "password"

  if <chr0> is 0

  <word0> <maybe any characters> "password"

  NOTE: there can be " or ' around the password, the words are case
  insensitive.
*/

static size_t escape_string_hide_passwords(const char *str, unsigned int len,
    char *result, size_t result_len,
    const char *word1, size_t word1_len,
    const char *word2, size_t word2_len,
    const char *word0, size_t word0_len,
    char chr0)
{
  const char *res_start= result;
  const char *res_end= result + result_len - 2;
  size_t d_len;

  while (len)
  {
    int word1_found= (word1 && len > word1_len + 1 &&
                      strncasecmp(str, word1, word1_len) == 0);
    int word0_found= (word0 && len > word0_len + 1 &&
                      strncasecmp(str, word0, word0_len) == 0);
    if (word1_found || word0_found)
    {
      const char *next_s;
      size_t c;

      if (word0_found)
      {
        next_s= str + word0_len;
        if (chr0)
        {
          SKIP_SPACES(next_s);
          if (len < (size_t)(next_s - str) + 1 + 1 ||
              next_s[0] != chr0)
            goto no_password;
          next_s++;
        }
        while (*next_s && *next_s != '\'' && *next_s != '"')
          ++next_s;
      }
      else
      {
        next_s= str + word1_len;
        if (word2)
        {
          SKIP_SPACES(next_s);
          if (len < (next_s - str) + word2_len + 1 ||
              strncasecmp(next_s, word2, word2_len) != 0)
            goto no_password;
          next_s+= word2_len;
        }

        while (*next_s && *next_s != '\'' && *next_s != '"')
          ++next_s;
      }

      d_len= next_s - str;
      if (result + d_len + 5 > res_end)
        break;

      for (c=0; c<d_len; c++)
        result[c]= is_space(str[c]) ? ' ' : str[c];

      if (*next_s)
      {
        const char b_char= *next_s++;
        memset(result + d_len, '*', 5);
        result+= d_len + 5;

        while (*next_s)
        {
          if (*next_s == b_char)
          {
            ++next_s;
            break;
          }
          if (*next_s == '\\')
          {
            if (next_s[1])
              next_s++;
          }
          next_s++;
        }
      }
      else
        result+= d_len;

      len-= (uint)(next_s - str);
      str= next_s;
      continue;
    }
no_password:
    if (result >= res_end)
      break;
    else
    {
      const char b_char= escaped_char(*str);
      if (b_char)
      {
        if (result+1 >= res_end)
          break;
        *(result++)= '\\';
        *(result++)= b_char;
      }
      else if (is_space(*str))
        *(result++)= ' ';
      else
        *(result++)= *str;
      str++;
      len--;
    }
  }
  *result= 0;
  return result - res_start;
}



static int do_log_user(const char *name, int len,
                       const char *proxy, int proxy_len)
{
  int result;

  if (!name)
    return 0;

  lock_operations.rd_lock();

  if (incl_user_coll.n_users)
  {
    result= coll_search(&incl_user_coll, name, len) != 0 ||
            (proxy && coll_search(&incl_user_coll, proxy, proxy_len) != 0);
  }
  else if (excl_user_coll.n_users)
  {
    result= coll_search(&excl_user_coll, name, len) == 0 &&
            (proxy && coll_search(&excl_user_coll, proxy, proxy_len) == 0);
  }
  else
    result= 1;

  lock_operations.rd_unlock();
  return result;
}


static int log_statement_ex(struct connection_info *cn,
                            unsigned long long ev_time_us, unsigned long thd_id,
                            int sql_cmd, const char *query,
                            unsigned int query_len, int error_code,
                            const char *type, int take_lock)
{
  size_t csize;
  char raw_message_loc[MAX_AUDIT_QUERY_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message= raw_message_loc + TIMESTAMP_OUTPUT_LENGTH;
  size_t message_size= MAX_AUDIT_QUERY_LENGTH;
  char *uh_buffer;
  size_t uh_buffer_size;
  const char *db;
  unsigned int db_length;
  long long query_id;
  int result;
  char *big_buffer= NULL;
  int cmdtype= sql_command_to_cmdtype(sql_cmd);

  if ((db= cn->db))
    db_length= cn->db_length;
  else
  {
    db= "";
    db_length= 0;
  }

  if (!(query_id= cn->query_id))
    query_id= query_counter++;

  if (query == 0)
  {
    /* Can happen after the error in mysqld_prepare_stmt() */
    query= cn->query;
    query_len= cn->query_length;
    if (query == 0 || query_len == 0)
      return 0;
  }

  if (query && !(events & EVENT_QUERY_ALL) &&
      (events & EVENT_QUERY && !cn->log_always))
  {
    if (!(events & cmdtype &
          (EVENT_QUERY_DDL | EVENT_QUERY_DML | EVENT_QUERY_DML_NO_SELECT |
           EVENT_QUERY_DCL)))
      return 0;
  }

  csize= log_header(message, message_size-1,
                    servhost, servhost_len,
                    cn->user, cn->user_length,cn->host, cn->host_length,
                    cn->ip, cn->ip_length, thd_id, cn->port, query_id, type);

  csize+= my_snprintf(message+csize, message_size - 1 - csize,
      ",%.*s,\'", db_length, db);

  if (query_log_limit > 0 && query_len > query_log_limit)
    query_len= query_log_limit;

  if (query_len > (message_size - csize)/2)
  {
    size_t big_buffer_alloced= (query_len * 2 + csize + 4095 + TIMESTAMP_OUTPUT_LENGTH) & ~4095L;
    if (!(big_buffer= static_cast<char*>(malloc(big_buffer_alloced))))
      return 0;

    char *new_message= big_buffer + TIMESTAMP_OUTPUT_LENGTH;
    memcpy(new_message, message, csize);
    message= new_message;
    message_size= big_buffer_alloced - TIMESTAMP_OUTPUT_LENGTH;
  }

  uh_buffer= message + csize;
  uh_buffer_size= message_size - csize;
  if (query_log_limit > 0 && uh_buffer_size > query_log_limit+2)
    uh_buffer_size= query_log_limit+2;

  switch (sql_cmd)
  {
    case SQLCOM_GRANT:
    case SQLCOM_CREATE_USER:
    case SQLCOM_ALTER_USER:
      csize+= escape_string_hide_passwords(query, query_len,
                                           uh_buffer, uh_buffer_size,
                                           "IDENTIFIED", 10, "BY", 2,
                                           "PASSWORD", 8, '(');
      break;
    case SQLCOM_CHANGE_MASTER:
      csize+= escape_string_hide_passwords(query, query_len,
                                           uh_buffer, uh_buffer_size,
                                           "MASTER_PASSWORD", 15, "=", 1,
                                           0, 0, 0);
      break;
    case SQLCOM_CREATE_SERVER:
    case SQLCOM_ALTER_SERVER:
      csize+= escape_string_hide_passwords(query, query_len,
                                           uh_buffer, uh_buffer_size,
                                           "PASSWORD", 8, NULL, 0,
                                           0, 0, 0);
      break;
    case SQLCOM_SET_OPTION:
      csize+= escape_string_hide_passwords(query, query_len,
                                           uh_buffer, uh_buffer_size,
                                           "PASSWORD", 8, "=", 1,
                                           "PASSWORD", 8, '(');
      break;
    default:
      csize+= escape_string(query, query_len,
                               uh_buffer, uh_buffer_size); 
      break;
  }
  csize+= my_snprintf(message+csize, message_size - 1 - csize,
                      "\',%d", error_code);
  message[csize]= '\n';
  result= write_log_maybe_lock(message, csize + 1, take_lock, ev_time_us);

  if (cn->sync_statement && output_type == OUTPUT_FILE && logfile)
  {
    cn->sync_statement= FALSE;
    (void) logger_sync(logfile);
  }

  if (big_buffer)
    free(big_buffer);

  return result;
}


static int log_statement(struct connection_info *cn,
                         const struct mysql_event_general *event,
                         const char *type, int sql_command)
{
  DBUG_PRINT("info", ("log_statement: event_subclass=%d, general_command=%.*s, "
                      "cn->query_id=%lld, cn->query=%.*s, event->query_id=%lld, "
                      "event->general_query=%.*s, type=%s",
                        event->event_subclass, event->general_command_length,
                        event->general_command,
                        cn->query_id,
                        cn->query_length == 0x4f4f4f4f ? 0 : cn->query_length,
                        cn->query == (const char *)0x4f4f4f4f4f4f4f4fL ? NULL : cn->query,
                        event->query_id, event->general_query_length,
                        event->general_query, type));
  return log_statement_ex(cn, event->general_time_microseconds,
                          event->general_thread_id,
                          sql_command, event->general_query,
                          event->general_query_length,
                          event->general_error_code, type, 1);
}


static int log_table(const struct connection_info *cn,
                     const struct mysql_event_table *event, const char *type)
{
  size_t csize;
  char raw_message[MAX_AUDIT_PAYLOAD_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message = raw_message + TIMESTAMP_OUTPUT_LENGTH;

  csize= log_header(message, MAX_AUDIT_PAYLOAD_LENGTH - 1,
                    servhost, servhost_len,
                    event->user, SAFE_STRLEN_UI(event->user),
                    event->host, SAFE_STRLEN_UI(event->host),
                    event->ip, SAFE_STRLEN_UI(event->ip),
                    event->thread_id, event->port, cn->query_id, type);
  csize+= my_snprintf(message+csize, MAX_AUDIT_PAYLOAD_LENGTH - 1 - csize, ",%.*s,%.*s,",
                     (int) event->database.length, event->database.str,
                     (int) event->table.length, event->table.str);
  message[csize]= '\n';
  return write_log_and_lock(message, csize + 1, my_hrtime().val);
}


static int log_rename(const struct connection_info *cn,
                      const struct mysql_event_table *event)
{
  size_t csize;
  char raw_message[MAX_AUDIT_PAYLOAD_LENGTH + TIMESTAMP_OUTPUT_LENGTH];
  char *message= raw_message + TIMESTAMP_OUTPUT_LENGTH;

  csize= log_header(message, MAX_AUDIT_PAYLOAD_LENGTH - 1,
                    servhost, servhost_len,
                    event->user, SAFE_STRLEN_UI(event->user),
                    event->host, SAFE_STRLEN_UI(event->host),
                    event->ip, SAFE_STRLEN_UI(event->ip),
                    event->thread_id, event->port, 
                    cn->query_id, "RENAME");
  csize+= my_snprintf(message+csize, MAX_AUDIT_PAYLOAD_LENGTH - 1 - csize,
                      ",%.*s,%.*s|%.*s.%.*s,",
                      (int) event->database.length, event->database.str,
                      (int) event->table.length, event->table.str,
                      (int) event->new_database.length, event->new_database.str,
                      (int) event->new_table.length, event->new_table.str);
  message[csize]= '\n';
  return write_log_and_lock(message, csize + 1, my_hrtime().val);
}


static int event_query_command(const struct mysql_event_general *event)
{
  return (event->general_command_length == 5 &&
           strncmp(event->general_command, "Query", 5) == 0) ||
         (event->general_command_length == 7 &&
           (strncmp(event->general_command, "Execute", 7) == 0 ||
             (event->general_error_code != 0 &&
              strncmp(event->general_command, "Prepare", 7) == 0)));
}


static void update_general_user(struct connection_info *cn,
    const struct mysql_event_general *event)
{
  char uh_buffer[768];
  size_t user_len, host_len, ip_len;
  if (cn->user_length == 0 && cn->host_length == 0 && cn->ip_length == 0 &&
      get_user_host(event->general_user, event->general_user_length,
                    uh_buffer, sizeof(uh_buffer),
                    &user_len, &host_len, &ip_len) == 0)
  {
    get_str_n(cn->user, &cn->user_length, sizeof(cn->user), 
              uh_buffer, user_len);
    get_str_n(cn->host, &cn->host_length, sizeof(cn->host), 
              uh_buffer+user_len+1, host_len);
    get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip), 
              uh_buffer+user_len+1+host_len+1, ip_len);
  }

}


static struct connection_info ci_disconnect_buffer;

#define AA_FREE_CONNECTION 1
#define AA_CHANGE_USER 2

static void update_connection_info(MYSQL_THD thd, struct connection_info *cn,
    unsigned int event_class, const void *ev, int *after_action)
{
  *after_action= 0;

  switch (event_class) {
  case MYSQL_AUDIT_GENERAL_CLASS:
  {
    const struct mysql_event_general *event =
      (const struct mysql_event_general *) ev;
    cn->port= event->port;
    DBUG_PRINT("info", ("update_connection_info: before: event_subclass=%d, "
                        "general_command=%.*s, cn->query_id=%lld, cn->query=%.*s, "
                        "event->query_id=%lld, event->general_query=%.*s",
                        event->event_subclass, event->general_command_length,
                        event->general_command,
                        cn->query_id,
                        cn->query_length == 0x4f4f4f4f ? 0 : cn->query_length,
                        cn->query == (const char *)0x4f4f4f4f4f4f4f4fL ? NULL : cn->query,
                        event->query_id, event->general_query_length,
                        event->general_query));
    switch (event->event_subclass) {
      case MYSQL_AUDIT_GENERAL_LOG:
      {
        int init_db_command= event->general_command_length == 7 &&
          strncmp(event->general_command, "Init DB", 7) == 0;
        if (!ci_needs_setup(cn))
        {
          if (init_db_command)
          {
            /* Change DB */
            get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
                      event->general_query, event->general_query_length);
          }
          cn->query_id= mode ? query_counter++ : event->query_id;
          cn->query= event->general_query;
          cn->query_length= event->general_query_length;
          update_general_user(cn, event);
        }
        else if (init_db_command)
          setup_connection_initdb(cn, event);
        else if (event_query_command(event))
          setup_connection_query(cn, event);
        else
          setup_connection_simple(cn);
      }
      break;

      case MYSQL_AUDIT_GENERAL_STATUS:
        if (event_query_command(event))
        {
          if (ci_needs_setup(cn))
            setup_connection_query(cn, event);

          if (mode == 0)
          {
            if (cn->db_length == 0 && event->database.length > 0)
              get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
                        event->database.str, event->database.length);

            cn->query_id= event->query_id;
          }
          if (event->general_error_code == 0)
          {
            /* We need to check if it's the USE command to change the DB */
            int use_command= event->general_query_length > 4 &&
              strncasecmp(event->general_query, "use ", 4) == 0;
            if (use_command)
            {
              /* Change DB */
              if (mode)
                get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
                    event->general_query + 4, event->general_query_length - 4);
              else
                get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
                    event->database.str, event->database.length);
            }
          }
          update_general_user(cn, event);
        }
        break;

      case MYSQL_AUDIT_GENERAL_ERROR:
        /*
          We need this because the MariaDB returns NULL query field for the
          MYSQL_AUDIT_GENERAL_STATUS in the mysqld_stmt_prepare.
          As a result we get empty QUERY field for errors.
        */
        if (ci_needs_setup(cn))
          setup_connection_query(cn, event);
        cn->query_id= mode ? query_counter++ : event->query_id;
        get_str_n(cn->query_buffer, &cn->query_length, sizeof(cn->query_buffer),
            event->general_query, event->general_query_length);
        cn->query= cn->query_buffer;
        break;
      default:;
    }
    DBUG_PRINT("info", ("update_connection_info: after: event_subclass=%d, "
                        "general_command=%.*s, cn->query_id=%lld, cn->query=%.*s, "
                        "event->query_id=%lld, event->general_query=%.*s",
                        event->event_subclass, event->general_command_length,
                        event->general_command,
                        cn->query_id,
                        cn->query_length == 0x4f4f4f4f ? 0 : cn->query_length,
                        cn->query == (const char *)0x4f4f4f4f4f4f4f4fL ? NULL : cn->query,
                        event->query_id, event->general_query_length,
                        event->general_query));
    break;
  }
  case MYSQL_AUDIT_TABLE_CLASS:
  {
    const struct mysql_event_table *event =
      (const struct mysql_event_table *) ev;
    cn->port= event->port;
    if (ci_needs_setup(cn))
      setup_connection_table(cn, event);

    if (cn->user_length == 0 && cn->host_length == 0 && cn->ip_length == 0)
    {
      get_str_n(cn->user, &cn->user_length, sizeof(cn->user),
                event->user, SAFE_STRLEN(event->user));
      get_str_n(cn->host, &cn->host_length, sizeof(cn->host),
                event->host, SAFE_STRLEN(event->host));
      get_str_n(cn->ip, &cn->ip_length, sizeof(cn->ip),
                event->ip, SAFE_STRLEN(event->ip));
    }

    if (cn->db_length == 0 && event->database.length != 0)
      get_str_n(cn->db, &cn->db_length, sizeof(cn->db),
                event->database.str, event->database.length);

    if (mode == 0)
      cn->query_id= event->query_id;
    break;
  }
  case MYSQL_AUDIT_CONNECTION_CLASS:
  {
    const struct mysql_event_connection *event =
      (const struct mysql_event_connection *) ev;
    cn->port= event->port;
    switch (event->event_subclass)
    {
      case MYSQL_AUDIT_CONNECTION_CONNECT:
        setup_connection_connect(thd, cn, event);
        break;
      case MYSQL_AUDIT_CONNECTION_CHANGE_USER:
        *after_action= AA_CHANGE_USER;
        break;
      default:;
    }
    break;
  }
  default:
    break;
  }
}


struct connection_info cn_error_buffer;


#define FILTER(MASK) (events == 0 || (events & MASK))
void auditing(MYSQL_THD thd, unsigned int event_class, const void *ev)
{
  struct connection_info *cn= 0;
  int after_action= 0;

  /* That one is important as this function can be called with      */
  /* &lock_operations locked when the server logs an error reported */
  /* by this plugin.                                                */
  if (!thd || internal_stop_logging)
    return;

  cn= get_loc_info(thd);
  /* Without loc_info (see get_loc_info) the session is not audited. */
  if (cn)
    update_connection_info(thd, cn, event_class, ev, &after_action);

  if (!logging)
  {
    if (cn)
      cn->log_always= 0;
    goto exit_func;
  }

  if (event_class == MYSQL_AUDIT_GENERAL_CLASS && FILTER(EVENT_QUERY) &&
      cn && (cn->log_always || do_log_user(cn->user, cn->user_length,
                                           cn->proxy, cn->proxy_length)))
  {
    const struct mysql_event_general *event =
      (const struct mysql_event_general *) ev;

    /*
      Only one subclass is logged.
    */
    if (event->event_subclass == MYSQL_AUDIT_GENERAL_STATUS &&
        event_query_command(event))
    {
      log_statement(cn, event, "QUERY", thd_sql_command(thd));
      cn->query_length= 0; /* So the log_current_query() won't log this again. */
      cn->query_id= 0;
      cn->log_always= 0;
    }
  }
  else if (event_class == MYSQL_AUDIT_TABLE_CLASS && FILTER(EVENT_TABLE) && cn)
  {
    const struct mysql_event_table *event =
      (const struct mysql_event_table *) ev;
    if (do_log_user(event->user, (int) SAFE_STRLEN(event->user),
                    cn->proxy, cn->proxy_length))
    {
      switch (event->event_subclass)
      {
        case MYSQL_AUDIT_TABLE_LOCK:
          log_table(cn, event, event->read_only ? "READ" : "WRITE");
          break;
        case MYSQL_AUDIT_TABLE_CREATE:
          log_table(cn, event, "CREATE");
          break;
        case MYSQL_AUDIT_TABLE_DROP:
          log_table(cn, event, "DROP");
          break;
        case MYSQL_AUDIT_TABLE_RENAME:
          log_rename(cn, event);
          break;
        case MYSQL_AUDIT_TABLE_ALTER:
          log_table(cn, event, "ALTER");
          break;
        default:
          break;
      }
    }
  }
  else if (event_class == MYSQL_AUDIT_CONNECTION_CLASS &&
           FILTER(EVENT_CONNECT) && cn)
  {
    const struct mysql_event_connection *event =
      (const struct mysql_event_connection *) ev;
    switch (event->event_subclass)
    {
      case MYSQL_AUDIT_CONNECTION_CONNECT:
        log_connection(cn, event, event->status ? "FAILED_CONNECT": "CONNECT");
        if (event->status == 0 && event->proxy_user && event->proxy_user[0])
          log_proxy(cn, event);
        break;
      case MYSQL_AUDIT_CONNECTION_DISCONNECT:
        log_connection_event(event, "DISCONNECT");
        break;
      case MYSQL_AUDIT_CONNECTION_CHANGE_USER:
        log_connection(cn, event, "CHANGEUSER");
        if (event->proxy_user && event->proxy_user[0])
          log_proxy(cn, event);
        break;
      default:;
    }
  }
exit_func:
  /*
    This must work always, whether logging is ON or not.
  */

  if (after_action)
  {
    switch (after_action) {
    case AA_CHANGE_USER:
    {
      const struct mysql_event_connection *event =
        (const struct mysql_event_connection *) ev;
      change_connection(cn, event);
      break;
    }
    default:
      break;
    }
  }
}


struct mysql_event_general_v8
{
  unsigned int event_class;
  unsigned int event_subclass;
  int general_error_code;
  unsigned long general_thread_id;
  const char *general_user;
  unsigned int general_user_length;
  const char *general_command;
  unsigned int general_command_length;
  const char *general_query;
  unsigned int general_query_length;
  struct charset_info_st *general_charset;
  unsigned long long general_time;
  unsigned long long general_rows;
};


/*
   As it's just too difficult to #include "sql_class.h",
   let's just copy the necessary part of the system_variables
   structure here.
*/
typedef struct loc_system_variables
{
  ulong dynamic_variables_version;
  char* dynamic_variables_ptr;
  uint dynamic_variables_head;    /* largest valid variable offset */
  uint dynamic_variables_size;    /* how many bytes are in use */
  
  ulonglong max_heap_table_size;
  ulonglong tmp_table_size;
  ulonglong long_query_time;
  ulonglong optimizer_switch;
  ulonglong sql_mode; ///< which non-standard SQL behaviour should be enabled
  ulonglong option_bits; ///< OPTION_xxx constants, e.g. OPTION_PROFILING
  ulonglong join_buff_space_limit;
  ulonglong log_slow_filter; 
  ulonglong log_slow_verbosity; 
  ulonglong bulk_insert_buff_size;
  ulonglong join_buff_size;
  ulonglong sortbuff_size;
  ulonglong group_concat_max_len;
  ha_rows select_limit;
  ha_rows max_join_size;
  ha_rows expensive_subquery_limit;
  ulong auto_increment_increment, auto_increment_offset;
  ulong lock_wait_timeout;
  ulong join_cache_level;
  ulong max_allowed_packet;
  ulong max_error_count;
  ulong max_length_for_sort_data;
  ulong max_sort_length;
  ulong max_tmp_tables;
  ulong max_insert_delayed_threads;
  ulong min_examined_row_limit;
  ulong net_buffer_length;
  ulong net_interactive_timeout;
  ulong net_read_timeout;
  ulong net_retry_count;
  ulong net_wait_timeout;
  ulong net_write_timeout;
  ulong optimizer_prune_level;
  ulong optimizer_search_depth;
  ulong preload_buff_size;
  ulong profiling_history_size;
  ulong read_buff_size;
  ulong read_rnd_buff_size;
  ulong mrr_buff_size;
  ulong div_precincrement;
  /* Total size of all buffers used by the subselect_rowid_merge_engine. */
  ulong rowid_merge_buff_size;
  ulong max_sp_recursion_depth;
  ulong default_week_format;
  ulong max_seeks_for_key;
  ulong range_alloc_block_size;
  ulong query_alloc_block_size;
  ulong query_prealloc_size;
  ulong trans_alloc_block_size;
  ulong trans_prealloc_size;
  ulong log_warnings;
  /* Flags for slow log filtering */
  ulong log_slow_rate_limit; 
  ulong binlog_format; ///< binlog format for this thd (see enum_binlog_format)
  ulong progress_report_time;
  my_bool binlog_annotate_row_events;
  my_bool binlog_direct_non_trans_update;
  my_bool sql_log_bin;
  ulong completion_type;
  ulong query_cache_type;
} LOC_SV;


static int init_done= 0;

static int server_audit_init(void*)
{
  thd_priv_host_ptr= reinterpret_cast<decltype(thd_priv_host_ptr)>
    (dlsym(RTLD_DEFAULT, "thd_priv_host"));

  if (gethostname(servhost, sizeof(servhost)))
    strcpy(servhost, "unknown");

  servhost_len= (uint)strlen(servhost);

  logger_init_mutexes();
  lock_operations.init();

  coll_init(&incl_user_coll);
  coll_init(&excl_user_coll);

  if (incl_users)
  {
    if (excl_users)
    {
      incl_users= excl_users= NULL;
      error_header();
      fprintf(stderr, "INCL_DML_USERS and EXCL_DML_USERS specified"
                      " simultaneously - both set to empty\n");
    }
    update_incl_users(NULL, NULL, NULL, &incl_users);
  }
  else if (excl_users)
  {
    update_excl_users(NULL, NULL, NULL, &excl_users);
  }
  if (timestamp_format && timestamp_format[0])
  {
    update_timestamp_format(NULL, NULL, NULL, ×tamp_format);
  }


  error_header();
  fprintf(stderr, "MariaDB Audit Plugin version %s%s STARTED.\n",
          PLUGIN_STR_VERSION, PLUGIN_DEBUG_VERSION);

  /* The Query Cache shadows TABLE events if the result is taken from it */
  /* so we warn users if both Query Caсhe and TABLE events enabled.      */
  if (FILTER(EVENT_TABLE))
  {
    ulonglong *qc_size= (ulonglong *) dlsym(RTLD_DEFAULT, "query_cache_size");
    if (qc_size == NULL || *qc_size != 0)
    {
      struct loc_system_variables *g_sys_var=
        (struct loc_system_variables *) dlsym(RTLD_DEFAULT,
                                          "global_system_variables");
      if (g_sys_var && g_sys_var->query_cache_type != 0)
      {
        error_header();
        fprintf(stderr, "Query cache is enabled with the TABLE events."
                        " Some table reads can be veiled.");
      }
    }
  }

  ci_disconnect_buffer.header= 10;
  ci_disconnect_buffer.thread_id= 0;
  ci_disconnect_buffer.query_id= 0;
  ci_disconnect_buffer.db_length= 0;
  ci_disconnect_buffer.user_length= 0;
  ci_disconnect_buffer.host_length= 0;
  ci_disconnect_buffer.ip_length= 0;
  ci_disconnect_buffer.query= empty_str;
  ci_disconnect_buffer.query_length= 0;

  if (logging)
    start_logging();

  init_done= 1;
  return 0;
}


static int server_audit_deinit(void *)
{
  if (!init_done)
    return 0;

  init_done= 0;
  coll_free(&incl_user_coll);
  coll_free(&excl_user_coll);
  stop_logging();
  lock_operations.destroy();
  return 0;
}


static void rotate_log(MYSQL_THD, st_mysql_sys_var *, void *,
                       const void *save)
{
  if (output_type == OUTPUT_FILE && logfile &&
      *static_cast<const my_bool*>(save))
    (void) logger_rotate(logfile);
}


static void sync_log(MYSQL_THD thd, st_mysql_sys_var *, void *,
                     const void *save)
{
  if (output_type == OUTPUT_FILE && logfile &&
      static_cast<const my_bool*>(save))
  {
    struct connection_info *cn= get_loc_info(thd);
    (void) logger_sync(logfile);
    if (cn)
      cn->sync_statement= TRUE;
  }
}


static struct st_mysql_audit maria_descriptor =
{
  MYSQL_AUDIT_INTERFACE_VERSION,
  NULL,
  auditing,
  { MYSQL_AUDIT_GENERAL_CLASSMASK |
    MYSQL_AUDIT_TABLE_CLASSMASK |
    MYSQL_AUDIT_CONNECTION_CLASSMASK }
};
maria_declare_plugin(server_audit)
{
  MYSQL_AUDIT_PLUGIN,
  &maria_descriptor,
  "SERVER_AUDIT",
  "Alexey Botchkov (MariaDB Corporation)",
  "Audit the server activity",
  PLUGIN_LICENSE_GPL,
  server_audit_init,
  server_audit_deinit,
  PLUGIN_VERSION,
  audit_status,
  vars,
  PLUGIN_STR_VERSION,
  MariaDB_PLUGIN_MATURITY_STABLE
}
maria_declare_plugin_end;


static void mark_always_logged(MYSQL_THD thd)
{
  struct connection_info *cn;
  if (thd && (cn= get_loc_info(thd)))
    cn->log_always= 1;
}


static void log_current_query(MYSQL_THD thd)
{
  struct connection_info *cn;
  if (!thd)
    return;
  cn= get_loc_info(thd);
  if (cn && !ci_needs_setup(cn) && cn->query_length)
  {
    cn->log_always= 1;
    log_statement_ex(cn, my_hrtime().val, thd_get_thread_id(thd),
                     thd_sql_command(thd), cn->query, cn->query_length, 0,
                     "QUERY", 0);
    cn->log_always= 0;
  }
}


static void update_file_path(MYSQL_THD thd, st_mysql_sys_var *, void *,
                             const void *save)
{
  char *new_name= *static_cast<char*const*>(save);
  if (!new_name)
    new_name= empty_str;
  else if (strlen(new_name) + 4  > FN_REFLEN)
  {
    error_header();
    fprintf(stderr,
            "server_audit_file_path can't exceed %d characters.\n",
            FN_REFLEN - 4);
    fprintf(stderr, "Log filename remains unchanged '%s'.\n", file_path);
    CLIENT_ERROR(1, "server_audit_file_path can't exceed %d characters.",
                 MYF(ME_WARNING), FN_REFLEN - 4);
    return;
  }

  lock_operations.wr_lock();
  internal_stop_logging++;
  error_header();
  fprintf(stderr, "Log file name was changed to '%s'.\n", new_name);

  if (logging)
    log_current_query(thd);

  if (logging && output_type == OUTPUT_FILE)
  {
    char *sav_path= file_path;

    file_path= new_name;
    stop_logging();
    if (start_logging())
    {
      file_path= sav_path;
      error_header();
      fprintf(stderr, "Reverting log filename back to '%s'.\n", file_path);
      logging= (start_logging() == 0);
      if (!logging)
      {
        error_header();
        fprintf(stderr, "Logging was disabled..\n");
        CLIENT_ERROR(1, "Logging was disabled.", MYF(ME_WARNING));
      }
      goto exit_func;
    }
  }

  strncpy(path_buffer, new_name, sizeof(path_buffer)-1);
  path_buffer[sizeof(path_buffer)-1]= 0;
  file_path= path_buffer;
exit_func:
  internal_stop_logging--;
  lock_operations.wr_unlock();
}


static int check_timestamp_format(MYSQL_THD thd, st_mysql_sys_var *,
                                  void *save, struct st_mysql_value *value)
{
  const char *format;
  int len= 0;

  format= value->val_str(value, NULL, &len);
  if ((size_t) len >= TIMESTAMP_FORMAT_LENGTH)
  {
    error_header();
    fprintf(stderr, "server_audit_timestamp_format can't exceed %d characters.\n",
            TIMESTAMP_FORMAT_LENGTH - 1);
    return 1;
  }
  *((const char **) save)= format;
  return 0;
}


static void update_timestamp_format(MYSQL_THD thd, st_mysql_sys_var *, void *,
                                    const void *save)
{
  char *new_val= *static_cast<char*const*>(save);
  if (!new_val)
    new_val= empty_str;

  lock_operations.wr_lock();
  strncpy(timestamp_format_buffer, new_val, sizeof(timestamp_format_buffer)-1);
  timestamp_format_buffer[sizeof(timestamp_format_buffer)-1]= 0;
  timestamp_format= timestamp_format_buffer;
  lock_operations.wr_unlock();
}


static void update_file_rotations(MYSQL_THD, st_mysql_sys_var *,
                                  void *, const void *save)
{
  lock_operations.wr_lock();
  rotations= *static_cast<const unsigned*>(save);
  error_header();
  fprintf(stderr, "Log file rotations was changed to '%d'.\n", rotations);

  if (logging && output_type == OUTPUT_FILE)
    logger_set_rotations(logfile, rotations);

  lock_operations.wr_unlock();
}


static void update_file_rotate_size(MYSQL_THD, st_mysql_sys_var *, void*,
                                    const void *save)
{
  lock_operations.wr_lock();
  file_rotate_size= *static_cast<const unsigned long long *>(save);
  error_header();
  fprintf(stderr, "Log file rotate size was changed to '%lld'.\n",
          file_rotate_size);

  if (logging && output_type == OUTPUT_FILE)
    logger_set_filesize_limit(logfile, file_rotate_size);
  lock_operations.wr_unlock();
}


static void update_file_buffer_size(MYSQL_THD, st_mysql_sys_var *, void *,
                                    const void *save)
{
  lock_operations.wr_lock();
  file_buffer_size= *static_cast<const unsigned*>(save);

  error_header();
  fprintf(stderr, "Log file buffer size was changed to '%u'.\n",
          file_buffer_size);

  if (logging && output_type == OUTPUT_FILE &&
      logger_resize_buffer(logfile, file_buffer_size))
  {
    stop_logging();
    error_header();
    fprintf(stderr, "Buffer resize failed. Logging was disabled..\n");
    CLIENT_ERROR(1, "Buffer resize failed. Logging was disabled.",
                 MYF(ME_WARNING));
  }

  lock_operations.wr_unlock();
}


static int check_users(void *save, struct st_mysql_value *value,
                       size_t s, const char *name)
{
  const char *users;
  int len= 0;

  users= value->val_str(value, NULL, &len);
  if ((size_t) len > s)
  {
    error_header();
    fprintf(stderr,
            "server_audit_%s_users value can't be longer than %zu characters.\n",
            name, s);
    return 1;
  }
  *((const char**)save)= users;
  return 0;
}

static int check_incl_users(MYSQL_THD, st_mysql_sys_var *,
                            void *save, struct st_mysql_value *value)
{
  return check_users(save, value, sizeof(incl_user_buffer), "incl");
}

static int check_excl_users(MYSQL_THD, st_mysql_sys_var *,
                            void *save, struct st_mysql_value *value)
{
  return check_users(save, value, sizeof(excl_user_buffer), "excl");
}


static void update_incl_users(MYSQL_THD thd, st_mysql_sys_var *, void *,
                              const void *save)
{
  char *new_users= *static_cast<char*const*>(save);
  if (!new_users)
    new_users= empty_str;
  size_t new_len= strlen(new_users) + 1;
  mark_always_logged(thd);

  if (new_len > sizeof(incl_user_buffer))
    new_len= sizeof(incl_user_buffer);

  lock_operations.wr_lock();
  memcpy(incl_user_buffer, new_users, new_len - 1);
  incl_user_buffer[new_len - 1]= 0;

  incl_users= incl_user_buffer;
  user_coll_fill(&incl_user_coll, incl_users, &excl_user_coll, 1);
  error_header();
  fprintf(stderr, "server_audit_incl_users set to '%s'.\n", incl_users);
  lock_operations.wr_unlock();
}


static void update_excl_users(MYSQL_THD thd, st_mysql_sys_var *, void *,
                              const void *save)
{
  char *new_users= *static_cast<char*const*>(save);
  if (!new_users)
    new_users= empty_str;
  size_t new_len= strlen(new_users) + 1;
  mark_always_logged(thd);

  if (new_len > sizeof(excl_user_buffer))
    new_len= sizeof(excl_user_buffer);

  lock_operations.wr_lock();
  memcpy(excl_user_buffer, new_users, new_len - 1);
  excl_user_buffer[new_len - 1]= 0;

  excl_users= excl_user_buffer;
  user_coll_fill(&excl_user_coll, excl_users, &incl_user_coll, 0);
  error_header();
  fprintf(stderr, "server_audit_excl_users set to '%s'.\n", excl_users);
  lock_operations.wr_unlock();
}


static void update_output_type(MYSQL_THD thd, st_mysql_sys_var *, void *,
                               const void *save)
{
  ulong new_output_type= *static_cast<const ulong*>(save);
  if (output_type == new_output_type)
    return;

  lock_operations.wr_lock();
  if (logging)
  {
    log_current_query(thd);
    stop_logging();
  }

  output_type= new_output_type;
  error_header();
  fprintf(stderr, "Output was redirected to '%s'\n",
          output_type_names[output_type]);

  if (logging)
    start_logging();
  lock_operations.wr_unlock();
}


static void update_syslog_facility(MYSQL_THD thd, st_mysql_sys_var *, void *,
                                   const void *save)
{
  ulong new_facility= *static_cast<const ulong*>(save);
  lock_operations.wr_lock();
  ulong old_facility= syslog_facility;
  if (old_facility != new_facility)
  {
    syslog_facility= new_facility;
    mark_always_logged(thd);
    error_header();
    fprintf(stderr, "SysLog facility was changed from '%s' to '%s'.\n",
            syslog_facility_names[old_facility],
            syslog_facility_names[new_facility]);
  }
  lock_operations.wr_unlock();
}


static void update_syslog_priority(MYSQL_THD thd, st_mysql_sys_var *, void *,
                                   const void *save)
{
  ulong new_priority= *static_cast<const ulong*>(save);
  lock_operations.wr_lock();
  ulong old_priority= syslog_priority;
  if (old_priority != new_priority)
  {
    syslog_priority= new_priority;
    mark_always_logged(thd);
    error_header();
    fprintf(stderr, "SysLog priority was changed from '%s' to '%s'.\n",
            syslog_priority_names[old_priority],
            syslog_priority_names[new_priority]);
  }
  lock_operations.wr_unlock();
}


static void update_logging(MYSQL_THD thd, st_mysql_sys_var *, void *,
                           const void *save)
{
  char new_logging= *static_cast<const char*>(save);
  lock_operations.wr_lock();
  if (new_logging != logging)
  {
    logging= new_logging;
    if (new_logging)
    {
      start_logging();
      if (!logging)
        CLIENT_ERROR(1, "Logging was disabled.", MYF(ME_WARNING));
      else
        mark_always_logged(thd);
    }
    else
    {
      log_current_query(thd);
      stop_logging();
    }
  }
  lock_operations.wr_unlock();
}


static void update_mode(MYSQL_THD thd, st_mysql_sys_var *, void *,
                        const void *save)
{
  unsigned new_mode= *static_cast<const unsigned*>(save);
  lock_operations.wr_lock();
  unsigned old_mode= mode;
  if (new_mode != old_mode)
  {
    mode= new_mode;
    mark_always_logged(thd);
    error_header();
    fprintf(stderr, "Logging mode was changed from %u to %u.\n",
            old_mode, new_mode);
  }
  lock_operations.wr_unlock();
}


static void update_syslog_ident(MYSQL_THD thd, st_mysql_sys_var *, void *,
                                const void *save)
{
  char *new_ident= *static_cast<char*const*>(save);
  lock_operations.wr_lock();
  if (!new_ident)
    *syslog_ident= '\0';
  else
  {
    strncpy(syslog_ident, new_ident, sizeof(syslog_ident)-1);
    syslog_ident[sizeof(syslog_ident)-1]= 0;
  }
  error_header();
  fprintf(stderr, "SYSLOG ident was changed to '%s'\n", syslog_ident);
  mark_always_logged(thd);
  if (logging && output_type == OUTPUT_SYSLOG)
  {
    stop_logging();
    start_logging();
  }
  lock_operations.wr_unlock();
}


IF_WIN(static,__attribute__ ((constructor)))
void audit_plugin_so_init()
{
  memset(locinfo_ini_value, 'O', sizeof(locinfo_ini_value)-1);
  locinfo_ini_value[sizeof(locinfo_ini_value)-1]= 0;
}

#ifdef _WIN32
BOOL WINAPI DllMain(HINSTANCE, DWORD fdwReason, LPVOID)
{
  if (fdwReason == DLL_PROCESS_ATTACH)
    audit_plugin_so_init();
  return 1;
}
#elif !defined SUX_LOCK_GENERIC
# ifdef __linux__
#  include <linux/futex.h>
#  include <sys/syscall.h>
#  define SRW_FUTEX(a,op,n) \
   syscall(SYS_futex, a, FUTEX_ ## op ## _PRIVATE, n, nullptr, nullptr, 0)
# elif defined __OpenBSD__
#  include <sys/time.h>
#  include <sys/futex.h>
#  define SRW_FUTEX(a,op,n) \
   futex((volatile uint32_t*) a, FUTEX_ ## op, n, nullptr, nullptr)
# elif defined __FreeBSD__
#  include <sys/types.h>
#  include <sys/umtx.h>
#  define FUTEX_WAKE UMTX_OP_WAKE_PRIVATE
#  define FUTEX_WAIT UMTX_OP_WAIT_UINT_PRIVATE
#  define SRW_FUTEX(a,op,n) _umtx_op(a, FUTEX_ ## op, n, nullptr, nullptr)
# elif defined __DragonFly__
#  include <unistd.h>
#  define FUTEX_WAKE(a,n) umtx_wakeup(a,n)
#  define FUTEX_WAIT(a,n) umtx_sleep(a,n,0)
#  define SRW_FUTEX(a,op,n) FUTEX_ ## op((volatile int*) a, int(n))
# else
#  error "no futex support nor #define SUX_LOCK_GENERIC"
# endif

# ifdef __GNUC__
#  pragma GCC visibility push(hidden) /* Avoid a symbol clash with InnoDB */
# endif

template<> inline void srw_mutex_impl<false>::wait(uint32_t lk) noexcept
{ SRW_FUTEX(&lock, WAIT, lk); }
template<> inline void srw_mutex_impl<false>::wake() noexcept
{ SRW_FUTEX(&lock, WAKE, 1); }
template<> inline void srw_mutex_impl<false>::wake_all() noexcept
{ SRW_FUTEX(&lock, WAKE, INT_MAX); }
template<> inline void ssux_lock_impl<false>::wait(uint32_t lk) noexcept
{ SRW_FUTEX(&readers, WAIT, lk); }
template<> inline void ssux_lock_impl<false>::wake() noexcept
{ SRW_FUTEX(&readers, WAKE, 1); }

template<>
void srw_mutex_impl<false>::wait_and_lock() noexcept
{
  uint32_t lk= WAITER + lock.fetch_add(WAITER, std::memory_order_relaxed);
  for (;;)
  {
    DBUG_ASSERT(~HOLDER & lk);
    if (lk & HOLDER)
    {
      wait(lk);
#if defined __i386__||defined __x86_64__
reload:
#endif
      lk= lock.load(std::memory_order_relaxed);
    }
    else
    {
#if defined __i386__||defined __x86_64__
      if (lock.fetch_or(HOLDER, std::memory_order_relaxed) & HOLDER)
        goto reload;
#else
      if ((lk= lock.fetch_or(HOLDER, std::memory_order_relaxed)) & HOLDER)
        continue;
      DBUG_ASSERT(lk);
#endif
      std::atomic_thread_fence(std::memory_order_acquire);
      return;
    }
  }
}

template<>
void ssux_lock_impl<false>::wr_wait(uint32_t lk) noexcept
{
  DBUG_ASSERT(writer.is_locked());
  DBUG_ASSERT(lk);
  DBUG_ASSERT(lk < WRITER);

  lk|= WRITER;

  do
  {
    DBUG_ASSERT(lk > WRITER);
    wait(lk);
    lk= readers.load(std::memory_order_acquire);
  }
  while (lk != WRITER);
}

template<bool spinloop>
void ssux_lock_impl<spinloop>::rd_lock_nospin() noexcept
{
  /* Subscribe to writer.wake() or write.wake_all() calls of
  concurrently executing rd_wait() or writer.wr_unlock(). */
  uint32_t wl= writer.WAITER +
    writer.lock.fetch_add(writer.WAITER, std::memory_order_acquire);

  for (;;)
  {
    if (writer.HOLDER & wl)
      writer.wait(wl);
    uint32_t lk= rd_lock_try_low();
    if (!lk)
      break;
    if (UNIV_UNLIKELY(lk == WRITER)) /* A wr_lock() just succeeded. */
      /* Immediately wake up (also) wr_lock(). We may also unnecessarily
      wake up other concurrent threads that are executing rd_wait().
      If we invoked writer.wake() here to wake up just one thread,
      we could wake up a rd_wait(), which then would invoke writer.wake(),
      waking up possibly another rd_wait(), and we could end up doing
      lots of non-productive context switching until the wr_lock()
      is finally woken up. */
      writer.wake_all();
    wl= writer.lock.load(std::memory_order_acquire);
    ut_ad(wl);
  }

  /* Unsubscribe writer.wake() and writer.wake_all(). */
  wl= writer.lock.fetch_sub(writer.WAITER, std::memory_order_release);
  ut_ad(wl);

  /* Wake any other threads that may be blocked in writer.wait().
  All other waiters than this rd_wait() would end up acquiring writer.lock
  and waking up other threads on unlock(). */
  if (wl > writer.WAITER)
    writer.wake_all();
}
#endif