自定义SQL实现PostgreSQL安全审计

create extension supa_audit cascade;
create table public.account(
id int primary key,
name text not null
);

select audit.enable_tracking('public.account'::regclass);

insert into public.account(id, name)
values (1, 'Chongchong');
update public.account set name = 'CC' where id = 1;
delete from public.account where id = 1;

truncate table public.account;

select * from audit.record_history

select audit.disable_tracking('public.account'::regclass);

create schema if not exists audit;

create table audit.record_version(
id bigserial primary key,
record_id uuid,
old_record_id uuid,
op varchar(8) not null,
ts timestamptz not null default now(),
table_oid oid not null,
table_schema name not null,
table_name name not null,
record jsonb,
old_record jsonb
);

create index record_version_ts
on audit.record_version
using brin(ts);

create index record_version_table_oid
on audit.record_version
using btree(table_oid);

[table_oid, primary_key_value_1, primary_key_value_2, ...]

create or replace function audit.primary_key_columns(entity_oid oid)
returns text[]
stable
security definer
language sql
as $$
-- Looks up the names of a table's primary key columns
select
coalesce(
array_agg(pa.attname::text order by pa.attnum),
array[]::text[]
) column_names
from
pg_index pi
join pg_attribute pa
on pi.indrelid = pa.attrelid
and pa.attnum = any(pi.indkey)
where
indrelid = $1
and indisprimary
$$;

create or replace function audit.to_record_id(
entity_oid oid,
pkey_cols text[],
rec jsonb
)
returns uuid
stable
language sql
as $$
select
case
when rec is null then null
-- if no primary key exists, use a random uuid
when pkey_cols = array[]::text[] then uuid_generate_v4()
else (
select
uuid_generate_v5(
'fd62bc3d-8d6e-43c2-919c-802ba3762271',
(
jsonb_build_array(to_jsonb($1))
|| jsonb_agg($3 ->> key_)
)::text
)
from
unnest($2) x(key_)
)
end
$$;

create index record_version_record_id
on audit.record_version(record_id)
where record_id is not null;
create index record_version_old_record_id
on audit.record_version(record_id)
where old_record_id is not null;

create or replace function audit.insert_update_delete_trigger()
returns trigger
security definer
language plpgsql
as $$
declare
pkey_cols text[] = audit.primary_key_columns(TG_RELID);
record_jsonb jsonb = to_jsonb(new);
record_id uuid = audit.to_record_id(TG_RELID, pkey_cols, record_jsonb);
old_record_jsonb jsonb = to_jsonb(old);
old_record_id uuid = audit.to_record_id(TG_RELID, pkey_cols, old_record_jsonb);
begin
insert into audit.record_version(
record_id,
old_record_id,
op,
table_oid,
table_schema,
table_name,
record,
old_record
)
select
record_id,
old_record_id,
TG_OP,
TG_RELID,
TG_TABLE_SCHEMA,
TG_TABLE_NAME,
record_jsonb,
old_record_jsonb;
return coalesce(new, old);
end;
$$;

select audit.enable_tracking('.
'::regclass);
禁用跟踪：
select audit.disable_tracking('.
'::regclass);
这些函数根据请求由表注册审计触发器：
create or replace function audit.enable_tracking(regclass)
returns void
volatile
security definer
language plpgsql
as $$
declare
statement_row text = format('
create trigger audit_i_u_d
before insert or update or delete
on %I
for each row
execute procedure audit.insert_update_delete_trigger();',
$1
);
pkey_cols text[] = audit.primary_key_columns($1);
begin
if pkey_cols = array[]::text[] then
raise exception 'Table % can not be audited because it has no primary key', $1;
end if;
if not exists(select 1 from pg_trigger where tgrelid = $1 and tgname = 'audit_i_u_d') then
execute statement_row;
end if;
end;
$$;
create or replace function audit.disable_tracking(regclass)
returns void
volatile
security definer
language plpgsql
as $$
declare
statement_row text = format(
'drop trigger if exists audit_i_u_d on %I;',
$1
);
begin
execute statement_row;
end;
$$;
性能开销
开启审计表后会降低插入、更新和删除的吞吐量。但是在吞吐量低于每秒1000次写入的情况下，其开销通常可以忽略不计。对于写入频率较高的表，建议使用pgAudit。
总结
通过简单纯sql语句就实现了Postgresql数据库的安全审计，总体上算起来实现才150行sql语句。大家可以自己手动尝试一下，主要是搞清楚其原理，如果生产环境中有需求还是建议用pgAudit。