Lab 09: Database Auditing — Triggers, Logs & Compliance¶

CREATE SCHEMA IF NOT EXISTS audit;

CREATE TABLE audit.event_log (
  id           BIGSERIAL    PRIMARY KEY,
  event_time   TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
  schema_name  TEXT         NOT NULL,
  table_name   TEXT         NOT NULL,
  operation    TEXT         NOT NULL
                            CHECK (operation IN ('INSERT','UPDATE','DELETE','SELECT')),
  db_user      TEXT         NOT NULL DEFAULT current_user,
  app_user     TEXT,                          -- set by application layer via SET LOCAL
  client_addr  INET,                          -- captured from pg_stat_activity if needed
  -- Change data
  record_id    BIGINT,
  old_data     JSONB,                         -- full row before change (UPDATE/DELETE)
  new_data     JSONB,                         -- full row after change  (INSERT/UPDATE)
  changed_cols TEXT[]                         -- column names that actually changed (UPDATE only)
);

-- IMMUTABILITY: revoke all access, then grant only INSERT and SELECT
REVOKE ALL ON audit.event_log FROM PUBLIC;

GRANT INSERT ON audit.event_log TO PUBLIC;   -- triggers can always write audit records
GRANT SELECT ON audit.event_log TO PUBLIC;   -- anyone can review the audit trail

-- No UPDATE or DELETE granted to anyone (superusers bypass this, see Part 3)

CREATE OR REPLACE FUNCTION audit.capture_change()
RETURNS TRIGGER AS $$
DECLARE
  v_old_data JSONB;
  v_new_data JSONB;
  v_changed  TEXT[];
  v_rec_id   BIGINT;
BEGIN
  -- ----------------------------------------------------------------
  -- Capture old and new row images as JSONB
  -- ----------------------------------------------------------------
  IF TG_OP = 'DELETE' THEN
    v_old_data := row_to_json(OLD)::JSONB;
    v_new_data := NULL;
    v_rec_id   := (row_to_json(OLD) ->> 'id')::BIGINT;

  ELSIF TG_OP = 'INSERT' THEN
    v_old_data := NULL;
    v_new_data := row_to_json(NEW)::JSONB;
    v_rec_id   := (row_to_json(NEW) ->> 'id')::BIGINT;

  ELSIF TG_OP = 'UPDATE' THEN
    v_old_data := row_to_json(OLD)::JSONB;
    v_new_data := row_to_json(NEW)::JSONB;
    v_rec_id   := (row_to_json(NEW) ->> 'id')::BIGINT;

    -- Identify which columns actually changed (skip unchanged columns)
    SELECT array_agg(old_j.key)
    INTO   v_changed
    FROM   jsonb_each(v_old_data) old_j
    WHERE  old_j.value IS DISTINCT FROM (v_new_data -> old_j.key);
  END IF;

  -- ----------------------------------------------------------------
  -- Write the audit record
  -- ----------------------------------------------------------------
  INSERT INTO audit.event_log (
    schema_name,    table_name,  operation,
    db_user,        record_id,
    old_data,       new_data,    changed_cols
  ) VALUES (
    TG_TABLE_SCHEMA, TG_TABLE_NAME, TG_OP,
    current_user,    v_rec_id,
    v_old_data,      v_new_data,   v_changed
  );

  -- Return NEW for INSERT/UPDATE, OLD for DELETE (PostgreSQL requirement)
  RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Audit all DML on the patients PHI table (from Lab 07)
CREATE TRIGGER audit_patients
  AFTER INSERT OR UPDATE OR DELETE
  ON secure_data.patients
  FOR EACH ROW
  EXECUTE FUNCTION audit.capture_change();

-- Audit all DML on the employees HR table (from Lab 08)
CREATE TRIGGER audit_employees
  AFTER INSERT OR UPDATE OR DELETE
  ON hr.employees
  FOR EACH ROW
  EXECUTE FUNCTION audit.capture_change();

