* Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd.
* Portions Copyright (c) 2004-2012, PostgreSQL Global Development Group
*
*
* pgaudit.cpp
* auditor process
*
* The audit collector (auditor) catches all audit output from
* the postmaster, backends, and other subprocesses by redirecting to a
* pipe, and writes it to a set of auditfiles.
* It's possible to have size and age limits for the auditfile configured
* in postgresql.conf. If these limits are reached or passed, the
* current auditfile is closed and a new one is created (rotated).
* The auditfiles are stored in a subdirectory (configurable in
* postgresql.conf), using a user-selectable naming scheme.
*
* IDENTIFICATION
* src/gausskernel/process/postmaster/pgaudit.cpp
*
* -------------------------------------------------------------------------
*/
#include "postgres.h"
#include "knl/knl_variable.h"
#include <fcntl.h>
#include <limits.h>
#include <sys/time.h>
#include "lib/stringinfo.h"
#include "libpq/libpq-be.h"
#include "libpq/pqsignal.h"
#include "libpq/sha2.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/pg_list.h"
#include "pgtime.h"
#include "postmaster/fork_process.h"
#include "postmaster/postmaster.h"
#include "pgaudit.h"
#include "gs_policy/curl_utils.h"
#include "pgxc/pgxc.h"
#include "storage/ipc.h"
#include "storage/smgr/fd.h"
#include "storage/latch.h"
#include "storage/pg_shmem.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/ps_status.h"
#include "utils/timestamp.h"
#include "utils/builtins.h"
#include "utils/acl.h"
#include "utils/elog.h"
#include "auditfuncs.h"
#include "gssignal/gs_signal.h"
#include "gs_policy/policy_common.h"
#include <string>
#include <fstream>
#include <sstream>
#include <iostream>
#ifdef HAVE_STDIO_H
#include <stdio.h>
#endif
#ifdef HAVE_STDLIB_H
#include <stdlib.h>
#endif
#ifdef HVE_STRING_H
#include <string.h>
#endif
#ifdef HAVE_OPENSSL_SHA_H
#include <openssl/sha.h>
#endif
#ifdef ENABLE_UT
#define static
#endif
* Primitive protocol structure for writing to sysauditor pipe(s). The idea
* here is to divide long messages into chunks that are not more than
* PIPE_BUF bytes long, which according to POSIX spec must be written into
* the pipe atomically. The pipe reader then uses the protocol headers to
* reassemble the parts of a message into a single string. The reader can
* also cope with non-protocol data coming down the pipe, though we cannot
* guarantee long strings won't get split apart.
*
* We use non-nul bytes in is_last to make the protocol a tiny bit
* more robust against finding a false double nul byte prologue. But
* we still might find it in the len and/or pid bytes unless we're careful.
*/
#ifdef PIPE_BUF
#if PIPE_BUF > 65536
#define PIPE_CHUNK_SIZE 65536
#else
#define PIPE_CHUNK_SIZE ((int)PIPE_BUF)
#endif
#else
#define PIPE_CHUNK_SIZE 512
#endif
#define MAX_QUEUE_SIZE 100000
#define MAX_CONNECTION_INFO_SIZE (MAXNUMLEN * 4 + NI_MAXHOST)
const static uint32 uint64_max_len = 20;
const uint32 AUDIT_INDEX_TABLE_VERSION_NUM = 92601;
typedef struct {
char nuls[2];
uint16 len;
ThreadId pid;
char is_last;
char data[1];
} PipeProtoHeader;
typedef union {
PipeProtoHeader proto;
char filler[PIPE_CHUNK_SIZE];
} PipeProtoChunk;
#define PIPE_HEADER_SIZE offsetof(PipeProtoHeader, data)
#define PIPE_MAX_PAYLOAD ((int)(PIPE_CHUNK_SIZE - PIPE_HEADER_SIZE))
#define PIPE_READ_INDEX(index) (index * 2)
#define PIPE_WRITE_INDEX(index) (index * 2 + 1)
#define AUDIT_THREADNUM_ENLARGE \
((uint32)g_instance.audit_cxt.thread_num > g_instance.audit_cxt.audit_indextbl->thread_num)
* We really want line-buffered mode for auditfile output, but Windows does
* not have it, and interprets _IOLBF as _IOFBF (bozos). So use _IONBF
* instead on Windows.
*/
#ifdef WIN32
#define LBF_MODE _IONBF
#define DIR_SEP "\\"
#else
#define LBF_MODE _IOLBF
#define DIR_SEP "/"
#endif
* We read() into a temp buffer twice as big as a chunk, so that any fragment
* left after processing can be moved down to the front and we'll still have
* room to read a full chunk.
*/
#define READ_BUF_SIZE (2 * PIPE_CHUNK_SIZE)
THR_LOCAL int PolicyAudit_RotationAge = 5;
* Brief : bitnum in integer Audit_Session
* Description :
*/
typedef enum {
SESSION_LOGIN_SUCCESS = 0,
SESSION_LOGIN_FAILED,
SESSION_LOGOUT
} SessionType;
* Private state
*/
static char* pgaudit_filename = "%s/%d_adt";
static char* policy_audit_filename = "%s/%lld_event.bin";
static int pgaudit_filemode = S_IRUSR | S_IWUSR;
static THR_LOCAL pg_time_t policy_next_rotation_time;
CurlUtils m_curlUtils;
* Buffers for saving partial messages from different backends.
*
* Keep NBUFFER_LISTS lists of these, with the entry for a given source pid
* being in the list numbered (pid % NBUFFER_LISTS), so as to cut down on
* the number of entries we have to examine for any one incoming message.
* There must never be more than one entry for the same source pid.
*
* An inactive buffer is not removed from its list, just held for re-use.
* An inactive buffer has pid == 0 and undefined contents of data.
*/
typedef struct {
ThreadId pid;
StringInfoData data;
} save_buffer;
#define NBUFFER_LISTS 256
* Flags set by interrupt handlers for later service in the main loop.
*/
#define SPACE_INTERVAL_SIZE (10 * 1024 * 1024)
const static uint64 SPACE_MAXIMUM_SIZE = (1024 * 1024 * 1024 * 1024L);
* Brief : audit index item in index table
* Description :
*/
typedef struct AuditIndexItem {
* file create time. used when scan all the audit data.
* if system time changed when auditor write into this file,
* then ctime would less than zero.
*/
pg_time_t ctime;
uint32 filenum;
uint32 filesize;
} AuditIndexItem;
* Brief : audit index table
* Description :
*/
typedef struct AuditIndexTableNew {
uint32 maxnum;
uint32 begidx;
uint32 thread_num;
volatile uint32 latest_idx;
uint32 curidx[MAX_AUDIT_NUM];
uint32 count;
pg_time_t last_audit_time;
AuditIndexItem data[1];
} AuditIndexTableNew;
* Brief : old audit index table
* Description :
*/
typedef struct AuditIndexTable {
uint32 maxnum;
uint32 begidx;
uint32 curidx;
uint32 count;
pg_time_t last_audit_time;
AuditIndexItem data[1];
} AuditIndexTable;
static const char audit_indextbl_file[] = "index_table_new";
static const char audit_indextbl_old_file[] = "index_table";
static const int indextbl_header_size = offsetof(AuditIndexTableNew, data);
static const int old_indextbl_header_size = offsetof(AuditIndexTable, data);
static const char* AuditTypeDescs[] = {"unknown",
"login_success",
"login_failed",
"user_logout",
"system_start",
"system_stop",
"system_recover",
"system_switch",
"lock_user",
"unlock_user",
"grant_role",
"revoke_role",
"user_violation",
"ddl_database",
"ddl_directory",
"ddl_tablespace",
"ddl_schema",
"ddl_user",
"ddl_table",
"ddl_index",
"ddl_view",
"ddl_trigger",
"ddl_function",
"ddl_resourcepool",
"ddl_workload",
"ddl_serverforhadoop",
"ddl_datasource",
"ddl_nodegroup",
"ddl_rowlevelsecurity",
"ddl_synonym",
"ddl_type",
"ddl_textsearch",
"dml_action",
"dml_action_select",
"internal_event",
"function_exec",
"system_function_exec",
"copy_to",
"copy_from",
"set_parameter",
"audit_policy",
"masking_policy",
"security_policy",
"ddl_sequence",
"ddl_key",
"ddl_package",
"ddl_model",
"ddl_globalconfig",
"ddl_publication_subscription",
"ddl_foreign_data_wrapper",
"ddl_sql_patch",
"ddl_event"
};
static const int AuditTypeNum = sizeof(AuditTypeDescs) / sizeof(char*);
#define AuditTypeDesc(type) (((type) > 0 && (type) < AuditTypeNum) ? AuditTypeDescs[(type)] : AuditTypeDescs[0])
static const char* AuditResultDescs[] = {"unknown", "ok", "failed"};
static const int AuditResultNum = sizeof(AuditResultDescs) / sizeof(char*);
#define AuditResultDesc(type) (((type) > 0 && (type) < AuditResultNum) ? AuditResultDescs[(type)] : AuditResultDescs[0])
* Brief : The audit message header
* Description : exactly 160 bits.
*/
typedef struct AuditMsgHdr {
char signature[2];
uint16 version;
uint16 fields;
uint16 flags;
pg_time_t time;
uint32 size;
} AuditMsgHdr;
#define AUDIT_TUPLE_NORMAL 1
#define AUDIT_TUPLE_DEAD 2
typedef struct AuditEncodedData {
AuditMsgHdr header;
char data[1];
} AuditEncodedData;
* AuditData holds the data accumulated during any one audit_report() cycle.
* Any non-NULL pointers must point to palloc'd data.
* (The const pointers are an exception; we assume they point at non-freeable
* constant strings.)
*/
typedef struct AuditData {
AuditMsgHdr header;
AuditType type;
AuditResult result;
char varstr[1];
} AuditData;
#define AUDIT_HEADER_SIZE offsetof(AuditData, varstr)
#define FILED_NULLABLE(field) (field ? field : _("null"))
#define PGAUDIT_RESTART_INTERVAL 60
#define PGAUDIT_QUERY_COLS 13
#define PGAUDIT_QUERY_COLS_NEW 15
#define MAXNUMLEN 16
#define WRITE_TO_AUDITPIPE (g_instance.audit_cxt.audit_init_done && t_thrd.role != AUDITOR)
#define WRITE_TO_STDAUDITFILE(ctype) (t_thrd.role == AUDITOR && ctype == STD_AUDIT_TYPE)
#define WRITE_TO_UNIAUDITFILE(ctype) (t_thrd.role == AUDITOR && ctype == UNIFIED_AUDIT_TYPE)
#define MAX_DATA_LEN 1024
#define SHA256_LENTH 32
#define SHA256_HEX_LENTH 512
#define SHA_LOG_MAX_TIMELEN 128
struct AuditEventInfo {
AuditEventInfo() : userid{0},
username(NULL),
dbname(NULL),
appname(NULL),
remotehost(NULL),
client_info{0},
threadid{0},
localport{0},
remoteport{0} {}
char userid[MAXNUMLEN];
const char* username;
const char* dbname;
const char* appname;
const char* remotehost;
char client_info[MAX_CONNECTION_INFO_SIZE];
char threadid[MAXNUMLEN * 4];
char localport[MAXNUMLEN];
char remoteport[MAXNUMLEN];
};
typedef enum {
SYSAUDITFILE_TYPE = 1,
POLICYAUDITFILE_TYPE,
UNKNOWNFILE_TYPE
} AuditFileType;
static void process_pipe_input(char* auditbuffer, int* bytes_in_auditbuffer);
static void flush_pipe_input(char* auditbuffer, int* bytes_in_auditbuffer);
static void pgaudit_write_file(char* buffer, int count);
static void auditfile_init(bool allow_errors = false);
static FILE *auditfile_open(pg_time_t timestamp, const char *mode, bool allow_errors,
const char *filename = pgaudit_filename, bool ignore_num = false);
static void auditfile_close(AuditFileType flag);
static void auditfile_rotate(bool time_based_rotation, bool size_based_rotation);
static void set_next_rotation_time(void);
static void pgaudit_cleanup(void);
static void pgaudit_exit(SIGNAL_ARGS);
static void sigHupHandler(SIGNAL_ARGS);
static void sigUsr1Handler(SIGNAL_ARGS);
static void sig_thread_quit_handler();
static void sig_thread_config_handler(int ¤tAuditRotationAge, int ¤tAuditRemainThreshold);
static void pgauditor_kill(int code, Datum arg);
static void audit_process_cxt_init();
static void audit_process_cxt_exit();
static void write_pipe_chunks(char* data, int len, AuditClassType type = STD_AUDIT_TYPE);
static void appendStringField(StringInfo str, const char* s);
static void pgaudit_close_file(FILE* fp, const char* file);
static void pgaudit_read_indexfile(const char* audit_directory);
static void pgaudit_update_indexfile(const char* mode, bool allow_errors);
static bool pgaudit_find_indexfile(void);
static void pgaudit_indexfile_upgrade(void);
static void pgaudit_indexfile_sync(const char* mode, bool allow_errors);
static void pgaudit_rewrite_indexfile(void);
static void pgaudit_indextbl_init_new(void);
static void pgaudit_reset_indexfile();
static const char* pgaudit_string_field(AuditData* adata, int num);
static void deserialization_to_tuple(Datum (&values)[PGAUDIT_QUERY_COLS_NEW],
AuditData *adata,
const AuditMsgHdr &header,
bool nulls[PGAUDIT_QUERY_COLS_NEW],
bool newVersion);
static void pgaudit_query_file(Tuplestorestate *state, TupleDesc tdesc, uint32 fnum, TimestampTz begtime,
TimestampTz endtime, const char *audit_directory, bool newVersion);
static TimestampTz pgaudit_headertime(uint32 fnum, const char *audit_directory);
static void pgaudit_query_valid_check(const ReturnSetInfo *rsinfo, FunctionCallInfoData *fcinfo, TupleDesc &tupdesc, bool newVersion);
static uint32 pgaudit_get_auditfile_num();
static void pgaudit_update_auditfile_time(pg_time_t timestamp, bool exist);
static void pgaudit_switch_next_auditfile();
static void pgaudit_send_data_to_elastic();
static void pgaudit_query_file_for_elastic();
static void elasic_search_connection_test();
static void policy_auditfile_rotate();
static void set_next_policy_rotation_time(void);
static void pgaudit_write_policy_audit_file(const char* buffer, int count);
inline bool pgaudit_need_check_time_rotation()
{
return (u_sess->attr.attr_security.Audit_RotationAge > 0 && !t_thrd.audit.rotation_disabled);
}
inline bool pgaudit_need_check_size_rotation()
{
return (!t_thrd.audit.rotation_requested && u_sess->attr.attr_security.Audit_RotationSize > 0 &&
!t_thrd.audit.rotation_disabled);
}
static void CheckAuditFile(void);
static bool pgaudit_invalid_header(const AuditMsgHdr* header, bool newVersion);
static void pgaudit_mark_corrupt_info(uint32 fnum);
static void audit_append_xid_info(const char *detail_info, char *detail_info_xid, uint32 len);
static bool audit_status_check_ok();
static bool pgaudit_need_sha_code();
static void generateAuditShaCode(AuditShaRecord *shaRecord, char* hexbuf);
static void init_audit_signal_handlers()
{
(void)gspqsignal(SIGHUP, sigHupHandler);
(void)gspqsignal(SIGINT, SIG_IGN);
(void)gspqsignal(SIGTERM, SIG_IGN);
(void)gspqsignal(SIGQUIT, pgaudit_exit);
(void)gspqsignal(SIGALRM, SIG_IGN);
(void)gspqsignal(SIGPIPE, SIG_IGN);
(void)gspqsignal(SIGUSR1, sigUsr1Handler);
(void)gspqsignal(SIGUSR2, SIG_IGN);
(void)gspqsignal(SIGCHLD, SIG_DFL);
(void)gspqsignal(SIGTTIN, SIG_DFL);
(void)gspqsignal(SIGTTOU, SIG_DFL);
(void)gspqsignal(SIGCONT, SIG_DFL);
(void)gspqsignal(SIGWINCH, SIG_DFL);
(void)gspqsignal(SIGURG, print_stack);
gs_signal_setmask(&t_thrd.libpq_cxt.UnBlockSig, NULL);
(void)gs_signal_unblock_sigusr2();
}
static void pgaudit_handle_exception()
{
t_thrd.log_cxt.error_context_stack = NULL;
t_thrd.log_cxt.call_stack = NULL;
HOLD_INTERRUPTS();
EmitErrorReport();
ereport(LOG, (errmsg("auditor thread with thread index : %d shutdown abnormaly", t_thrd.audit.cur_thread_idx)));
(void)MemoryContextSwitchTo(t_thrd.mem_cxt.pgAuditLocalContext);
FlushErrorState();
MemoryContextResetAndDeleteChildren(t_thrd.mem_cxt.pgAuditLocalContext);
LWLockReleaseAll();
if (t_thrd.utils_cxt.CurrentResourceOwner) {
ResourceOwnerRelease(t_thrd.utils_cxt.CurrentResourceOwner, RESOURCE_RELEASE_BEFORE_LOCKS, false, true);
}
RESUME_INTERRUPTS();
* Sleep at least 1 second after any error. A write error is likely
* to be repeated, and we don't want to be filling the error logs as
* fast as we can.
*/
pg_usleep(1000000L);
return;
}
static void sig_thread_quit_handler()
{
* audit master thread should try its best to do the flush job when exit:
* 1. wait other audit thread exit, so that index file will not change
* 2. flush index audit file
*/
if (t_thrd.audit.cur_thread_idx == 0) {
while (t_thrd.audit.need_exit) {
if (pg_atomic_read_u32(&g_instance.audit_cxt.current_audit_index) == 1) {
break;
}
}
pgaudit_cleanup();
pgaudit_update_indexfile(PG_BINARY_W, true);
}
ereport(LOG, (errmsg("auditor thread exit, id : %d", t_thrd.audit.cur_thread_idx)));
* From here on, elog(ERROR) should end with exit(1), not send
* control back to the sigsetjmp block above.
*/
u_sess->attr.attr_common.ExitOnAnyError = true;
}
static void sig_thread_config_handler(int ¤tAuditRotationAge, int ¤tAuditRemainThreshold)
{
ProcessConfigFile(PGC_SIGHUP);
* If rotation time parameter changed, reset next rotation time,
* but don't immediately force a rotation.
*/
if (currentAuditRotationAge != u_sess->attr.attr_security.Audit_RotationAge) {
currentAuditRotationAge = u_sess->attr.attr_security.Audit_RotationAge;
set_next_rotation_time();
}
if (t_thrd.audit.cur_thread_idx == 0 &&
currentAuditRemainThreshold != u_sess->attr.attr_security.Audit_RemainThreshold) {
currentAuditRemainThreshold = u_sess->attr.attr_security.Audit_RemainThreshold;
pgaudit_update_indexfile(PG_BINARY_W, true);
pgaudit_reset_indexfile();
}
* If we had a rotation-disabling failure, re-enable rotation
* attempts after SIGHUP, and force one immediately.
*/
if (t_thrd.audit.rotation_disabled) {
t_thrd.audit.rotation_disabled = false;
t_thrd.audit.rotation_requested = true;
}
}
static void pgaudit_exit_clean(int code, Datum arg)
{
if (t_thrd.audit.cur_thread_idx == 0) {
m_curlUtils.~CurlUtils();
}
}
static bool pgaudit_need_sha_code()
{
if (t_thrd.proc == NULL) {
return false;
}
return t_thrd.proc->workingVersionNum >= AUDIT_SHA_VERSION_NUM;
}
* Brief : audit sha code
* Description : audit sha code
* the fileds are arraged as below sequence, Note it's not liable to modify them as to keep compatibility of version
* time|type|result|userid|username|dbname|client_info|object_name|detail_info|nodename|threadid|localport|remoteport
*/
static void generateAuditShaCode(AuditShaRecord *shaRecord, char* hexbuf)
{
if (shaRecord == NULL) {
return;
}
unsigned char shacode[SHA256_LENTH + 1] = {0};
size_t textLen;
int rc;
textLen = sizeof(AuditType) + 1 + sizeof(AuditResult) + 1
+ (shaRecord->userid == NULL ? 0 : strlen(shaRecord->userid) + 1)
+ (shaRecord->username == NULL ? 0 : strlen(shaRecord->username) + 1)
+ (shaRecord->dbname == NULL ? 0: strlen(shaRecord->dbname) + 1)
+ (shaRecord->clientInfo == NULL ? 0 : strlen(shaRecord->clientInfo) + 1)
+ (shaRecord->objectName == NULL ? 0 : strlen(shaRecord->objectName) + 1)
+ (shaRecord->detailInfo == NULL ? 0 : strlen(shaRecord->detailInfo) + 1)
+ (shaRecord->nodename == NULL ? 0 : strlen(shaRecord->nodename) + 1)
+ (shaRecord->threadid == NULL ? 0 : strlen(shaRecord->threadid) + 1)
+ (shaRecord->localport == NULL ? 0 : strlen(shaRecord->localport) + 1)
+ (shaRecord->remoteport == NULL ? 0 : strlen(shaRecord->remoteport) + 1);
char* text = NULL;
text = (char*)palloc(textLen);
if (text == NULL) {
return;
}
rc = snprintf_s(text,
textLen,
textLen-1,
"%d%d%s%s%s%s%s%s%s%s%s%s",
shaRecord->type,
shaRecord->result,
shaRecord->userid,
shaRecord->username,
shaRecord->dbname,
shaRecord->clientInfo,
shaRecord->objectName,
shaRecord->detailInfo,
shaRecord->nodename,
shaRecord->threadid,
shaRecord->localport,
shaRecord->remoteport);
securec_check_intval(rc, , );
if (text != NULL) {
SHA256((const unsigned char*)text, strlen((const char*)text), shacode);
sha_bytes_to_hex64((uint8*)shacode, hexbuf);
pfree(text);
}
return;
}
* Main entry point for auditor process
* argc/argv parameters are valid only in EXEC_BACKEND case.
*/
NON_EXEC_STATIC void PgAuditorMain()
{
char auditbuffer[READ_BUF_SIZE + 1] = {0};
int bytes_in_auditbuffer = 0;
int currentAuditRotationAge;
int currentAuditRemainThreshold;
pg_time_t now;
IsUnderPostmaster = true;
t_thrd.proc_cxt.MyProcPid = gs_thread_self();
t_thrd.proc_cxt.MyStartTime = time(NULL);
now = t_thrd.proc_cxt.MyStartTime;
t_thrd.role = AUDITOR;
(void)audit_load_thread_index();
init_ps_display("auditor process", "", "", "");
InitLatch(&t_thrd.audit.sysAuditorLatch);
* Properly accept or ignore signals the postmaster might send us
*
* Note: we ignore all termination signals, and instead exit only when all
* upstream processes are gone, to ensure we don't miss any dying gasps of
* broken backends...
*/
init_audit_signal_handlers();
(void)gspqsignal(SIGURG, print_stack);
if (t_thrd.mem_cxt.pgAuditLocalContext == NULL)
t_thrd.mem_cxt.pgAuditLocalContext = AllocSetContextCreate(t_thrd.top_mem_cxt,
"audit memory context",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE * 3,
ALLOCSET_DEFAULT_MAXSIZE * 3);
on_shmem_exit(pgauditor_kill, (Datum)0);
(void)MemoryContextSwitchTo(t_thrd.mem_cxt.pgAuditLocalContext);
sigjmp_buf local_sigjmp_buf;
int curTryCounter;
int* oldTryCounter = NULL;
if (sigsetjmp(local_sigjmp_buf, 1) != 0) {
gstrace_tryblock_exit(true, oldTryCounter);
pgaudit_handle_exception();
}
oldTryCounter = gstrace_tryblock_entry(&curTryCounter);
t_thrd.log_cxt.PG_exception_stack = &local_sigjmp_buf;
currentAuditRotationAge = u_sess->attr.attr_security.Audit_RotationAge;
currentAuditRemainThreshold = u_sess->attr.attr_security.Audit_RemainThreshold;
if (t_thrd.audit.cur_thread_idx == 0) {
m_curlUtils.initialize(false, "", "", "");
on_proc_exit(pgaudit_exit_clean, 0);
elasic_search_connection_test();
audit_process_cxt_init();
} else {
while (g_instance.audit_cxt.audit_init_done == 0) {
if (pg_atomic_read_u32(&g_instance.audit_cxt.audit_init_done) == 1) {
break;
}
}
}
set_next_rotation_time();
auditfile_init();
for (;;) {
bool time_based_rotation = false;
bool size_based_rotation = false;
long cur_timeout;
unsigned int cur_flags;
unsigned int rc = 0;
ResetLatch(&t_thrd.audit.sysAuditorLatch);
* Quit if we get SIGQUIT from the postmaster.
*/
if (t_thrd.audit.need_exit) {
sig_thread_quit_handler();
break;
}
if (t_thrd.audit.cur_thread_idx == 0) {
policy_auditfile_rotate();
pgaudit_send_data_to_elastic();
}
* Process any requests or signals received recently.
*/
if (t_thrd.audit.got_SIGHUP) {
t_thrd.audit.got_SIGHUP = false;
sig_thread_config_handler(currentAuditRotationAge, currentAuditRemainThreshold);
}
if (pgaudit_need_check_time_rotation()) {
now = (pg_time_t)time(NULL);
if (now >= t_thrd.audit.next_rotation_time)
t_thrd.audit.rotation_requested = time_based_rotation = true;
}
if (pgaudit_need_check_size_rotation()) {
int64 filesize = (t_thrd.audit.sysauditFile != NULL) ? ftell(t_thrd.audit.sysauditFile) : 0;
if (filesize >= (int64)u_sess->attr.attr_security.Audit_RotationSize * 1024L ||
filesize >= (int64)u_sess->attr.attr_security.Audit_SpaceLimit * 1024L) {
t_thrd.audit.rotation_requested = size_based_rotation = true;
}
}
if (t_thrd.audit.rotation_requested) {
* Force rotation when both values are zero. It means the request
* was sent by pg_rotate_auditfile.
*/
if (!time_based_rotation && !size_based_rotation) {
size_based_rotation = true;
}
auditfile_rotate(time_based_rotation, size_based_rotation);
}
if (t_thrd.audit.cur_thread_idx == 0) {
pgaudit_cleanup();
}
* Calculate time till next time-based rotation, so that we don't
* sleep longer than that. We assume the value of "now" obtained
* above is still close enough. Note we can't make this calculation
* until after calling auditfile_rotate(), since it will advance
* t_thrd.audit.next_rotation_time.
*/
if (u_sess->attr.attr_security.Audit_RotationAge > 0 && !t_thrd.audit.rotation_disabled) {
if (now < t_thrd.audit.next_rotation_time)
cur_timeout = (t_thrd.audit.next_rotation_time - now) * 1000L;
else
cur_timeout = 0;
cur_flags = WL_TIMEOUT;
} else {
cur_timeout = -1L;
cur_flags = 0;
}
* Sleep until there's something to do
*/
rc = WaitLatchOrSocket(&t_thrd.audit.sysAuditorLatch, WL_LATCH_SET | WL_SOCKET_READABLE | cur_flags,
g_instance.audit_cxt.sys_audit_pipes[PIPE_READ_INDEX(t_thrd.audit.cur_thread_idx)],
cur_timeout);
if (rc & WL_SOCKET_READABLE) {
int bytesRead = read(g_instance.audit_cxt.sys_audit_pipes[PIPE_READ_INDEX(t_thrd.audit.cur_thread_idx)],
auditbuffer + bytes_in_auditbuffer, sizeof(auditbuffer) - bytes_in_auditbuffer - 1);
if (bytesRead > 0) {
CheckAuditFile();
}
if (bytesRead < 0) {
if (errno != EINTR)
ereport(LOG, (errcode_for_socket_access(), errmsg("could not read from auditor pipe: %m")));
} else if (bytesRead > 0) {
bytes_in_auditbuffer += bytesRead;
process_pipe_input(auditbuffer, &bytes_in_auditbuffer);
continue;
} else {
* Zero bytes read when select() is saying read-ready means
* EOF on the pipe: that is, there are no longer any processes
* with the pipe write end open. Therefore, the postmaster
* and all backends are shut down, and we are done.
*/
t_thrd.audit.pipe_eof_seen = true;
flush_pipe_input(auditbuffer, &bytes_in_auditbuffer);
}
}
if (t_thrd.audit.pipe_eof_seen) {
break;
}
}
ereport(LOG, (errmsg("auditor shutting down")));
if (t_thrd.audit.sysauditFile) {
fclose(t_thrd.audit.sysauditFile);
t_thrd.audit.sysauditFile = NULL;
}
if (t_thrd.audit.policyauditFile) {
fclose(t_thrd.audit.policyauditFile);
t_thrd.audit.policyauditFile = NULL;
}
if (t_thrd.mem_cxt.pgAuditLocalContext != NULL) {
MemoryContextDelete(t_thrd.mem_cxt.pgAuditLocalContext);
t_thrd.mem_cxt.pgAuditLocalContext = NULL;
}
if (t_thrd.audit.cur_thread_idx == 0) {
audit_process_cxt_exit();
}
proc_exit(0);
}
void pgaudit_send_data_to_elastic()
{
if (IS_PGXC_COORDINATOR && g_instance.attr.attr_security.use_elastic_search) {
pgaudit_query_file_for_elastic();
}
}
* Start all audit subprocesses
*/
void pgaudit_start_all(void)
{
if (g_instance.pid_cxt.PgAuditPID == NULL) {
return;
}
for (int i = 0; i < g_instance.audit_cxt.thread_num; ++i) {
if (g_instance.pid_cxt.PgAuditPID[i] == 0) {
g_instance.pid_cxt.PgAuditPID[i] = pgaudit_start();
ereport(LOG, (errmsg("auditor process %d started, pid=%lu", i, g_instance.pid_cxt.PgAuditPID[i])));
}
}
}
* Stop all audit subprocesses
*/
void pgaudit_stop_all(void)
{
for (int i = 0; i < g_instance.audit_cxt.thread_num; ++i) {
if (g_instance.pid_cxt.PgAuditPID[i] != 0) {
Assert(!dummyStandbyMode);
signal_child(g_instance.pid_cxt.PgAuditPID[i], SIGQUIT, -1);
}
}
}
* Postmaster subroutine to start a sysauditor subprocess.
*/
ThreadId pgaudit_start(void)
{
ThreadId sysauditorPid;
if (!u_sess->attr.attr_security.Audit_enabled)
return 0;
sysauditorPid = initialize_util_thread(AUDITOR);
if (sysauditorPid != 0) {
return (ThreadId)sysauditorPid;
}
return 0;
}
void allow_immediate_pgaudit_restart(void)
{
t_thrd.audit.last_pgaudit_start_time = 0;
}
* pipe protocol handling
* --------------------------------
*/
* Process data received through the sysauditor pipe.
*
* This routine interprets the audit pipe protocol which sends audit messages as
* (hopefully atomic) chunks - such chunks are detected and reassembled here.
*
* The protocol has a header that starts with two nul bytes, then has a 16 bit
* length, the pid of the sending process, and a flag to indicate if it is
* the last chunk in a message. Incomplete chunks are saved until we read some
* more, and non-final chunks are accumulated until we get the final chunk.
*
* All of this is to avoid 2 problems:
* . partial messages being written to auditfiles (messes rotation), and
* . messages from different backends being interleaved (messages garbled).
*
* Any non-protocol messages are written out directly. These should only come
* from non-PostgreSQL sources, however (e.g. third party libraries writing to
* stderr).
*
* auditbuffer is the data input buffer, and *bytes_in_auditbuffer is the number
* of bytes present. On exit, any not-yet-eaten data is left-justified in
* auditbuffer, and *bytes_in_auditbuffer is updated.
*/
static void process_pipe_input(char* auditbuffer, int* bytes_in_auditbuffer)
{
char* cursor = auditbuffer;
int count = *bytes_in_auditbuffer;
while (count >= (int)sizeof(PipeProtoHeader)) {
PipeProtoHeader p;
int chunklen;
errno_t errorno = EOK;
errorno = memcpy_s(&p, sizeof(PipeProtoHeader), cursor, sizeof(PipeProtoHeader));
securec_check(errorno, "\0", "\0");
if (p.nuls[0] == '\0' && p.nuls[1] == '\0' && p.len > 0 && p.len <= PIPE_MAX_PAYLOAD && p.pid != 0 &&
(p.is_last == 't' || p.is_last == 'f')) {
List* buffer_list = NULL;
ListCell* cell = NULL;
save_buffer* existing_slot = NULL;
save_buffer* free_slot = NULL;
StringInfo str;
chunklen = PIPE_HEADER_SIZE + p.len;
if (count < chunklen) {
break;
}
buffer_list = t_thrd.audit.buffer_lists[p.pid % NBUFFER_LISTS];
foreach (cell, buffer_list) {
save_buffer* buf = (save_buffer*)lfirst(cell);
if (buf->pid == p.pid) {
existing_slot = buf;
break;
}
if (buf->pid == 0 && free_slot == NULL) {
free_slot = buf;
}
}
if (p.is_last == 'f') {
* Save a complete non-final chunk in a per-pid buffer
*/
if (existing_slot != NULL) {
str = &(existing_slot->data);
appendBinaryStringInfo(str, cursor + PIPE_HEADER_SIZE, p.len);
} else {
if (free_slot == NULL) {
* Need a free slot, but there isn't one in the list,
* so create a new one and extend the list with it.
*/
free_slot = (save_buffer*)palloc(sizeof(save_buffer));
buffer_list = lappend(buffer_list, free_slot);
t_thrd.audit.buffer_lists[p.pid % NBUFFER_LISTS] = buffer_list;
}
free_slot->pid = p.pid;
str = &(free_slot->data);
initStringInfo(str);
appendBinaryStringInfo(str, cursor + PIPE_HEADER_SIZE, p.len);
}
} else {
* Final chunk --- add it to anything saved for that pid, and
* either way write the whole thing out.
*/
if (existing_slot != NULL) {
str = &(existing_slot->data);
appendBinaryStringInfo(str, cursor + PIPE_HEADER_SIZE, p.len);
pgaudit_write_file(str->data, str->len);
pgaudit_write_policy_audit_file(str->data, str->len);
existing_slot->pid = 0;
pfree(str->data);
} else {
pgaudit_write_file(cursor + PIPE_HEADER_SIZE, p.len);
pgaudit_write_policy_audit_file(cursor + PIPE_HEADER_SIZE, p.len);
}
}
cursor += chunklen;
count -= chunklen;
} else {
* Look for the start of a protocol header. If found, dump data
* up to there and repeat the loop. Otherwise, dump it all and
* fall out of the loop. (Note: we want to dump it all if at all
* possible, so as to avoid dividing non-protocol messages across
* auditfiles. We expect that in many scenarios, a non-protocol
* message will arrive all in one read(), and we want to respect
* the read() boundary if possible.)
*/
for (chunklen = 1; chunklen < count; chunklen++) {
if (cursor[chunklen] == '\0') {
break;
}
}
pgaudit_write_file(cursor, chunklen);
pgaudit_write_policy_audit_file(cursor, chunklen);
cursor += chunklen;
count -= chunklen;
}
}
if (count > 0 && cursor != auditbuffer) {
errno_t errorno;
errorno = memmove_s(auditbuffer, READ_BUF_SIZE, cursor, count);
securec_check(errorno, "\0", "\0");
}
*bytes_in_auditbuffer = count;
}
* Force out any buffered data
*
* This is currently used only at sysauditor shutdown, but could perhaps be
* useful at other times, so it is careful to leave things in a clean state.
*/
static void flush_pipe_input(char* auditbuffer, int* bytes_in_auditbuffer)
{
int i;
for (i = 0; i < NBUFFER_LISTS; i++) {
List* list = t_thrd.audit.buffer_lists[i];
ListCell* cell = NULL;
foreach (cell, list) {
save_buffer* buf = (save_buffer*)lfirst(cell);
if (buf->pid != 0) {
StringInfo str = &(buf->data);
pgaudit_write_file(str->data, str->len);
pgaudit_write_policy_audit_file(str->data, str->len);
buf->pid = 0;
pfree(str->data);
}
}
}
* Force out any remaining pipe data as-is; we don't bother trying to
* remove any protocol headers that may exist in it.
*/
if (*bytes_in_auditbuffer > 0) {
pgaudit_write_file(auditbuffer, *bytes_in_auditbuffer);
pgaudit_write_policy_audit_file(auditbuffer, *bytes_in_auditbuffer);
}
*bytes_in_auditbuffer = 0;
}
* auditfile routines
* --------------------------------
*/
* Write data to the currently open auditfile
*
* This is exported so that elog.c can call it when t_thrd.audit.am_sysauditor is true.
* This allows the sysauditor process to record elog messages of its own,
* even though its stderr does not point at the sysaudit pipe.
*/
static void pgaudit_write_file(char* buffer, int count)
{
int rc;
pg_time_t curtime;
errno_t errorno = EOK;
if (buffer == NULL || t_thrd.audit.sysauditFile == NULL)
return;
curtime = time(NULL);
errorno = memcpy_s(buffer + offsetof(AuditMsgHdr, time), READ_BUF_SIZE - offsetof(AuditMsgHdr, time), &curtime,
sizeof(pg_time_t));
securec_check(errorno, "\0", "\0");
errorno = memcpy_s(
buffer + offsetof(AuditMsgHdr, size), READ_BUF_SIZE - offsetof(AuditMsgHdr, size), &count, sizeof(uint32));
securec_check(errorno, "\0", "\0");
errno = 0;
* create a new audit file to avoid the confusion caused by system clock change */
FILE* fh = NULL;
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
bool haslock = true;
if (g_instance.audit_cxt.audit_indextbl) {
AuditIndexItem *cur_item =
g_instance.audit_cxt.audit_indextbl->data +
g_instance.audit_cxt.audit_indextbl->curidx[t_thrd.audit.cur_thread_idx];
if (curtime < cur_item->ctime) {
LWLockRelease(AuditIndexFileLock);
haslock = false;
auditfile_close(SYSAUDITFILE_TYPE);
fh = auditfile_open((pg_time_t)time(NULL), "a", true);
if (fh != NULL) {
t_thrd.audit.sysauditFile = fh;
}
}
}
uint32 retry_cnt = 0;
if (haslock) {
LWLockRelease(AuditIndexFileLock);
}
retry1:
rc = fwrite(buffer, 1, count, t_thrd.audit.sysauditFile);
if (rc != count) {
* If no disk space, we will retry, and we can not report a log as
* there is not space to write.
*/
if (errno == ENOSPC) {
pg_usleep(1000000);
if (retry_cnt % 30 == 0) {
ereport(WARNING, (errmsg("No free space left on audit disk.")));
retry_cnt = 0;
}
retry_cnt += 1;
goto retry1;
}
ereport(ERROR, (errcode_for_file_access(), errmsg("could not write to audit file: %m")));
}
* The contents of the audit logfile haven't newline which is difference from syslog, so
* LBF_MODE set by setvbuf can't make sure the write buffer be fflushed into the logfile
* immediately. We need call the fflush function here by ourself to make sure this.
* NOTICE : in some version of glibc, ftell have the flush feature built-in but it's not
* standard practice to rely ftell to flush, so fflush here is the most assured.
*/
(void)fflush(t_thrd.audit.sysauditFile);
}
static void pgaudit_write_policy_audit_file(const char* buffer, int count)
{
if (!IS_PGXC_COORDINATOR || !g_instance.attr.attr_security.use_elastic_search) {
return;
}
if (t_thrd.audit.policyauditFile == NULL) {
return;
}
uint32 retry_cnt = 0;
retry:
int rc = fwrite(buffer, 1, count, t_thrd.audit.policyauditFile);
if (rc != count) {
* If no disk space, we will retry, and we can not report a log as
* there is not space to write.
*/
if (errno == ENOSPC) {
pg_usleep(1000000);
if (retry_cnt % 30 == 0) {
ereport(WARNING, (errmsg("No free space left on audit disk.")));
retry_cnt = 0;
}
retry_cnt += 1;
goto retry;
}
}
(void)fflush(t_thrd.audit.policyauditFile);
}
* Brief : initialize the audit file.
* Description : set parameter allow_erros as error level, do not allow error as default
*/
static void auditfile_init(bool allow_errors)
{
* The initial auditfile is created right in the postmaster, to verify that
* the Audit_directory is writable.
*/
if (!t_thrd.audit.sysauditFile) {
t_thrd.audit.sysauditFile = auditfile_open(time(NULL), "a", allow_errors);
if (t_thrd.audit.sysauditFile != NULL) {
audit_report(AUDIT_INTERNAL_EVENT, AUDIT_OK, "file", "create a new audit file");
}
}
if (!t_thrd.audit.policyauditFile && IS_PGXC_COORDINATOR && (t_thrd.audit.cur_thread_idx == 0)) {
t_thrd.audit.policyauditFile = auditfile_open(time(NULL), "a", allow_errors, policy_audit_filename, true);
}
}
* Brief : open a new audit file.
* Description :
* Open a new auditfile with proper permissions and buffering options.
*
* If allow_errors is true, we just audit any open failure and return NULL
* (with errno still correct for the fopen failure).
* Otherwise, errors are treated as fatal.
*/
static FILE *auditfile_open(pg_time_t timestamp, const char *mode, bool allow_errors, const char *_filename,
bool ignore_num)
{
FILE *fh = NULL;
char* filename = NULL;
uint32 fnum = 0;
int thread_idx = t_thrd.audit.cur_thread_idx;
struct stat st;
bool exist = false;
if (!ignore_num && g_instance.audit_cxt.audit_indextbl) {
fnum = pgaudit_get_auditfile_num();
}
ereport(DEBUG1, (errmsg("audit thread idx: %d auditfile_open ok fnum : %d", thread_idx, fnum)));
filename = (char*)palloc(MAXPGPATH);
int rc = snprintf_s(
filename, MAXPGPATH, MAXPGPATH - 1, _filename, g_instance.attr.attr_security.Audit_directory, fnum);
securec_check_intval(rc,, NULL);
* Note we do not let pgaudit_filemode disable IWUSR, since we certainly want
* to be able to write the files ourselves.
*/
if (stat(filename, &st) == 0)
exist = true;
fh = fopen(filename, mode);
if (fh != NULL) {
setvbuf(fh, NULL, LBF_MODE, 0);
if (!ignore_num && g_instance.audit_cxt.audit_indextbl) {
pgaudit_update_auditfile_time(timestamp, exist);
pgaudit_update_indexfile(PG_BINARY_W, true);
}
} else {
int save_errno = errno;
ereport(allow_errors ? LOG : FATAL,
(errcode_for_file_access(), errmsg("could not open audit file \"%s\": %m", filename)));
errno = save_errno;
}
if (!exist) {
if (chmod(filename, S_IWUSR | (mode_t)pgaudit_filemode) < 0) {
int save_errno = errno;
ereport(allow_errors ? LOG : FATAL,
(errcode_for_file_access(), errmsg("could not chmod audit file \"%s\": %m", filename)));
errno = save_errno;
}
}
pfree(filename);
return fh;
}
* Brief : close the audit file.
* Description :
*/
static void auditfile_close(AuditFileType flag)
{
if (t_thrd.audit.sysauditFile == NULL)
return;
if ((flag == SYSAUDITFILE_TYPE) && g_instance.audit_cxt.audit_indextbl != NULL) {
pgaudit_switch_next_auditfile();
pgaudit_update_indexfile(PG_BINARY_W, true);
}
if (flag == SYSAUDITFILE_TYPE) {
fclose(t_thrd.audit.sysauditFile);
t_thrd.audit.sysauditFile = NULL;
}
if (flag == POLICYAUDITFILE_TYPE) {
fclose(t_thrd.audit.policyauditFile);
t_thrd.audit.policyauditFile = NULL;
}
return;
}
static void policy_auditfile_rotate()
{
if (t_thrd.audit.policyauditFile == NULL)
return;
pg_time_t fntime = (pg_time_t) time(NULL);
int rc;
* When doing a time-based rotation, invent the new auditfile name based on
* the planned rotation time, not current time, to avoid "slippage" in the
* file name when we don't do the rotation immediately.
*/
int64 filesize = ftell(t_thrd.audit.policyauditFile);
if ((fntime + PolicyAudit_RotationAge) > policy_next_rotation_time && filesize > 0) {
auditfile_close(POLICYAUDITFILE_TYPE);
char oldname[MAXPGPATH] = {0};
rc = snprintf_s(oldname, sizeof(oldname), sizeof(oldname) - 1, "%s" DIR_SEP "0_event.bin",
g_instance.attr.attr_security.Audit_directory);
securec_check_ss(rc, "\0", "\0");
char newfile[MAXPGPATH] = {0};
rc = snprintf_s(newfile, sizeof(oldname), sizeof(oldname) - 1, "%s" DIR_SEP "done" DIR_SEP "%lld_event.bin",
g_instance.attr.attr_security.Audit_directory, (long long)fntime);
securec_check_ss(rc, "\0", "\0");
if (rename(oldname, newfile) != 0) {
ereport(LOG, (errmsg("can't rename \"%s\" to \"%s\": %m", oldname, newfile)));
}
FILE* fh = auditfile_open(fntime, "a", true, policy_audit_filename, true);
if (fh == NULL) {
* ENFILE/EMFILE are not too surprising on a busy system; just
* keep using the old file till we manage to get a new one.
* Otherwise, assume something's wrong with Audit_directory and stop
* trying to create files.
*/
if (errno != ENFILE && errno != EMFILE) {
ereport(LOG,
(errmsg("disabling automatic rotation (use SIGHUP to re-enable)")));
}
return;
}
t_thrd.audit.policyauditFile = fh;
set_next_policy_rotation_time();
}
}
* Brief : perform audit file rotation
* Description :
*/
static void auditfile_rotate(bool time_based_rotation, bool size_based_rotation)
{
pg_time_t fntime;
FILE* fh = NULL;
t_thrd.audit.rotation_requested = false;
* When doing a time-based rotation, invent the new auditfile name based on
* the planned rotation time, not current time, to avoid "slippage" in the
* file name when we don't do the rotation immediately.
*/
if (time_based_rotation)
fntime = t_thrd.audit.next_rotation_time;
else
fntime = time(NULL);
if (time_based_rotation || size_based_rotation) {
auditfile_close(SYSAUDITFILE_TYPE);
fh = auditfile_open(fntime, "a", true);
if (fh == NULL) {
* ENFILE/EMFILE are not too surprising on a busy system; just
* keep using the old file till we manage to get a new one.
* Otherwise, assume something's wrong with Audit_directory and stop
* trying to create files.
*/
if (errno != ENFILE && errno != EMFILE) {
ereport(LOG, (errmsg("disabling automatic rotation (use SIGHUP to re-enable)")));
t_thrd.audit.rotation_disabled = true;
}
return;
}
t_thrd.audit.sysauditFile = fh;
audit_report(AUDIT_INTERNAL_EVENT, AUDIT_OK, "file", "create a new audit file");
}
set_next_rotation_time();
}
static void set_next_policy_rotation_time(void)
{
if (PolicyAudit_RotationAge <= 0) {
return;
}
policy_next_rotation_time = (pg_time_t) time(NULL);
policy_next_rotation_time += PolicyAudit_RotationAge;
}
* Determine the next planned rotation time, and store in t_thrd.audit.next_rotation_time.
*/
static void set_next_rotation_time(void)
{
pg_time_t now;
struct pg_tm* tm = NULL;
int rotinterval;
if (u_sess->attr.attr_security.Audit_RotationAge <= 0)
return;
* The requirements here are to choose the next time > now that is a
* "multiple" of the audit rotation interval. "Multiple" can be interpreted
* fairly loosely. In this version we align to audit_timezone rather than
* GMT.
*/
rotinterval = u_sess->attr.attr_security.Audit_RotationAge * SECS_PER_MINUTE;
now = (pg_time_t)time(NULL);
tm = pg_localtime(&now, log_timezone);
if (tm == NULL) {
ereport(ERROR, (errmsg("pg_localtime must not be null!")));
}
now += tm->tm_gmtoff;
now -= now % rotinterval;
now += rotinterval;
now -= tm->tm_gmtoff;
t_thrd.audit.next_rotation_time = now;
}
void pgaudit_gen_auditfile_warning(pg_time_t remain_time, uint4 filesize)
{
if ((u_sess->attr.attr_security.Audit_CleanupPolicy || remain_time == 0) &&
(g_instance.audit_cxt.pgaudit_totalspace + filesize >=
(uint64)u_sess->attr.attr_security.Audit_SpaceLimit * 1024L))
#ifdef HAVE_LONG_LONG_INT
ereport(WARNING, (errmsg("audit file total space(%lld B) exceed guc parameter(audit_space_limit: %d KB)",
(long long int)(g_instance.audit_cxt.pgaudit_totalspace + filesize),
u_sess->attr.attr_security.Audit_SpaceLimit)));
#else
ereport(WARNING,
(errmsg("audit file total space(%ld B) exceed guc parameter(audit_space_limit: %d KB)",
g_instance.audit_cxt.pgaudit_totalspace + filesize),
u_sess->attr.attr_security.Audit_SpaceLimit)));
#endif
else if (u_sess->attr.attr_security.Audit_CleanupPolicy == 0 && remain_time &&
(g_instance.audit_cxt.pgaudit_totalspace + filesize >=
(uint64)u_sess->attr.attr_security.Audit_SpaceLimit * 1024L))
#ifdef HAVE_LONG_LONG_INT
ereport(WARNING, (errmsg("Based on time-priority policy, the oldest audit file is beyond %d days or "
"audit file total space(%lld B) exceed guc parameter(audit_space_limit: %d KB)",
u_sess->attr.attr_security.Audit_RemainAge,
(long long int)(g_instance.audit_cxt.pgaudit_totalspace + filesize),
u_sess->attr.attr_security.Audit_SpaceLimit)));
#else
ereport(WARNING, (errmsg("Based on time-priority policy, the oldest audit file is beyond %d days or "
"audit file total space(%ld B) exceed guc parameter(audit_space_limit: %d KB)",
u_sess->attr.attr_security.Audit_RemainAge,
(g_instance.audit_cxt.pgaudit_totalspace + filesize),
u_sess->attr.attr_security.Audit_SpaceLimit)));
#endif
if (g_instance.audit_cxt.audit_indextbl->count > (uint32)u_sess->attr.attr_security.Audit_RemainThreshold)
ereport(WARNING,
(errmsg("audit file total count(%u) exceed guc parameter(audit_file_remain_threshold: %d)",
g_instance.audit_cxt.audit_indextbl->count, u_sess->attr.attr_security.Audit_RemainThreshold)));
ereport(WARNING, (errmsg("%s", t_thrd.audit.pgaudit_filepath)));
}
bool should_keep_basedon_timepolicy(pg_time_t remain_time, uint4 filesize, uint32 index, const AuditIndexItem *item)
{
if (g_instance.audit_cxt.audit_indextbl->count <= (uint32)u_sess->attr.attr_security.Audit_RemainThreshold &&
remain_time && (g_instance.audit_cxt.pgaudit_totalspace + filesize <= SPACE_MAXIMUM_SIZE)) {
if ((uint64)(g_instance.audit_cxt.pgaudit_totalspace + filesize -
(uint64)u_sess->attr.attr_security.Audit_SpaceLimit * 1024L) >= t_thrd.audit.space_beyond_size) {
ereport(WARNING,
(errmsg("audit file total space(%lld B) exceed guc parameter(audit_space_limit: %d KB) about %d MB",
(long long int)(g_instance.audit_cxt.pgaudit_totalspace + filesize),
u_sess->attr.attr_security.Audit_SpaceLimit,
(int)(t_thrd.audit.space_beyond_size / (1024 * 1024)))));
t_thrd.audit.space_beyond_size += SPACE_INTERVAL_SIZE;
}
AuditIndexItem *next =
g_instance.audit_cxt.audit_indextbl->data + ((index + 1) % g_instance.audit_cxt.audit_indextbl->maxnum);
if (remain_time >= (g_instance.audit_cxt.audit_indextbl->last_audit_time - item->ctime) ||
(next && (remain_time > (g_instance.audit_cxt.audit_indextbl->last_audit_time - next->ctime)))) {
ereport(WARNING, (errmsg("should_keep_basedon_timepolicy not removed as in remain time")));
return true;
}
}
return false;
}
* pgaudit_cleanup
*
* Check audit data cleanup condition and delete old audit file then returns
*/
static void pgaudit_cleanup(void)
{
uint32 index = 0;
AuditIndexItem* item = NULL;
bool truncated = false;
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (g_instance.audit_cxt.audit_indextbl == NULL) {
LWLockRelease(AuditIndexFileLock);
return;
}
pg_time_t remain_time = (int64)u_sess->attr.attr_security.Audit_RemainAge * SECS_PER_DAY;
uint64 filesize = u_sess->attr.attr_security.Audit_RotationSize * g_instance.audit_cxt.thread_num *
1024L;
index = g_instance.audit_cxt.audit_indextbl->begidx;
while (g_instance.audit_cxt.pgaudit_totalspace + filesize >=
((uint64)u_sess->attr.attr_security.Audit_SpaceLimit * 1024L) ||
g_instance.audit_cxt.audit_indextbl->count > (uint32)u_sess->attr.attr_security.Audit_RemainThreshold) {
errno_t errorno = EOK;
item = g_instance.audit_cxt.audit_indextbl->data + index;
uint32 fnum = item->filenum;
* a. it must be time-based policy and the specified value is valid;
* b. the remained time of oldest audit file is beyond the specified value;
* c. the total size is not beyond the maximum space size.
*/
if (u_sess->attr.attr_security.Audit_CleanupPolicy == 0 &&
should_keep_basedon_timepolicy(remain_time, filesize, index, item)) {
break;
}
int rc = snprintf_s(t_thrd.audit.pgaudit_filepath, MAXPGPATH, MAXPGPATH - 1, pgaudit_filename,
g_instance.attr.attr_security.Audit_directory, item->filenum);
securec_check_intval(rc, , );
struct stat statbuf;
if (stat(t_thrd.audit.pgaudit_filepath, &statbuf) == 0 && unlink(t_thrd.audit.pgaudit_filepath) < 0) {
ereport(WARNING, (errmsg("could not remove audit file: %m")));
break;
}
truncated = true;
pgaudit_gen_auditfile_warning(remain_time, filesize);
if (g_instance.audit_cxt.pgaudit_totalspace >= item->filesize) {
g_instance.audit_cxt.pgaudit_totalspace -= item->filesize;
}
if (g_instance.audit_cxt.audit_indextbl->count > 0) {
--g_instance.audit_cxt.audit_indextbl->count;
}
errorno = memset_s(item, sizeof(AuditIndexItem), 0, sizeof(AuditIndexItem));
securec_check(errorno, "\0", "\0");
if (t_thrd.role != AUDITOR) {
rc = snprintf_truncated_s(t_thrd.audit.pgaudit_filepath, MAXPGPATH, "remove an audit file(number: %u)",
fnum);
securec_check_ss(rc, "\0", "\0");
audit_report(AUDIT_INTERNAL_EVENT, AUDIT_OK, "file", t_thrd.audit.pgaudit_filepath);
}
uint32 earliest_idx = g_instance.audit_cxt.audit_indextbl->latest_idx - g_instance.audit_cxt.thread_num;
if (index == earliest_idx) {
break;
}
g_instance.audit_cxt.audit_indextbl->begidx = (index + 1) % g_instance.audit_cxt.audit_indextbl->maxnum;
index = g_instance.audit_cxt.audit_indextbl->begidx;
}
LWLockRelease(AuditIndexFileLock);
if (truncated) {
pgaudit_update_indexfile(PG_BINARY_W, true);
}
}
* signal handler routines
* --------------------------------
*/
static void pgaudit_exit(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.audit.need_exit = true;
SetLatch(&t_thrd.audit.sysAuditorLatch);
errno = save_errno;
}
static void sigHupHandler(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.audit.got_SIGHUP = true;
SetLatch(&t_thrd.audit.sysAuditorLatch);
errno = save_errno;
}
static void sigUsr1Handler(SIGNAL_ARGS)
{
int save_errno = errno;
t_thrd.audit.rotation_requested = true;
SetLatch(&t_thrd.audit.sysAuditorLatch);
errno = save_errno;
}
* Send data to the syslogger using the chunked protocol
*
* Note: when there are multiple backends writing into the syslogger pipe,
* it's critical that each write go into the pipe indivisibly, and not
* get interleaved with data from other processes. Fortunately, the POSIX
* spec requires that writes to pipes be atomic so long as they are not
* more than PIPE_BUF bytes long. So we divide long messages into chunks
* that are no more than that length, and send one chunk per write() call.
* The collector process knows how to reassemble the chunks.
*
* Because of the atomic write requirement, there are only two possible
* results from write() here: -1 for failure, or the requested number of
* bytes. There is not really anything we can do about a failure; retry would
* probably be an infinite loop, and we can't even report the error usefully.
* (There is noplace else we could send it!) So we might as well just ignore
* the result from write(). However, on some platforms you get a compiler
* warning from ignoring write()'s result, so do a little dance with casting
* rc to void to shut up the compiler.
*/
static void write_pipe_chunks(char* data, int len, AuditClassType type)
{
static volatile uint32 pipe_count = 0;
int thread_num = g_instance.audit_cxt.thread_num;
* for audit process, postgres will send messages to pipes with round-robin
* for unified audit process, just first pipe will used
*/
int last_pipe_count = pg_atomic_fetch_add_u32(&pipe_count, 1);
int cur_pipe_idx = (type == STD_AUDIT_TYPE) ? (((last_pipe_count + thread_num) % thread_num) * 2 + 1) : 1;
ereport(DEBUG1, (errmsg("write_pipe_chunks for pipe : %d", cur_pipe_idx)));
PipeProtoChunk p;
errno_t errorno = EOK;
int rc;
Assert(len > 0);
p.proto.nuls[0] = p.proto.nuls[1] = '\0';
p.proto.pid = t_thrd.proc_cxt.MyProcPid;
while (len > PIPE_MAX_PAYLOAD) {
p.proto.is_last = 'f';
p.proto.len = PIPE_MAX_PAYLOAD;
errorno = memcpy_s(p.proto.data, PIPE_MAX_PAYLOAD, data, PIPE_MAX_PAYLOAD);
securec_check(errorno, "\0", "\0");
rc = write(g_instance.audit_cxt.sys_audit_pipes[cur_pipe_idx], &p, PIPE_HEADER_SIZE + PIPE_MAX_PAYLOAD);
(void)rc;
data += PIPE_MAX_PAYLOAD;
len -= PIPE_MAX_PAYLOAD;
}
p.proto.is_last = 't';
p.proto.len = len;
errorno = memcpy_s(p.proto.data, PIPE_MAX_PAYLOAD, data, len);
securec_check(errorno, "\0", "\0");
rc = write(g_instance.audit_cxt.sys_audit_pipes[cur_pipe_idx], &p, PIPE_HEADER_SIZE + len);
if (rc == -1) {
ereport(ERROR, (errmsg("write into pipe error")));
}
(void)rc;
}
* Brief : append a string field to a streamed data
* Description :
*/
static void appendStringField(StringInfo str, const char* s)
{
int size = 0;
if (str == NULL)
return;
if (s == NULL)
appendBinaryStringInfo(str, (char*)&size, sizeof(int));
else {
size = strlen(s) + 1;
appendBinaryStringInfo(str, (char*)&size, sizeof(int));
appendBinaryStringInfo(str, s, size);
}
}
static pg_time_t current_timestamp()
{
struct timeval te;
gettimeofday(&te, NULL);
pg_time_t milliseconds = te.tv_sec * 1000LL + te.tv_usec / 1000;
return milliseconds;
}
static bool audit_status_check_ok()
{
#ifdef ENABLE_MULTIPLE_NODES
if (!u_sess->attr.attr_security.Audit_enabled || (PGSharedMemoryAttached() && t_thrd.postmaster_cxt.HaShmData &&
(STANDBY_MODE == t_thrd.postmaster_cxt.HaShmData->current_mode ||
PENDING_MODE == t_thrd.postmaster_cxt.HaShmData->current_mode)))
return false;
#else
if (!u_sess->attr.attr_security.Audit_enabled || (PGSharedMemoryAttached() && t_thrd.postmaster_cxt.HaShmData &&
PENDING_MODE == t_thrd.postmaster_cxt.HaShmData->current_mode))
return false;
#endif
return true;
}
* check the valid for specific audit type
*/
static bool audit_type_validcheck(AuditType type)
{
unsigned int type_status = 0;
switch (type) {
case AUDIT_LOGIN_SUCCESS:
type_status = CHECK_AUDIT_LOGIN(SESSION_LOGIN_SUCCESS);
break;
case AUDIT_LOGIN_FAILED:
type_status = CHECK_AUDIT_LOGIN(SESSION_LOGIN_FAILED);
break;
case AUDIT_USER_LOGOUT:
type_status = CHECK_AUDIT_LOGIN(SESSION_LOGOUT);
break;
case AUDIT_SYSTEM_START:
case AUDIT_SYSTEM_STOP:
case AUDIT_SYSTEM_RECOVER:
case AUDIT_SYSTEM_SWITCH:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_ServerAction;
break;
case AUDIT_LOCK_USER:
case AUDIT_UNLOCK_USER:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_LockUser;
break;
case AUDIT_GRANT_ROLE:
case AUDIT_REVOKE_ROLE:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_PrivilegeAdmin;
break;
case AUDIT_USER_VIOLATION:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_UserViolation;
break;
case AUDIT_DDL_DATABASE:
type_status = CHECK_AUDIT_DDL(DDL_DATABASE);
break;
case AUDIT_DDL_DIRECTORY:
type_status = CHECK_AUDIT_DDL(DDL_DIRECTORY);
break;
case AUDIT_DDL_TABLESPACE:
type_status = CHECK_AUDIT_DDL(DDL_TABLESPACE);
break;
case AUDIT_DDL_SCHEMA:
type_status = CHECK_AUDIT_DDL(DDL_SCHEMA);
break;
case AUDIT_DDL_USER:
type_status = CHECK_AUDIT_DDL(DDL_USER);
break;
case AUDIT_DDL_TABLE:
type_status = CHECK_AUDIT_DDL(DDL_TABLE);
break;
case AUDIT_DDL_INDEX:
type_status = CHECK_AUDIT_DDL(DDL_INDEX);
break;
case AUDIT_DDL_VIEW:
type_status = CHECK_AUDIT_DDL(DDL_VIEW);
break;
case AUDIT_DDL_EVENT:
type_status = CHECK_AUDIT_DDL(DDL_EVENT);
break;
case AUDIT_DDL_TRIGGER:
type_status = CHECK_AUDIT_DDL(DDL_TRIGGER);
break;
case AUDIT_DDL_FUNCTION:
type_status = CHECK_AUDIT_DDL(DDL_FUNCTION);
break;
case AUDIT_DDL_PACKAGE:
type_status = CHECK_AUDIT_DDL(DDL_PACKAGE);
break;
case AUDIT_DDL_RESOURCEPOOL:
type_status = CHECK_AUDIT_DDL(DDL_RESOURCEPOOL);
break;
case AUDIT_DDL_GLOBALCONFIG:
type_status = CHECK_AUDIT_DDL(DDL_GLOBALCONFIG);
break;
case AUDIT_DDL_WORKLOAD:
type_status = CHECK_AUDIT_DDL(DDL_WORKLOAD);
break;
case AUDIT_DDL_SERVERFORHADOOP:
type_status = CHECK_AUDIT_DDL(DDL_SERVERFORHADOOP);
break;
case AUDIT_DDL_DATASOURCE:
type_status = CHECK_AUDIT_DDL(DDL_DATASOURCE);
break;
case AUDIT_DDL_NODEGROUP:
type_status = CHECK_AUDIT_DDL(DDL_NODEGROUP);
break;
case AUDIT_DDL_ROWLEVELSECURITY:
type_status = CHECK_AUDIT_DDL(DDL_ROWLEVELSECURITY);
break;
case AUDIT_DDL_SYNONYM:
type_status = CHECK_AUDIT_DDL(DDL_SYNONYM);
break;
case AUDIT_DDL_TYPE:
type_status = CHECK_AUDIT_DDL(DDL_TYPE);
break;
case AUDIT_DDL_TEXTSEARCH:
type_status = CHECK_AUDIT_DDL(DDL_TEXTSEARCH);
break;
case AUDIT_DDL_SEQUENCE:
type_status = CHECK_AUDIT_DDL(DDL_SEQUENCE);
break;
case AUDIT_DDL_KEY:
type_status = CHECK_AUDIT_DDL(DDL_KEY);
break;
case AUDIT_DDL_MODEL:
type_status = CHECK_AUDIT_DDL(DDL_MODEL);
break;
case AUDIT_DDL_PUBLICATION_SUBSCRIPTION:
type_status = CHECK_AUDIT_DDL(DDL_PUBLICATION_SUBSCRIPTION);
break;
case AUDIT_DDL_FOREIGN_DATA_WRAPPER:
type_status = CHECK_AUDIT_DDL(DDL_FOREIGN_DATA_WRAPPER);
break;
case AUDIT_DML_ACTION:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_DML;
break;
case AUDIT_DML_ACTION_SELECT:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_DML_SELECT;
break;
case AUDIT_FUNCTION_EXEC:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_Exec;
break;
case AUDIT_SYSTEM_FUNCTION_EXEC:
type_status = (unsigned int)u_sess->attr.attr_security.audit_system_function_exec;
break;
case AUDIT_POLICY_EVENT:
case MASKING_POLICY_EVENT:
case SECURITY_EVENT:
case AUDIT_INTERNAL_EVENT:
type_status = 1;
break;
case AUDIT_COPY_TO:
case AUDIT_COPY_FROM:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_Copy;
break;
case AUDIT_SET_PARAMETER:
type_status = (unsigned int)u_sess->attr.attr_security.Audit_Set;
break;
case AUDIT_DDL_SQL_PATCH:
type_status = CHECK_AUDIT_DDL(DDL_SQL_PATCH);
break;
case AUDIT_UNKNOWN_TYPE:
default:
type_status = 0;
ereport(WARNING, (errmsg("unknown audit type, discard it.")));
break;
}
if (audit_check_full_audit_user() && type != AUDIT_UNKNOWN_TYPE) {
type_status = 1;
}
return type_status > 0;
}
static bool audit_get_clientinfo(AuditType type, const char* object_name, AuditEventInfo &event_info)
{
* Note that the number of field should keep the same with PGAUDIT_QUERY_COLS
* which will make sure the data size will match with fields number. even though the value of field is null
* it still should be appended to the buf with the size 0.
*/
if (u_sess->proc_cxt.MyProcPort == NULL) {
return true;
}
char *userid = event_info.userid;
const char **username = &(event_info.username);
const char **dbname = &(event_info.dbname);
const char **appname = &(event_info.appname);
const char **remotehost = &(event_info.remotehost);
char *threadid = event_info.threadid;
char *localport = event_info.localport;
char *remoteport = event_info.remoteport;
errno_t errorno = EOK;
if (u_sess->misc_cxt.CurrentUserName != NULL) {
*username = u_sess->misc_cxt.CurrentUserName;
} else {
*username = u_sess->proc_cxt.MyProcPort->user_name;
}
* append user id information, get user id from table as invalid in session
* not safe when access table when run logout process as not in normal transaction
*/
Oid useroid = GetCurrentUserId();
if (type == AUDIT_USER_LOGOUT && !OidIsValid(useroid)) {
return false;
}
if (*username != NULL && useroid == 0) {
ResourceOwner currentOwner = NULL;
currentOwner = t_thrd.utils_cxt.CurrentResourceOwner;
ResourceOwner tmpOwner = ResourceOwnerCreate(t_thrd.utils_cxt.CurrentResourceOwner, "CheckUserOid",
THREAD_GET_MEM_CXT_GROUP(MEMORY_CONTEXT_SECURITY));
t_thrd.utils_cxt.CurrentResourceOwner = tmpOwner;
useroid = get_role_oid(*username, true);
ResourceOwnerRelease(tmpOwner, RESOURCE_RELEASE_BEFORE_LOCKS, true, true);
ResourceOwnerRelease(tmpOwner, RESOURCE_RELEASE_LOCKS, true, true);
ResourceOwnerRelease(tmpOwner, RESOURCE_RELEASE_AFTER_LOCKS, true, true);
t_thrd.utils_cxt.CurrentResourceOwner = currentOwner;
ResourceOwnerDelete(tmpOwner);
}
errorno = snprintf_s(userid, MAXNUMLEN, MAXNUMLEN - 1, "%u", useroid);
securec_check_ss(errorno, "", "");
if (*username == NULL || (**username) == '\0') {
*username = _("[unknown]");
}
*dbname = u_sess->proc_cxt.MyProcPort->database_name;
*appname = u_sess->attr.attr_common.application_name;
*remotehost = u_sess->proc_cxt.MyProcPort->remote_host;
switch (type) {
case AUDIT_POLICY_EVENT:
case MASKING_POLICY_EVENT:
case SECURITY_EVENT:
*remotehost = object_name;
break;
default:
break;
}
t_thrd.audit.user_login_time = GetCurrentTimestamp();
errorno = snprintf_s(threadid,
MAXNUMLEN * 4,
MAXNUMLEN * 4 - 1,
"%lu@%ld",
t_thrd.proc_cxt.MyProcPid,
t_thrd.audit.user_login_time);
securec_check_ss(errorno, "\0", "\0");
int portNum;
if (IsHAPort(u_sess->proc_cxt.MyProcPort)) {
portNum = g_instance.attr.attr_network.PoolerPort;
} else {
portNum = g_instance.attr.attr_network.PostPortNumber;
}
errorno = snprintf_s(localport, MAXNUMLEN, MAXNUMLEN - 1, "%d", portNum);
securec_check_ss(errorno, "\0", "\0");
errorno = snprintf_s(remoteport, MAXNUMLEN, MAXNUMLEN - 1, "%s", u_sess->proc_cxt.MyProcPort->remote_port);
securec_check_ss(errorno, "\0", "\0");
if (*dbname == NULL || (**dbname) == '\0')
*dbname = _("[unknown]");
if (*appname == NULL || (**appname) == '\0')
*appname = _("[unknown]");
if (*remotehost == NULL || (**remotehost) == '\0')
*remotehost = _("[unknown]");
errorno = snprintf_s(event_info.client_info,
MAX_CONNECTION_INFO_SIZE,
MAX_CONNECTION_INFO_SIZE - 1,
"%s@%s",
*appname,
*remotehost);
securec_check_ss(errorno, "\0", "\0");
return true;
}
* Brief : report audit info to the system auditor
* Description : called by all backends, the main routines is as below
* 1. verify type and process to decide whehter to report audit or not
* 2. get all audit info from connection
* 3. append audit info to string buffer
* 4. last, write audit file or send to auditor process to deal with
* the fileds are arraged as below sequence, Note it's not liable to modify them as to keep compatibility of version
* header|userid|username|dbname|client_info|object_name|detail_info|nodename|threadid|localport|remoteport
*/
void audit_report(AuditType type, AuditResult result, const char *object_name, const char *detail_info,
AuditClassType ctype)
{
bool newVersion = false;
if (!audit_status_check_ok() || (detail_info == NULL)) {
return;
}
if (!audit_type_validcheck(type)) {
return;
}
StringInfoData buf;
AuditData adata;
AuditEventInfo event_info;
if (!audit_get_clientinfo(type, object_name, event_info)) {
return;
}
if (audit_check_client_blacklist(event_info.client_info)) {
return;
}
if (pgaudit_need_sha_code()) {
newVersion = true;
}
char *userid = event_info.userid;
const char* username = event_info.username;
const char* dbname = event_info.dbname;
char* client_info = event_info.client_info;
char* threadid = event_info.threadid;
char* localport = event_info.localport;
char* remoteport = event_info.remoteport;
char *detail_info_xid = NULL;
bool audit_xid_info = (u_sess->attr.attr_security.audit_xid_info == 1);
if (audit_xid_info) {
uint32 len = uint64_max_len + strlen("xid=, ") + strlen(detail_info) + 1;
detail_info_xid = (char *)palloc0(len);
audit_append_xid_info(detail_info, detail_info_xid, len);
}
adata.header.signature[0] = 'A';
adata.header.signature[1] = 'U';
adata.header.version = 0;
if (newVersion) {
adata.header.fields = PGAUDIT_QUERY_COLS_NEW;
} else {
adata.header.fields = PGAUDIT_QUERY_COLS;
}
adata.header.flags = AUDIT_TUPLE_NORMAL;
adata.header.time = current_timestamp();
adata.header.size = 0;
adata.type = type;
adata.result = result;
char hexbuf[SHA256_LENTH * ENCRY_LENGTH_DOUBLE + 1] = {0};
AuditShaRecord shaRecord;
if (newVersion) {
shaRecord.type = type;
shaRecord.result = result;
shaRecord.userid = userid;
shaRecord.username = username;
shaRecord.dbname = dbname;
shaRecord.clientInfo = (client_info[0] != '\0') ? client_info : NULL;
shaRecord.objectName = object_name;
shaRecord.detailInfo = (!audit_xid_info) ? detail_info : detail_info_xid;
shaRecord.nodename = g_instance.attr.attr_common.PGXCNodeName;
shaRecord.threadid = (threadid[0] != '\0') ? threadid : NULL;
shaRecord.localport = (localport[0] != '\0') ? localport : NULL;
shaRecord.remoteport = (remoteport[0] != '\0') ? remoteport : NULL;
generateAuditShaCode(&shaRecord, hexbuf);
}
initStringInfo(&buf);
appendBinaryStringInfo(&buf, (char*)&adata, AUDIT_HEADER_SIZE);
appendStringField(&buf, userid);
appendStringField(&buf, username);
appendStringField(&buf, dbname);
appendStringField(&buf, (client_info[0] != '\0') ? client_info : NULL);
appendStringField(&buf, object_name);
appendStringField(&buf, (!audit_xid_info) ? detail_info : detail_info_xid);
appendStringField(&buf, g_instance.attr.attr_common.PGXCNodeName);
appendStringField(&buf, (threadid[0] != '\0') ? threadid : NULL);
appendStringField(&buf, (localport[0] != '\0') ? localport : NULL);
appendStringField(&buf, (remoteport[0] != '\0') ? remoteport : NULL);
if (newVersion) {
appendStringField(&buf, (hexbuf[0] != '\0') ? (const char*)hexbuf : NULL);
}
* Use the chunking protocol if we know the syslogger should be
* catching stderr output, and we are not ourselves the syslogger.
* Otherwise, just do a vanilla write to stderr.
*/
if (WRITE_TO_AUDITPIPE) {
write_pipe_chunks(buf.data, buf.len, ctype);
} else if (WRITE_TO_STDAUDITFILE(ctype)) {
pgaudit_write_file(buf.data, buf.len);
} else if (WRITE_TO_UNIAUDITFILE(ctype)) {
pgaudit_write_policy_audit_file(buf.data, buf.len);
} else if (detail_info != NULL) {
ereport(LOG, (errmsg("discard audit data: %s", (!audit_xid_info) ? detail_info : detail_info_xid)));
}
if (detail_info_xid != NULL) {
pfree(detail_info_xid);
}
if (buf.len > 0) {
pfree(buf.data);
} else {
ereport(LOG, (errmsg("audit buf data empty")));
}
}
static void pgaudit_close_file(FILE* fp, const char* file)
{
if (NULL == file || NULL == fp) {
return;
}
if (ferror(fp)) {
ereport(LOG, (errcode_for_file_access(), errmsg("could not write audit file \"%s\": %m", file)));
if (FreeFile(fp) < 0) {
ereport(LOG, (errcode_for_file_access(), errmsg("could not close audit file \"%s\": %m", file)));
}
} else if (FreeFile(fp) < 0) {
ereport(LOG, (errcode_for_file_access(), errmsg("could not close audit file \"%s\": %m", file)));
}
}
static void pgaudit_read_indexfile(const char* audit_directory)
{
FILE* fp = NULL;
struct stat statbuf;
char tblfile_path[MAXPGPATH] = {0};
size_t nread = 0;
AuditIndexTableNew indextbl;
int rc = snprintf_s(tblfile_path, MAXPGPATH, MAXPGPATH - 1, "%s/%s", audit_directory, audit_indextbl_file);
securec_check_intval(rc,,);
if (pgaudit_find_indexfile()) {
pgaudit_rewrite_indexfile();
}
* Check whether the map file is exist
* there will be no index audit table file here when audit process first init
* return directly and keep the index audit table values NULL, pgaudit_update_indexfile will flush new one
*/
if (stat(tblfile_path, &statbuf) != 0) {
return;
}
fp = AllocateFile(tblfile_path, PG_BINARY_R);
if (NULL == fp) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit index table file \"%s\": %m", tblfile_path)));
return;
}
nread = fread(&indextbl, indextbl_header_size, 1, fp);
if (1 == nread) {
errno_t errorno = EOK;
if (indextbl.maxnum == 0 || indextbl.maxnum > (1024 * 1024 + 1)) {
ereport(ERROR, (errcode(ERRCODE_SYSTEM_ERROR), errmsg("fail to read indextbl maxnum")));
}
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
pfree_ext(g_instance.audit_cxt.audit_indextbl);
g_instance.audit_cxt.audit_indextbl = (AuditIndexTableNew *)MemoryContextAllocZero(
g_instance.audit_cxt.global_audit_context,
(indextbl.maxnum * sizeof(AuditIndexItem) + indextbl_header_size));
errorno =
memcpy_s(g_instance.audit_cxt.audit_indextbl,
indextbl.maxnum * sizeof(AuditIndexItem) + indextbl_header_size, &indextbl, indextbl_header_size);
securec_check(errorno, "\0", "\0");
nread = fread(g_instance.audit_cxt.audit_indextbl->data, sizeof(AuditIndexItem), indextbl.maxnum, fp);
if (nread != indextbl.maxnum) {
ereport(WARNING,
(errcode_for_file_access(), errmsg("could not read audit index file \"%s\": %m", tblfile_path)));
}
LWLockRelease(AuditIndexFileLock);
}
pgaudit_close_file(fp, tblfile_path);
}
* Brief : write the index table into file from memory.
* Description :
*/
static void pgaudit_update_indexfile(const char* mode, bool allow_errors)
{
FILE* fp = NULL;
char tblfile_path[MAXPGPATH] = {0};
size_t nwritten = 0;
size_t count = 0;
int rc = snprintf_s(tblfile_path,
MAXPGPATH,
MAXPGPATH - 1,
"%s/%s",
g_instance.attr.attr_security.Audit_directory,
audit_indextbl_file);
securec_check_intval(rc,,);
fp = AllocateFile(tblfile_path, mode);
if (NULL == fp) {
if (allow_errors) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit index table file \"%s\": %m", tblfile_path)));
return;
} else {
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
pfree_ext(g_instance.audit_cxt.audit_indextbl);
LWLockRelease(AuditIndexFileLock);
ereport(FATAL,
(errcode_for_file_access(), errmsg("could not open audit index table file \"%s\": %m", tblfile_path)));
}
}
pgaudit_indexfile_upgrade();
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (g_instance.audit_cxt.audit_indextbl != NULL) {
count = g_instance.audit_cxt.audit_indextbl->maxnum * sizeof(AuditIndexItem) + indextbl_header_size;
nwritten = fwrite(g_instance.audit_cxt.audit_indextbl, 1, count, fp);
if (nwritten != count)
ereport(allow_errors ? LOG : FATAL,
(errcode_for_file_access(), errmsg("could not write to audit index file: %m")));
}
LWLockRelease(AuditIndexFileLock);
ereport(DEBUG1, (errmsg("pgaudit_update_indexfile index size: %ld", (long)ftell(fp))));
pgaudit_close_file(fp, tblfile_path);
}
static uint32 pgaudit_get_max_fnum(uint32 old_thread_num)
{
uint32 currrent_max_fnum = 0;
for (uint32 i = 0; i < old_thread_num; ++i) {
AuditIndexItem *cur_item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[i];
if (currrent_max_fnum < cur_item->filenum) {
currrent_max_fnum = cur_item->filenum;
}
}
return currrent_max_fnum;
}
* Brief : pgaudit_find_indexfile
* Description : find pgaudit old index file function
*/
static bool pgaudit_find_indexfile(void)
{
struct stat statbuf;
char tblfile_path[MAXPGPATH] = {0};
int rc = snprintf_s(tblfile_path,
MAXPGPATH,
MAXPGPATH - 1,
"%s/%s",
g_instance.attr.attr_security.Audit_directory,
audit_indextbl_old_file);
securec_check_ss(rc, "\0", "\0");
if (stat(tblfile_path, &statbuf) == 0) {
return true;
}
return false;
}
* Brief : pgaudit_indexfile_upgrade
* Description : index table file upgrade function
*/
static void pgaudit_indexfile_upgrade(void)
{
struct stat statbuf;
char tblfile_path[MAXPGPATH] = {0};
if (t_thrd.proc == NULL) {
return;
}
int rc = snprintf_s(tblfile_path,
MAXPGPATH,
MAXPGPATH - 1,
"%s/%s",
g_instance.attr.attr_security.Audit_directory,
audit_indextbl_old_file);
securec_check_intval(rc,,);
ereport(DEBUG1, (errmsg("audit upgrade processing index file upgrade enter")));
if (t_thrd.proc->workingVersionNum >= AUDIT_INDEX_TABLE_VERSION_NUM) {
if (stat(tblfile_path, &statbuf) == 0) {
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (unlink(tblfile_path) < 0) {
ereport(WARNING, (errmsg("could not remove audit old index table file \"%s\": %m", tblfile_path)));
}
LWLockRelease(AuditIndexFileLock);
}
} else {
if (stat(tblfile_path, &statbuf) != 0) {
ereport(WARNING, (errmsg("could not find audit old index table file \"%s\": %m", tblfile_path)));
}
pgaudit_indexfile_sync(PG_BINARY_W, true);
}
}
* Brief : pgaudit_indexfile_sync
* Description : sync old and new index table file function
*/
static void pgaudit_indexfile_sync(const char* mode, bool allow_errors)
{
FILE* fp = NULL;
struct stat statbuf;
char tblfile_path[MAXPGPATH] = {0};
size_t nwritten = 0;
size_t count = 0;
int rc = snprintf_s(tblfile_path, MAXPGPATH, MAXPGPATH - 1, "%s/%s", g_instance.attr.attr_security.Audit_directory,
audit_indextbl_old_file);
securec_check_intval(rc, ,);
ereport(DEBUG1, (errmsg("audit upgrade processing index file sync enter")));
if (stat(tblfile_path, &statbuf) == 0) {
fp = AllocateFile(tblfile_path, mode);
if (NULL == fp) {
ereport(allow_errors ? LOG : FATAL, (errcode_for_file_access(),
errmsg("could not open audit index table file \"%s\": %m", tblfile_path)));
return;
}
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (g_instance.audit_cxt.audit_indextbl != NULL) {
ereport(LOG, (errmsg("audit upgrade processing audit_indextbl != NULL")));
g_instance.audit_cxt.audit_indextbl_old->maxnum = g_instance.audit_cxt.audit_indextbl->maxnum;
g_instance.audit_cxt.audit_indextbl_old->begidx = g_instance.audit_cxt.audit_indextbl->begidx;
g_instance.audit_cxt.audit_indextbl_old->curidx = g_instance.audit_cxt.audit_indextbl->curidx[0];
g_instance.audit_cxt.audit_indextbl_old->count = g_instance.audit_cxt.audit_indextbl->count;
g_instance.audit_cxt.audit_indextbl_old->last_audit_time =
g_instance.audit_cxt.audit_indextbl->last_audit_time;
errno_t errorno = EOK;
errorno = memcpy_s(g_instance.audit_cxt.audit_indextbl_old->data,
g_instance.audit_cxt.audit_indextbl->maxnum * sizeof(AuditIndexItem),
g_instance.audit_cxt.audit_indextbl->data,
g_instance.audit_cxt.audit_indextbl->maxnum * sizeof(AuditIndexItem));
securec_check(errorno, "\0", "\0");
}
if (g_instance.audit_cxt.audit_indextbl_old != NULL) {
count = g_instance.audit_cxt.audit_indextbl_old->maxnum * sizeof(AuditIndexItem) + old_indextbl_header_size;
nwritten = fwrite(g_instance.audit_cxt.audit_indextbl_old, 1, count, fp);
if (nwritten != count)
ereport(allow_errors ? LOG : FATAL,
(errcode_for_file_access(), errmsg("could not write to audit old index file: %m")));
}
LWLockRelease(AuditIndexFileLock);
ereport(LOG, (errmsg("pgaudit_indexfile_sync index size: %ld", (long)ftell(fp))));
pgaudit_close_file(fp, tblfile_path);
}
}
* Brief : pgaudit_rewrite_indexfile
* Description : read old index file and rewrite to new index file function
*/
static void pgaudit_rewrite_indexfile(void)
{
FILE* fp = NULL;
char tblfile_path[MAXPGPATH] = {0};
size_t nread = 0;
AuditIndexTable old_index_tbl;
int rc = snprintf_s(tblfile_path,
MAXPGPATH,
MAXPGPATH - 1,
"%s/%s",
g_instance.attr.attr_security.Audit_directory,
audit_indextbl_old_file);
securec_check_intval(rc,,);
fp = AllocateFile(tblfile_path, PG_BINARY_R);
if (NULL == fp) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit old index table file \"%s\": %m", tblfile_path)));
return;
}
ereport(LOG, (errmsg("audit upgrade processing rewrite enter")));
nread = fread(&old_index_tbl, old_indextbl_header_size, 1, fp);
if (1 == nread) {
errno_t errorno = EOK;
if (old_index_tbl.maxnum == 0 || old_index_tbl.maxnum > (1024 * 1024 + 1)) {
ereport(ERROR, (errcode(ERRCODE_SYSTEM_ERROR), errmsg("fail to read indextbl maxnum")));
}
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
pfree_ext(g_instance.audit_cxt.audit_indextbl);
pfree_ext(g_instance.audit_cxt.audit_indextbl_old);
g_instance.audit_cxt.audit_indextbl_old = (AuditIndexTable *)MemoryContextAllocZero(
g_instance.audit_cxt.global_audit_context,
(old_index_tbl.maxnum * sizeof(AuditIndexItem) + old_indextbl_header_size));
errorno = memcpy_s(g_instance.audit_cxt.audit_indextbl_old,
old_index_tbl.maxnum * sizeof(AuditIndexItem) + old_indextbl_header_size,
&old_index_tbl, old_indextbl_header_size);
securec_check(errorno, "\0", "\0");
g_instance.audit_cxt.audit_indextbl = (AuditIndexTableNew *)MemoryContextAllocZero(
g_instance.audit_cxt.global_audit_context,
(old_index_tbl.maxnum * sizeof(AuditIndexItem) + indextbl_header_size));
g_instance.audit_cxt.audit_indextbl->maxnum = old_index_tbl.maxnum;
g_instance.audit_cxt.audit_indextbl->count = old_index_tbl.count;
g_instance.audit_cxt.audit_indextbl->begidx = old_index_tbl.begidx;
g_instance.audit_cxt.audit_indextbl->last_audit_time = old_index_tbl.last_audit_time;
g_instance.audit_cxt.audit_indextbl->thread_num = 1;
g_instance.audit_cxt.audit_indextbl->curidx[0] = old_index_tbl.curidx;
g_instance.audit_cxt.audit_indextbl->latest_idx = old_index_tbl.curidx + 1;
nread = fread(g_instance.audit_cxt.audit_indextbl->data, sizeof(AuditIndexItem), old_index_tbl.maxnum, fp);
if (nread != old_index_tbl.maxnum) {
ereport(WARNING,
(errcode_for_file_access(), errmsg("could not read audit index file \"%s\": %m", tblfile_path)));
}
LWLockRelease(AuditIndexFileLock);
}
pgaudit_close_file(fp, tblfile_path);
}
* Brief : init index table in memory
* Description :
* 1. init the index file table based on thread num if index table file is not exit
* 2. calculate pgaudit_totalspace based on cuurent audit files
* it's safe no lock here As only PM or audit master thread can invoke init func
*/
static void pgaudit_indextbl_init_new(void)
{
pgaudit_read_indexfile(g_instance.attr.attr_security.Audit_directory);
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (g_instance.audit_cxt.audit_indextbl == NULL) {
ereport(LOG, (errmsg("pgaudit_indextbl_init_new first init")));
g_instance.audit_cxt.audit_indextbl =
(AuditIndexTableNew *)MemoryContextAllocZero(g_instance.audit_cxt.global_audit_context,
(u_sess->attr.attr_security.Audit_RemainThreshold + 1) * sizeof(AuditIndexItem) + indextbl_header_size);
g_instance.audit_cxt.audit_indextbl->maxnum = u_sess->attr.attr_security.Audit_RemainThreshold + 1;
g_instance.audit_cxt.audit_indextbl->count = 0;
g_instance.audit_cxt.audit_indextbl->begidx = 0;
g_instance.audit_cxt.audit_indextbl->thread_num = g_instance.audit_cxt.thread_num;
for (int i = 0; i < g_instance.audit_cxt.thread_num; ++i) {
g_instance.audit_cxt.audit_indextbl->curidx[i] = i;
AuditIndexItem *item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[i];
item->filenum = i;
}
g_instance.audit_cxt.audit_indextbl->latest_idx = g_instance.audit_cxt.thread_num;
}
uint32 index = 0;
AuditIndexItem *item = NULL;
* thread num changed routine
* for thread num enlarge: update latest_idx、curidxes、items and thread_num, if new audit file did not exist,
* audit thread will reinit new one
* for thread num shrink: dismiss the old ones and update curidxes、thread_num but not udpate latest_idx
*/
if (g_instance.audit_cxt.audit_indextbl->thread_num != (uint32)g_instance.audit_cxt.thread_num) {
uint32 old_thread_num = g_instance.audit_cxt.audit_indextbl->thread_num;
uint32 new_thread_num = (uint32)g_instance.audit_cxt.thread_num;
uint32 latest_idx = g_instance.audit_cxt.audit_indextbl->latest_idx;
if (AUDIT_THREADNUM_ENLARGE) {
* before arrage fnums for new audit threads, get the last max fnum
* then increase the fnum based on the max fnum
*/
uint32 old_max_fnum = pgaudit_get_max_fnum(old_thread_num);
uint32 step = new_thread_num - old_thread_num;
for (uint32 i = 0; i < step; ++i) {
uint32 new_idx = old_thread_num + i;
g_instance.audit_cxt.audit_indextbl->curidx[new_idx] = latest_idx + i;
AuditIndexItem *new_item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[new_idx];
new_item->filenum = (old_max_fnum + i + 1);
}
g_instance.audit_cxt.audit_indextbl->latest_idx += step;
} else {
uint32 step = old_thread_num - new_thread_num;
uint32 *curidxes = g_instance.audit_cxt.audit_indextbl->curidx;
errno_t errorno =
memmove_s(curidxes, MAX_AUDIT_NUM * sizeof(uint32), curidxes + step, new_thread_num * sizeof(uint32));
securec_check(errorno, "\0", "\0");
}
g_instance.audit_cxt.audit_indextbl->thread_num = (uint32)g_instance.audit_cxt.thread_num;
}
uint32 earliest_idx = 0;
if ((int32)g_instance.audit_cxt.audit_indextbl->latest_idx >= g_instance.audit_cxt.thread_num) {
earliest_idx = g_instance.audit_cxt.audit_indextbl->latest_idx - g_instance.audit_cxt.thread_num;
index = g_instance.audit_cxt.audit_indextbl->begidx;
} else {
earliest_idx = g_instance.audit_cxt.audit_indextbl->maxnum + g_instance.audit_cxt.audit_indextbl->latest_idx -
g_instance.audit_cxt.thread_num;
index = g_instance.audit_cxt.audit_indextbl->latest_idx;
}
g_instance.audit_cxt.pgaudit_totalspace = 0;
do {
item = g_instance.audit_cxt.audit_indextbl->data + index;
g_instance.audit_cxt.pgaudit_totalspace += item->filesize;
if (index == earliest_idx) {
break;
}
index = (index + 1) % g_instance.audit_cxt.audit_indextbl->maxnum;
} while (true);
t_thrd.audit.space_beyond_size =
(g_instance.audit_cxt.pgaudit_totalspace / SPACE_INTERVAL_SIZE) * SPACE_INTERVAL_SIZE + SPACE_INTERVAL_SIZE;
LWLockRelease(AuditIndexFileLock);
return;
}
static void pgaudit_update_maxnum()
{
errno_t errorno = EOK;
int thread_num = g_instance.attr.attr_security.audit_thread_num;
int new_indextbl_data_lenth = (u_sess->attr.attr_security.Audit_RemainThreshold + 1) * sizeof(AuditIndexItem);
AuditIndexTableNew *new_indextbl = (AuditIndexTableNew *)MemoryContextAllocZero(
g_instance.audit_cxt.global_audit_context, new_indextbl_data_lenth + indextbl_header_size);
new_indextbl->begidx = 0;
new_indextbl->maxnum = u_sess->attr.attr_security.Audit_RemainThreshold + 1;
new_indextbl->last_audit_time = g_instance.audit_cxt.audit_indextbl->last_audit_time;
new_indextbl->thread_num = g_instance.audit_cxt.audit_indextbl->thread_num;
if (g_instance.audit_cxt.audit_indextbl->count > 0) {
AuditIndexItem *item = NULL;
uint32 latest_idx = g_instance.audit_cxt.audit_indextbl->latest_idx;
uint32 last_idx = (latest_idx == 0) ? (g_instance.audit_cxt.audit_indextbl->maxnum - 1): (latest_idx - 1);
uint32 index = g_instance.audit_cxt.audit_indextbl->begidx;
uint32 pos = new_indextbl->begidx;
do {
item = g_instance.audit_cxt.audit_indextbl->data + index;
errorno = memcpy_s(new_indextbl->data + pos, (new_indextbl_data_lenth - (pos * sizeof(AuditIndexItem))),
item, sizeof(AuditIndexItem));
securec_check(errorno, "\0", "\0");
new_indextbl->count++;
* finished copy old index table file from range [begin, latest_idx)
* then update new index table file curidxes
*/
if (index == last_idx) {
for (int i = 0; i < thread_num; ++i) {
new_indextbl->curidx[i] = pos - thread_num + 1 + i;
}
new_indextbl->latest_idx = pos + 1;
break;
}
pos++;
index = (index + 1) % g_instance.audit_cxt.audit_indextbl->maxnum;
} while (true);
}
pfree(g_instance.audit_cxt.audit_indextbl);
g_instance.audit_cxt.audit_indextbl = new_indextbl;
}
* Brief : reset the index file based on new parameter
* Description :
* 1. clean up the current auditfiles
* 2. alloc new index table using new parameter & swap old one
* 3. flush the index table from memory
*/
static void pgaudit_reset_indexfile()
{
uint32 old_maxnum = g_instance.audit_cxt.audit_indextbl->maxnum;
if (old_maxnum > (uint32)u_sess->attr.attr_security.Audit_RemainThreshold + 1) {
int rc = snprintf_s(t_thrd.audit.pgaudit_filepath, MAXPGPATH, MAXPGPATH - 1, "%s/%s",
g_instance.attr.attr_security.Audit_directory, audit_indextbl_file);
securec_check_intval(rc,,);
if (unlink(t_thrd.audit.pgaudit_filepath) < 0)
ereport(WARNING, (errmsg("could not remove audit index table file: %m")));
pgaudit_cleanup();
ereport(LOG, (errmsg("pgaudit pgaudit_reset_indexfile clean up audit files trigger by parameter changing")));
}
if (old_maxnum != (uint32)u_sess->attr.attr_security.Audit_RemainThreshold + 1) {
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
pgaudit_update_maxnum();
LWLockRelease(AuditIndexFileLock);
}
pgaudit_update_indexfile(PG_BINARY_W, true);
}
* Brief : get the specified string field.
* Description :
*/
static const char* pgaudit_string_field(AuditData* adata, int num)
{
int index = 0;
uint32 size = 0;
uint32 datalen = 0;
const char* field = NULL;
if (adata == NULL)
return NULL;
datalen = adata->header.size - AUDIT_HEADER_SIZE;
field = adata->varstr;
do {
errno_t errorno = EOK;
errorno = memcpy_s(&size, sizeof(uint32), field, sizeof(uint32));
securec_check(errorno, "\0", "\0");
datalen -= sizeof(uint32);
if (size > datalen) {
return NULL;
}
field = field + sizeof(uint32);
if (index == num) {
break;
}
field = field + size;
datalen -= size;
index++;
} while (index <= num);
if (size == 0)
return NULL;
return field;
}
static char* serialize_event_to_json(AuditData *adata, long long eventTime)
{
AuditElasticEvent event;
WRITE_JSON_START(AuditElasticEvent, &event);
event.aDataType = AuditTypeDesc(adata->type);
WRITE_JSON_STRING(aDataType);
event.aDataResult = AuditResultDesc(adata->result);
WRITE_JSON_STRING(aDataResult);
event.auditUserId = pgaudit_string_field(adata, AUDIT_USER_ID);
WRITE_JSON_STRING(auditUserId);
event.auditUserName = pgaudit_string_field(adata, AUDIT_USER_NAME);
WRITE_JSON_STRING(auditUserName);
event.auditDatabaseName = pgaudit_string_field(adata, AUDIT_DATABASE_NAME);
WRITE_JSON_STRING(auditDatabaseName);
event.clientConnInfo = pgaudit_string_field(adata, AUDIT_CLIENT_CONNINFO);
WRITE_JSON_STRING(clientConnInfo);
event.objectName = pgaudit_string_field(adata, AUDIT_OBJECT_NAME);
WRITE_JSON_STRING(objectName);
event.detailInfo = pgaudit_string_field(adata, AUDIT_DETAIL_INFO);
WRITE_JSON_STRING(detailInfo);
event.nodeNameInfo = pgaudit_string_field(adata, AUDIT_NODENAME_INFO);
WRITE_JSON_STRING(nodeNameInfo);
event.threadIdInfo = pgaudit_string_field(adata, AUDIT_THREADID_INFO);
WRITE_JSON_STRING(threadIdInfo);
event.localPortInfo = pgaudit_string_field(adata, AUDIT_LOCALPORT_INFO);
WRITE_JSON_STRING(localPortInfo);
event.remotePortInfo = pgaudit_string_field(adata, AUDIT_REMOTEPORT_INFO);
WRITE_JSON_STRING(remotePortInfo);
event.shaCode = pgaudit_string_field(adata, AUDIT_SHACODE);
WRITE_JSON_STRING(shaCode);
event.eventTime = eventTime;
WRITE_JSON_INT(eventTime);
WRITE_JSON_END();
}
struct AuditElasticIndex {
const char* _index;
const char* _type;
};
char* serialize_index_to_json(const char* audit_str)
{
AuditElasticIndex index;
WRITE_JSON_START(AuditElasticIndex, &index);
index._index = audit_str;
WRITE_JSON_STRING(_index);
index._type = audit_str;
WRITE_JSON_STRING(_type);
WRITE_JSON_END();
}
void fix_json(char* json)
{
for (int i = 0; i < (int)strlen(json); i++) {
if (json[i] == '\n') {
json[i] = ' ';
}
if (json[i] == 0) {
break;
}
}
}
static void pgaudit_query_file_for_elastic()
{
AuditMsgHdr header;
AuditData *adata = NULL;
int rc;
char file_path[MAXPGPATH] = {0};
rc = snprintf_s(file_path, sizeof(file_path), sizeof(file_path) - 1, "%s" DIR_SEP "done",
g_instance.attr.attr_security.Audit_directory);
securec_check_ss(rc, "\0", "\0");
std::vector<std::string> done_files;
get_files_list(file_path, done_files, ".bin", MAX_QUEUE_SIZE);
while (!done_files.empty()) {
for (const std::string& file_item : done_files) {
rc = snprintf_s(file_path, sizeof(file_path), sizeof(file_path) - 1, "%s" DIR_SEP "done" DIR_SEP "%s",
g_instance.attr.attr_security.Audit_directory, file_item.c_str());
securec_check_ss(rc, "\0", "\0");
FILE* fp = AllocateFile(file_path, PG_BINARY_R);
if (fp == NULL) {
ereport(WARNING,
(errcode_for_file_access(), errmsg("could not open audit file \"%s\": %m", file_path)));
continue;
}
FILE* bulkfile = fopen("audit_bulk.json", "w");
int event_count = 0;
do {
errno_t errorno = EOK;
int nread = fread((char*)&header, sizeof(AuditMsgHdr), 1, fp);
if (nread == 0) {
break;
}
if (header.signature[0] != 'A' ||
header.signature[1] != 'U' ||
header.version != 0 ||
(header.fields != PGAUDIT_QUERY_COLS && header.fields != PGAUDIT_QUERY_COLS_NEW) ||
(header.size <= sizeof(AuditMsgHdr))) {
ereport(LOG, (errmsg("invalid data in audit file \"%s\"", file_path)));
break;
}
adata = (AuditData *)palloc(header.size);
errorno = memcpy_s(adata, header.size, &header, sizeof(AuditMsgHdr));
securec_check(errorno, "\0", "\0");
nread = fread((char*)adata + sizeof(AuditMsgHdr), header.size - sizeof(AuditMsgHdr), 1, fp);
if (nread != 1) {
ereport(WARNING,
(errcode_for_file_access(), errmsg("could not read audit file \"%s\": %m", file_path)));
pfree(adata);
break;
}
char* json = serialize_event_to_json(adata, adata->header.time);
ereport(DEBUG1, (errmsg("++++++++++++++++++++++++++ Write to JSON adata->type = %d(%s)", adata->type,
AuditTypeDesc(adata->type))));
fix_json(json);
const char* indexType = "audit";
switch (adata->type) {
case AUDIT_POLICY_EVENT:
indexType = "audit_policy";
break;
case MASKING_POLICY_EVENT:
indexType = "masking_policy";
break;
case SECURITY_EVENT:
indexType = "security_management";
break;
default:
break;
}
char* jsonIndex = serialize_index_to_json(indexType);
fix_json(jsonIndex);
if (bulkfile) {
fprintf(bulkfile, "{\"index\":%s}\n", jsonIndex);
fprintf(bulkfile, "%s\n", json);
++event_count;
}
pfree(adata);
} while (true);
if (unlink(file_path)) {
ereport(WARNING, (errmsg("could not unlink file \"%s\": %m", file_path)));
}
if (bulkfile != NULL) {
fclose(bulkfile);
}
pgaudit_close_file(fp, file_path);
if (event_count) {
* total curl command is as below, first for http as sencond for https protocal
* curl -XPOST http://ip:9200/audit/events/_bulk?pretty -H 'Content-type: application/json'
* --data-binary @audit_bulk.json curl -XPOST -k https://ip:9200/audit/events/_bulk?pretty -H
* 'Content-type: application/json' --data-binary @audit_bulk.json
*/
std::string url = ((std::string) g_instance.attr.attr_security.elastic_search_ip_addr) +
((std::string) ":9200/audit/events/_bulk?pretty");
m_curlUtils.http_post_file_request(url, "audit_bulk.json");
}
}
rc = snprintf_s(file_path, sizeof(file_path), sizeof(file_path) - 1, "%s" DIR_SEP "done",
g_instance.attr.attr_security.Audit_directory);
securec_check_ss(rc, "\0", "\0");
done_files.clear();
get_files_list(file_path, done_files, ".bin", MAX_QUEUE_SIZE);
}
}
* Brief : scan the specified audit file into tuple
* Description : Note we use old/new version to differ whether there is user_id field in the file.
* for expanding new field later, maybe we will depend on version id to implement
* backward compatibility but not bool variable
*/
static void deserialization_to_tuple(Datum (&values)[PGAUDIT_QUERY_COLS_NEW],
AuditData *adata,
const AuditMsgHdr &header,
bool nulls[PGAUDIT_QUERY_COLS_NEW],
bool newVersion)
{
AuditShaRecord shaRecord;
const char* saved_hexbuf = NULL;
char hexbuf[SHA256_LENTH * ENCRY_LENGTH_DOUBLE + 1] = {0};
int i = 0;
values[i++] = TimestampTzGetDatum(time_t_to_timestamptz(adata->header.time));
values[i++] = CStringGetTextDatum(AuditTypeDesc(adata->type));
values[i++] = CStringGetTextDatum(AuditResultDesc(adata->result));
shaRecord.type = adata->type;
shaRecord.result = adata->result;
* new format of the audit file under correct record
* the older audit file do not have userid info, so let it to be null
*/
int index_field = 0;
const char* field = NULL;
bool new_version = (header.fields == PGAUDIT_QUERY_COLS || header.fields == PGAUDIT_QUERY_COLS_NEW);
field = new_version ? pgaudit_string_field(adata, index_field++) : NULL;
shaRecord.userid = (char*)field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.username = field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.dbname = field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.clientInfo = (char*)field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.objectName = field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.detailInfo = field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.nodename = field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.threadid = (char*)field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.localport = (char*)field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
field = pgaudit_string_field(adata, index_field++);
shaRecord.remoteport = (char*)field;
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
if (header.fields == PGAUDIT_QUERY_COLS_NEW) {
field = pgaudit_string_field(adata, index_field++);
} else {
field = NULL;
}
values[i++] = CStringGetTextDatum(FILED_NULLABLE(field));
if (newVersion) {
saved_hexbuf = field;
if (header.fields == PGAUDIT_QUERY_COLS_NEW) {
if (saved_hexbuf != NULL && saved_hexbuf[0] != '\0') {
bool verifyResult = false;
generateAuditShaCode(&shaRecord, hexbuf);
if (strcmp((const char*)hexbuf, (const char*)saved_hexbuf) == 0) {
verifyResult = true;
}
values[i++] = BoolGetDatum(verifyResult);
} else {
values[i] = CStringGetTextDatum(FILED_NULLABLE(NULL));
nulls[i++] = true;
}
} else {
values[i] = CStringGetTextDatum(FILED_NULLABLE(NULL));
nulls[i++] = true;
}
}
if (newVersion) {
Assert(i == PGAUDIT_QUERY_COLS_NEW);
} else {
Assert(i == PGAUDIT_QUERY_COLS);
}
}
static void pgaudit_query_file(Tuplestorestate *state, TupleDesc tdesc, uint32 fnum, TimestampTz begtime,
TimestampTz endtime, const char *audit_directory, bool newVersion)
{
FILE* fp = NULL;
size_t nread = 0;
TimestampTz datetime;
AuditMsgHdr header;
AuditData* adata = NULL;
if (state == NULL || tdesc == NULL)
return;
int rcs =
snprintf_s(t_thrd.audit.pgaudit_filepath, MAXPGPATH, MAXPGPATH - 1, pgaudit_filename, audit_directory, fnum);
securec_check_intval(rcs,,);
fp = AllocateFile(t_thrd.audit.pgaudit_filepath, PG_BINARY_R);
if (fp == NULL) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit file \"%s\": %m", t_thrd.audit.pgaudit_filepath)));
return;
}
do {
Datum values[PGAUDIT_QUERY_COLS_NEW] = {0};
bool nulls[PGAUDIT_QUERY_COLS_NEW] = {0};
errno_t errorno = EOK;
* two scenarios tell that the audit file corrupt
* 1. fail to parse the header length
* 2. header encoding is not valid
*/
if (fgetc(fp) == EOF) {
break;
}
(void)fseek(fp, -1, SEEK_CUR);
size_t header_available = fread(&header, sizeof(AuditMsgHdr), 1, fp);
if (header_available != 1 || pgaudit_invalid_header(&header, newVersion)) {
ereport(LOG, (errmsg("invalid data in audit file \"%s\"", t_thrd.audit.pgaudit_filepath)));
pgaudit_mark_corrupt_info(fnum);
break;
}
adata = (AuditData*)palloc(header.size);
errorno = memcpy_s(adata, header.size, &header, sizeof(AuditMsgHdr));
securec_check(errorno, "\0", "\0");
nread = fread((char*)adata + sizeof(AuditMsgHdr), header.size - sizeof(AuditMsgHdr), 1, fp);
if (nread != 1) {
ereport(LOG,
(errcode_for_file_access(),
errmsg("could not read audit file \"%s\": %m", t_thrd.audit.pgaudit_filepath)));
pgaudit_mark_corrupt_info(fnum);
pfree(adata);
break;
}
datetime = time_t_to_timestamptz(adata->header.time);
if (datetime >= begtime && datetime < endtime && header.flags == AUDIT_TUPLE_NORMAL) {
deserialization_to_tuple(values, adata, header, nulls, newVersion);
tuplestore_putvalues(state, tdesc, values, nulls);
}
pfree(adata);
} while (true);
pgaudit_close_file(fp, t_thrd.audit.pgaudit_filepath);
}
* Brief : scan the specified audit file to delete audit.
* Description :
*/
static void pgaudit_delete_file(uint32 fnum, TimestampTz begtime, TimestampTz endtime)
{
int fd = -1;
ssize_t nread = 0;
TimestampTz datetime;
AuditMsgHdr header;
int rc = snprintf_s(t_thrd.audit.pgaudit_filepath,
MAXPGPATH,
MAXPGPATH - 1,
pgaudit_filename,
g_instance.attr.attr_security.Audit_directory,
fnum);
securec_check_intval(rc,,);
fd = open(t_thrd.audit.pgaudit_filepath, O_RDWR, pgaudit_filemode);
if (fd < 0) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit file \"%s\": %m", t_thrd.audit.pgaudit_filepath)));
return;
}
do {
nread = read(fd, &header, sizeof(AuditMsgHdr));
if (nread <= 0)
break;
if (header.signature[0] != 'A' ||
header.signature[1] != 'U' ||
header.version != 0 ||
!(header.fields == (PGAUDIT_QUERY_COLS - 1) ||
header.fields == PGAUDIT_QUERY_COLS ||
header.fields == PGAUDIT_QUERY_COLS_NEW)) {
ereport(LOG, (errmsg("invalid data in audit file \"%s\"", t_thrd.audit.pgaudit_filepath)));
break;
}
datetime = time_t_to_timestamptz(header.time);
if (datetime >= begtime && datetime < endtime && header.flags == AUDIT_TUPLE_NORMAL) {
long offset = sizeof(AuditMsgHdr);
header.flags = AUDIT_TUPLE_DEAD;
if (lseek(fd, -offset, SEEK_CUR) < 0) {
ereport(WARNING, (errcode_for_file_access(), errmsg("could not seek in audit file: %m")));
break;
}
if (write(fd, &header, sizeof(AuditMsgHdr)) != sizeof(AuditMsgHdr)) {
ereport(WARNING, (errcode_for_file_access(), errmsg("could not write to audit file: %m")));
break;
}
}
if (lseek(fd, header.size - sizeof(AuditMsgHdr), SEEK_CUR) < 0) {
ereport(WARNING, (errcode_for_file_access(), errmsg("could not seek in audit file: %m")));
break;
}
} while (true);
close(fd);
}
static bool pgaudit_check_system(TimestampTz begtime, TimestampTz endtime, uint32 index, const char *audit_dir = NULL)
{
bool satisfied = false;
TimestampTz curr_filetime = 0;
TimestampTz next_filetime = 0;
AuditIndexItem* item = t_thrd.audit.audit_indextbl->data + index;
uint32 earliest_idx = t_thrd.audit.audit_indextbl->latest_idx - g_instance.audit_cxt.thread_num;
if (item->ctime > 0) {
curr_filetime = time_t_to_timestamptz(item->ctime);
if ((index >= earliest_idx && index < t_thrd.audit.audit_indextbl->latest_idx)) {
if (curr_filetime <= endtime) {
satisfied = true;
}
} else {
item = t_thrd.audit.audit_indextbl->data + (index + 1) % t_thrd.audit.audit_indextbl->maxnum;
if (item->ctime > 0) {
next_filetime = time_t_to_timestamptz(item->ctime);
* check whether the time quantum between begtime and endtime
* intersect with the time quantum between curr_filetime and next_filetime
*/
curr_filetime = (curr_filetime > begtime) ? curr_filetime : begtime;
next_filetime = (next_filetime < endtime) ? next_filetime : endtime;
if (curr_filetime <= next_filetime) {
satisfied = true;
} else {
* under multi-thread situation
*/
audit_dir = (audit_dir == NULL) ? g_instance.attr.attr_security.Audit_directory : audit_dir;
TimestampTz curr_headertime = pgaudit_headertime(index, audit_dir);
TimestampTz next_headertime = pgaudit_headertime(index + 1, audit_dir);
if (curr_headertime <= next_headertime) {
satisfied = true;
}
}
} else if (curr_filetime <= begtime || curr_filetime <= endtime) {
satisfied = true;
}
}
} else {
satisfied = true;
}
return satisfied;
}
static TimestampTz pgaudit_headertime(uint32 fnum, const char *audit_directory)
{
int fd = -1;
ssize_t nread = 0;
TimestampTz datetime;
AuditMsgHdr header;
char pgaudit_filepath[MAXPGPATH];
int rc = snprintf_s(pgaudit_filepath, MAXPGPATH, MAXPGPATH - 1, pgaudit_filename,
audit_directory, fnum);
securec_check_intval(rc, , time_t_to_timestamptz(0));
fd = open(pgaudit_filepath, O_RDWR, pgaudit_filemode);
if (fd < 0) {
ereport(LOG,
(errcode_for_file_access(), errmsg("could not open audit file \"%s\": %m", pgaudit_filepath)));
return time_t_to_timestamptz(0);
}
nread = read(fd, &header, sizeof(AuditMsgHdr));
if (nread <= 0) {
close(fd);
return time_t_to_timestamptz(0);
}
datetime = time_t_to_timestamptz(header.time);
close(fd);
return datetime;
}
* Brief : whether the invoke is allowed for query audit.
* Description :
*/
static void pgaudit_query_valid_check(const ReturnSetInfo *rsinfo, FunctionCallInfoData *fcinfo, TupleDesc &tupdesc, bool newVersion)
{
Oid roleid = InvalidOid;
roleid = GetUserId();
if (!has_auditadmin_privilege(roleid)) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("permission denied to query audit")));
}
if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo)) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("set-valued function called in context that cannot accept a set")));
}
if (!((unsigned int)rsinfo->allowedModes & SFRM_Materialize)) {
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("materialize mode required, but it is not allowed in this context")));
}
if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) {
ereport(ERROR, (errcode(ERRCODE_SYSTEM_ERROR), errmsg("return type must be a row type")));
}
if (newVersion) {
if (tupdesc->natts != PGAUDIT_QUERY_COLS_NEW) {
ereport(ERROR, (errcode(ERRCODE_SYSTEM_ERROR), errmsg("attribute count of the return row type not matched")));
}
} else {
if (tupdesc->natts != PGAUDIT_QUERY_COLS) {
ereport(ERROR, (errcode(ERRCODE_SYSTEM_ERROR), errmsg("attribute count of the return row type not matched")));
}
}
}
* Brief : query audit information between begin time and end time.
* Description :
*/
Datum pg_query_audit(PG_FUNCTION_ARGS)
{
ReturnSetInfo* rsinfo = (ReturnSetInfo*)fcinfo->resultinfo;
TupleDesc tupdesc = NULL;
Tuplestorestate* tupstore = NULL;
MemoryContext per_query_ctx = NULL;
MemoryContext oldcontext = NULL;
MemoryContext query_audit_ctx = NULL;
TimestampTz begtime = PG_GETARG_TIMESTAMPTZ(0);
TimestampTz endtime = PG_GETARG_TIMESTAMPTZ(1);
char* audit_dir = NULL;
char real_audit_dir[PATH_MAX] = {0};
bool newVersion = false;
if (pgaudit_need_sha_code()) {
newVersion = true;
}
pgaudit_query_valid_check(rsinfo, fcinfo, tupdesc, newVersion);
* When g_instance.audit_cxt.audit_indextbl is not NULL,
* but its origin memory context is NULL, free it will generate core
*/
if (PG_NARGS() == PG_QUERY_AUDIT_ARGS_MAX) {
audit_dir = text_to_cstring(PG_GETARG_TEXT_PP(PG_QUERY_AUDIT_ARGS_MAX - 1));
}
audit_dir = (audit_dir == NULL) ? g_instance.attr.attr_security.Audit_directory : audit_dir;
if (realpath(audit_dir, real_audit_dir) == NULL) {
ereport(ERROR, (errmsg("Failed to canonicalization path of audit_directory.")));
}
* load the index audit table from global index audit table instance
* then use the local thread one when iterate all audit files
*/
pgaudit_read_indexfile(real_audit_dir);
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
if (g_instance.audit_cxt.audit_indextbl != NULL) {
int indextbl_len =
(u_sess->attr.attr_security.Audit_RemainThreshold + 1) * sizeof(AuditIndexItem) + indextbl_header_size;
t_thrd.audit.audit_indextbl = (AuditIndexTableNew *)palloc0(indextbl_len);
error_t errorno =
memcpy_s(t_thrd.audit.audit_indextbl, indextbl_len, g_instance.audit_cxt.audit_indextbl, indextbl_len);
securec_check(errorno, "\0", "\0");
}
LWLockRelease(AuditIndexFileLock);
per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
oldcontext = MemoryContextSwitchTo(per_query_ctx);
tupstore = tuplestore_begin_heap(true, false, u_sess->attr.attr_memory.work_mem);
rsinfo->returnMode = SFRM_Materialize;
rsinfo->setResult = tupstore;
rsinfo->setDesc = tupdesc;
MemoryContextSwitchTo(oldcontext);
query_audit_ctx = AllocSetContextCreate(per_query_ctx, "query audit file", ALLOCSET_DEFAULT_SIZES);
if (begtime < endtime && t_thrd.audit.audit_indextbl != NULL && t_thrd.audit.audit_indextbl->count > 0) {
bool satisfied = false;
uint32 index = 0;
uint32 fnum = 0;
AuditIndexItem* item = NULL;
index = t_thrd.audit.audit_indextbl->begidx;
do {
item = t_thrd.audit.audit_indextbl->data + index;
fnum = item->filenum;
satisfied = pgaudit_check_system(begtime, endtime, index, real_audit_dir);
if (satisfied) {
oldcontext = MemoryContextSwitchTo(query_audit_ctx);
pgaudit_query_file(tupstore, tupdesc, fnum, begtime, endtime, real_audit_dir, newVersion);
MemoryContextSwitchTo(oldcontext);
MemoryContextReset(query_audit_ctx);
satisfied = false;
}
ereport(DEBUG5, (errmsg("pg_query_audit current fnum: %u", fnum)));
if (index == (t_thrd.audit.audit_indextbl->latest_idx - 1)) {
break;
}
index = (index + 1) % t_thrd.audit.audit_indextbl->maxnum;
} while (true);
}
pfree_ext(t_thrd.audit.audit_indextbl);
MemoryContextDelete(query_audit_ctx);
tuplestore_donestoring(tupstore);
return (Datum)0;
}
* Brief : delete audit information between begin time and end time.
* Description :
*/
Datum pg_delete_audit(PG_FUNCTION_ARGS)
{
TimestampTz begtime = PG_GETARG_TIMESTAMPTZ(0);
TimestampTz endtime = PG_GETARG_TIMESTAMPTZ(1);
t_thrd.audit.Audit_delete = true;
Oid roleid = GetUserId();
if (!has_auditadmin_privilege(roleid)) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("permission denied to delete audit")));
}
* load the index audit table from global index audit table instance
* then use the local thread one when iterate all audit files
*/
pgaudit_read_indexfile(g_instance.attr.attr_security.Audit_directory);
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
if (g_instance.audit_cxt.audit_indextbl != NULL) {
int indextbl_len =
(u_sess->attr.attr_security.Audit_RemainThreshold + 1) * sizeof(AuditIndexItem) + indextbl_header_size;
t_thrd.audit.audit_indextbl = (AuditIndexTableNew *)palloc0(indextbl_len);
error_t errorno =
memcpy_s(t_thrd.audit.audit_indextbl, indextbl_len, g_instance.audit_cxt.audit_indextbl, indextbl_len);
securec_check(errorno, "\0", "\0");
}
LWLockRelease(AuditIndexFileLock);
if (begtime < endtime && (t_thrd.audit.audit_indextbl != NULL) && t_thrd.audit.audit_indextbl->count > 0) {
bool satisfied = false;
uint32 index;
uint32 fnum;
AuditIndexItem* item = NULL;
index = t_thrd.audit.audit_indextbl->begidx;
do {
item = t_thrd.audit.audit_indextbl->data + index;
fnum = item->filenum;
satisfied = pgaudit_check_system(begtime, endtime, index);
if (satisfied) {
pgaudit_delete_file(fnum, begtime, endtime);
satisfied = false;
}
if (index == t_thrd.audit.audit_indextbl->latest_idx - 1) {
break;
}
index = (index + 1) % t_thrd.audit.audit_indextbl->maxnum;
} while (true);
}
pfree_ext(t_thrd.audit.audit_indextbl);
PG_RETURN_VOID();
}
* if use_elastic_search is set on, user should make sure connection is ok,
* or process will not start successfully
*/
static void elasic_search_connection_test()
{
if (!g_instance.attr.attr_security.use_elastic_search) {
return;
}
std::string url = ((std::string) g_instance.attr.attr_security.elastic_search_ip_addr) +
((std::string) ":9200/audit/events/_bulk?pretty");
(void)m_curlUtils.http_post_file_request(url, "", true);
}
* check and reinit the audit files
* 1. whether the audit file is exist
* 2. recognize the corrupt file
* 3. try to reinit audit file if sysauditFile is NULL
*/
static void CheckAuditFile(void)
{
uint32 fnum = 0;
AuditIndexItem *item = NULL;
struct stat statBuf;
errno_t rc;
int thread_idx = t_thrd.audit.cur_thread_idx;
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
item = g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[thread_idx];
fnum = item->filenum;
LWLockRelease(AuditIndexFileLock);
rc = snprintf_s(t_thrd.audit.pgaudit_filepath, MAXPGPATH, MAXPGPATH - 1, pgaudit_filename,
g_instance.attr.attr_security.Audit_directory, fnum);
securec_check_ss(rc, "\0", "\0");
* If the current write file is not exist, we'll reinit the sysauditFile.
* Also, when we querying the auditfile and finding invalid header file,
* we'll truncate it. And in there we'll reinit it too.
* allow error here as process have been rnning but not startup
*/
if (lstat(t_thrd.audit.pgaudit_filepath, &statBuf) == -1 ||
(lstat(t_thrd.audit.pgaudit_filepath, &statBuf) == 0 && statBuf.st_size == 0)) {
if (t_thrd.audit.sysauditFile != NULL) {
fclose(t_thrd.audit.sysauditFile);
t_thrd.audit.sysauditFile = NULL;
}
}
uint32 corrupt_audit_fnum = pg_atomic_read_u32(&g_instance.audit_cxt.audit_coru_fnum[thread_idx]);
if (corrupt_audit_fnum != UINT32_MAX && corrupt_audit_fnum == fnum) {
ereport(WARNING, (errmsg("invalid data in audit file fnum %d", fnum)));
int fd = open(t_thrd.audit.pgaudit_filepath, O_RDWR | O_TRUNC, pgaudit_filemode);
if (fd < 0) {
ereport(ERROR, (errcode_for_file_access(),
errmsg("could not truncate audit file \"%s\": %m", t_thrd.audit.pgaudit_filepath)));
} else {
close(fd);
}
if (t_thrd.audit.sysauditFile != NULL) {
fclose(t_thrd.audit.sysauditFile);
t_thrd.audit.sysauditFile = NULL;
}
pg_atomic_write_u32(&g_instance.audit_cxt.audit_coru_fnum[thread_idx], UINT32_MAX);
}
* make sure init audit file if pgaudit_filepath accessable
* directly return when sysauditFile exist
*/
auditfile_init(true);
}
static bool pgaudit_invalid_header(const AuditMsgHdr* header, bool newVersion)
{
if (newVersion) {
return ((header->signature[0]) != 'A' || header->signature[1] != 'U' || header->version != 0 ||
!(header->fields == (PGAUDIT_QUERY_COLS - 1) || header->fields == PGAUDIT_QUERY_COLS || header->fields == PGAUDIT_QUERY_COLS_NEW) ||
(header->size <= sizeof(AuditMsgHdr)) ||
(header->size >= (uint32)u_sess->attr.attr_security.Audit_RotationSize * 1024L));
} else {
return ((header->signature[0]) != 'A' || header->signature[1] != 'U' || header->version != 0 ||
!(header->fields == (PGAUDIT_QUERY_COLS - 1) || header->fields == PGAUDIT_QUERY_COLS) ||
(header->size <= sizeof(AuditMsgHdr)) ||
(header->size >= (uint32)u_sess->attr.attr_security.Audit_RotationSize * 1024L));
}
}
* mark corrupt fnum by postgres thread
* used for reinit audit files in audit thread
*/
static void pgaudit_mark_corrupt_info(uint32 fnum)
{
* only the writing audit files could be mark corrupt info
* ignore the old audit files here
*/
int thread_num = g_instance.audit_cxt.thread_num;
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
* iterate the all writing index looking for the thread idx of fnum
*/
int thread_idx = -1;
for (int i = 0; i < thread_num; ++i) {
AuditIndexItem *item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[i];
if (fnum == item->filenum) {
thread_idx = i;
break;
}
}
LWLockRelease(AuditIndexFileLock);
if (thread_idx == -1) {
return;
}
ereport(WARNING, (errmsg("audit file num %d is corrupted.", fnum)));
* if any other thread have updated the audit_fnum or fnum is older one, do nothing but break here
*/
uint32 audit_corrupt_fnum = pg_atomic_read_u32(&g_instance.audit_cxt.audit_coru_fnum[thread_idx]);
if (audit_corrupt_fnum < fnum || audit_corrupt_fnum == UINT32_MAX) {
while (!pg_atomic_compare_exchange_u32(&g_instance.audit_cxt.audit_coru_fnum[thread_idx], &audit_corrupt_fnum,
fnum)) {
audit_corrupt_fnum = pg_atomic_read_u32(&g_instance.audit_cxt.audit_coru_fnum[thread_idx]);
if (audit_corrupt_fnum >= fnum && audit_corrupt_fnum != UINT32_MAX) {
break;
}
}
}
}
static void audit_append_xid_info(const char *detail_info, char *detail_info_xid, uint32 len)
{
Assert(u_sess->attr.attr_security.audit_xid_info == 1);
int rc = 0;
TransactionId xid = InvalidTransactionId;
if (IsTransactionState() && !RecoveryInProgress() &&
!(t_thrd.xlog_cxt.LocalXLogInsertAllowed == 0 && g_instance.streaming_dr_cxt.isInSwitchover == true)) {
xid = GetCurrentTransactionId();
rc = snprintf_s(detail_info_xid, len, len - 1, "xid=%llu, %s", xid, detail_info);
securec_check_ss(rc, "\0", "\0");
} else {
rc = snprintf_s(detail_info_xid, len, len - 1, "xid=NA, %s", detail_info);
securec_check_ss(rc, "\0", "\0");
}
}
* Brief : audit process init for multi-thread manage
* Description : init audit global env for audit threads including
* 1. index file lock
* 2. pipes for audit
* 3. audit logs & path
* 4. audit index file
*/
static void audit_process_cxt_init()
{
Assert(t_thrd.role == AUDITOR);
Assert(t_thrd.audit.cur_thread_idx == 0);
if (g_instance.audit_cxt.audit_init_done == 1) {
return;
}
ereport(LOG, (errmsg("audit_process_cxt_init enter")));
errno_t errorno = 0;
int thread_num = g_instance.attr.attr_security.audit_thread_num;
MemoryContext oldcontext = MemoryContextSwitchTo(g_instance.audit_cxt.global_audit_context);
g_instance.audit_cxt.thread_num = thread_num;
if (g_instance.audit_cxt.sys_audit_pipes == NULL) {
g_instance.audit_cxt.sys_audit_pipes =
(int *)palloc0(sizeof(int) * 2 * thread_num);
int *&pipes = g_instance.audit_cxt.sys_audit_pipes;
errorno = memset_s(pipes, sizeof(pipes), -1, sizeof(pipes));
securec_check(errorno, "\0", "\0");
for (int i = 0; i < thread_num; ++i) {
if (pipe(&pipes[PIPE_READ_INDEX(i)]) < 0) {
ereport(FATAL, (errcode_for_socket_access(), (errmsg("could not create pipe for sysaudit: %m"))));
}
ereport(LOG, (errmsg("audit_process_cxt_init pipe init successfully for pipe : %d file descriptor: %d", i,
g_instance.audit_cxt.sys_audit_pipes[i * 2])));
}
}
char Audit_directory_Done[MAXPGPATH] = {0};
int rc = snprintf_s(Audit_directory_Done, sizeof(Audit_directory_Done), sizeof(Audit_directory_Done) - 1, "%s/done",
g_instance.attr.attr_security.Audit_directory);
securec_check_ss(rc, "\0", "\0");
(void)pg_mkdir_p(g_instance.attr.attr_security.Audit_directory, S_IRWXU);
(void)pg_mkdir_p(Audit_directory_Done, S_IRWXU);
g_instance.audit_cxt.audit_indextbl = NULL;
pgaudit_indextbl_init_new();
pgaudit_update_indexfile(PG_BINARY_A, false);
g_instance.audit_cxt.audit_init_done = 1;
(void)MemoryContextSwitchTo(oldcontext);
ereport(LOG, (errmsg("audit_process_cxt_init success")));
}
* Brief : audit process exit
* Description : when exit the audit thread in PM thread, release related pipes
* audit master thread will do the index audit file flush job, not do it here
*/
static void audit_process_cxt_exit()
{
Assert(t_thrd.role == AUDITOR);
if (g_instance.audit_cxt.audit_init_done == 0) {
return;
}
g_instance.audit_cxt.audit_init_done = 0;
int thread_num = g_instance.attr.attr_security.audit_thread_num;
int *sys_audit_pipe = g_instance.audit_cxt.sys_audit_pipes;
while (true) {
if (pg_atomic_read_u32(&g_instance.audit_cxt.current_audit_index) == 1) {
break;
}
}
for (int i = 0; i < thread_num; ++i) {
if (sys_audit_pipe[PIPE_READ_INDEX(i)] > 0) {
close(sys_audit_pipe[PIPE_READ_INDEX(i)]);
sys_audit_pipe[PIPE_READ_INDEX(i)] = -1;
}
}
pfree_ext(g_instance.audit_cxt.sys_audit_pipes);
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
pfree_ext(g_instance.audit_cxt.audit_indextbl);
LWLockRelease(AuditIndexFileLock);
}
* Brief : load audit thread index
* Description :
*/
int audit_load_thread_index()
{
if (t_thrd.audit.cur_thread_idx != -1) {
return t_thrd.audit.cur_thread_idx;
}
int idx = pg_atomic_fetch_add_u32(&g_instance.audit_cxt.current_audit_index, 1);
t_thrd.audit.cur_thread_idx = idx;
Assert(t_thrd.audit.cur_thread_idx >= 0);
Assert(t_thrd.audit.cur_thread_idx < g_instance.audit_cxt.thread_num);
return t_thrd.audit.cur_thread_idx;
}
* Brief : get audit file num for current thread index
* Description :
*/
uint32 pgaudit_get_auditfile_num()
{
Assert(t_thrd.role == AUDITOR);
uint32 fnum = 0;
int thread_idx = t_thrd.audit.cur_thread_idx;
LWLockAcquire(AuditIndexFileLock, LW_SHARED);
AuditIndexItem *item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[thread_idx];
fnum = item->filenum;
LWLockRelease(AuditIndexFileLock);
return fnum;
}
* Brief : update index table file when new file openned
* Description : flush index table file will just invoke later
*/
void pgaudit_update_auditfile_time(pg_time_t timestamp, bool exist)
{
int thread_idx = t_thrd.audit.cur_thread_idx;
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
if (!exist) {
AuditIndexItem *item =
g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[thread_idx];
item->ctime = timestamp;
++g_instance.audit_cxt.audit_indextbl->count;
}
LWLockRelease(AuditIndexFileLock);
}
* Brief : iterate the next audit file & update the index table info
* Description : calc the total space here when rotate one audit file
*/
void pgaudit_switch_next_auditfile()
{
AuditIndexItem *item = NULL;
uint32 new_fnum = 0;
int thread_idx = t_thrd.audit.cur_thread_idx;
LWLockAcquire(AuditIndexFileLock, LW_EXCLUSIVE);
item = g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[thread_idx];
item->filesize = ftell(t_thrd.audit.sysauditFile);
pg_time_t curtime = time(NULL);
g_instance.audit_cxt.audit_indextbl->last_audit_time = curtime;
g_instance.audit_cxt.pgaudit_totalspace += item->filesize;
uint32 current_max_fnum = pgaudit_get_max_fnum(g_instance.audit_cxt.thread_num);
uint32 new_idx = pg_atomic_fetch_add_u32(&g_instance.audit_cxt.audit_indextbl->latest_idx, 1);
g_instance.audit_cxt.audit_indextbl->latest_idx =
(g_instance.audit_cxt.audit_indextbl->latest_idx) % g_instance.audit_cxt.audit_indextbl->maxnum;
g_instance.audit_cxt.audit_indextbl->curidx[thread_idx] = (new_idx) % g_instance.audit_cxt.audit_indextbl->maxnum;
item = g_instance.audit_cxt.audit_indextbl->data + g_instance.audit_cxt.audit_indextbl->curidx[thread_idx];
item->filenum = ++current_max_fnum;
LWLockRelease(AuditIndexFileLock);
ereport(DEBUG1, (errmsg("pgaudit_switch_next_auditfile new fnum :%d cur pgaudit_totalspace: %ld MB", new_fnum,
g_instance.audit_cxt.pgaudit_totalspace)));
}
* Brief : do the thread index decreasing when thread exit
* Description :
*/
static void pgauditor_kill(int code, Datum arg)
{
pg_atomic_fetch_sub_u32(&g_instance.audit_cxt.current_audit_index, 1);
}
* Brief : check auditor thread
* Description :
*/
bool pg_auditor_thread(ThreadId pid)
{
if (g_instance.pid_cxt.PgAuditPID == NULL) {
return false;
}
for (int i = 0; i < g_instance.audit_cxt.thread_num; ++i) {
if (pid == g_instance.pid_cxt.PgAuditPID[i]) {
return true;
}
}
return false;
}