-- 1. INSERT: adds a test employee
INSERT INTO hr.employees (name, email, department, ssn, salary)
VALUES ('Test User', 'test@corp.com', 'IT', '999-99-9999', 65000);

-- 2. UPDATE: raises salary by $5,000
UPDATE hr.employees
SET    salary = 70000
WHERE  name   = 'Test User';

-- 3. DELETE: removes the test employee
DELETE FROM hr.employees
WHERE  name = 'Test User';

-- 4. Review the complete audit trail
SELECT
  event_time,
  table_name,
  operation,
  db_user,
  record_id,
  old_data  ->> 'salary'  AS old_salary,
  new_data  ->> 'salary'  AS new_salary,
  changed_cols
FROM audit.event_log
WHERE table_name = 'employees'
ORDER BY id;

-- Create a low-privilege analyst role for testing tamper resistance
DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'user_analyst') THEN
    CREATE ROLE user_analyst LOGIN PASSWORD 'Analyst_Test#2026' NOINHERIT;
  END IF;
END;
$$;

GRANT USAGE ON SCHEMA audit TO user_analyst;
GRANT SELECT ON audit.event_log TO user_analyst;
-- Note: no UPDATE or DELETE granted

SET ROLE user_analyst;

-- Attempt 1: delete an audit record
DELETE FROM audit.event_log LIMIT 1;

RESET ROLE;

SET ROLE user_analyst;

-- Attempt 2: update an audit record
UPDATE audit.event_log SET db_user = 'nobody' WHERE id = 1;

RESET ROLE;

-- Compliance report: salary increases exceeding 20% (possible fraud indicator)
SELECT
  event_time,
  db_user,
  record_id,
  (old_data ->> 'salary')::NUMERIC                                   AS old_salary,
  (new_data ->> 'salary')::NUMERIC                                   AS new_salary,
  ROUND(
    100.0
    * ABS(  (new_data ->> 'salary')::NUMERIC
          - (old_data ->> 'salary')::NUMERIC)
    / (old_data ->> 'salary')::NUMERIC,
    1
  )                                                                  AS pct_change
FROM audit.event_log
WHERE table_name = 'employees'
  AND operation  = 'UPDATE'
  AND old_data ? 'salary'
  AND ABS(  (new_data ->> 'salary')::NUMERIC
          - (old_data ->> 'salary')::NUMERIC)
      / (old_data ->> 'salary')::NUMERIC > 0.20
ORDER BY pct_change DESC;

SELECT
  table_name,
  operation,
  COUNT(*)                           AS event_count,
  COUNT(DISTINCT db_user)            AS distinct_users,
  MIN(event_time)                    AS first_event,
  MAX(event_time)                    AS last_event
FROM audit.event_log
GROUP BY table_name, operation
ORDER BY table_name, operation;

-- VERIFY LAB 09
SELECT
  -- Audit table exists in audit schema
  (SELECT COUNT(*)
     FROM pg_tables
    WHERE schemaname = 'audit'
      AND tablename  = 'event_log')::INT                              AS audit_table_exists,

  -- Trigger on hr.employees is active
  (SELECT COUNT(*)
     FROM pg_trigger
    WHERE tgname = 'audit_employees')::INT                           AS trigger_exists,

  -- At least 3 employee audit events exist (INSERT + UPDATE + DELETE)
  (SELECT COUNT(*)
     FROM audit.event_log
    WHERE table_name = 'employees')::INT                             AS employee_events,

  -- All three DML operations are represented
  (SELECT COUNT(DISTINCT operation)
     FROM audit.event_log
    WHERE table_name = 'employees')::INT                             AS distinct_operations,

  -- At least one UPDATE has changed_cols populated
  (SELECT COUNT(*)
     FROM audit.event_log
    WHERE operation    = 'UPDATE'
      AND changed_cols IS NOT NULL)::INT                             AS updates_with_changed_cols;