Audit Trails and Forensics

CREATE TABLE audit_log (
    id            BIGSERIAL PRIMARY KEY,
    at            TIMESTAMPTZ NOT NULL DEFAULT now(),

    -- WHO
    actor_id        UUID,
    actor_type      TEXT NOT NULL,          -- user | service | system | anonymous
    impersonated_by UUID,                   -- the REAL actor, when relevant
    tenant_id       UUID,

    -- WHAT
    event         TEXT NOT NULL,            -- from a fixed vocabulary
    target_type   TEXT,
    target_id     TEXT,
    outcome       TEXT NOT NULL,            -- success | denied | error

    -- WHERE
    ip            INET,
    user_agent    TEXT,
    request_id    UUID,
    session_tag   TEXT,                     -- a truncated HASH, never the id

    -- WHAT CHANGED (mutations only, and scrubbed of secrets)
    changes       JSONB,                    -- { field: { from, to } }
    metadata      JSONB,

    -- Tamper evidence
    prev_hash     TEXT,
    hash          TEXT NOT NULL
);

-- Append only. The application role may INSERT and SELECT, nothing else.
REVOKE UPDATE, DELETE ON audit_log FROM app_rw;
CREATE RULE audit_no_update AS ON UPDATE TO audit_log DO INSTEAD NOTHING;
CREATE RULE audit_no_delete AS ON DELETE TO audit_log DO INSTEAD NOTHING;

-- Partition by month: retention becomes DROP PARTITION, and queries prune.
CREATE TABLE audit_log_2026_08 PARTITION OF audit_log
  FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');

-- Investigation indexes
CREATE INDEX ON audit_log (actor_id, at DESC);
CREATE INDEX ON audit_log (target_type, target_id, at DESC);
CREATE INDEX ON audit_log (ip, at DESC);
CREATE INDEX ON audit_log (tenant_id, at DESC);
CREATE INDEX ON audit_log (event, at DESC);
CREATE INDEX ON audit_log USING gin (metadata jsonb_path_ops);

-- Capture before-and-after on mutations, so a dispute has an answer.
--   changes: {"role": {"from": "user", "to": "admin"},
--             "email": {"from": "[email protected]", "to": "[email protected]"}}
-- Scrub secrets first: never a password hash, a token, or a key.

-- Write these as saved queries BEFORE you need them. During an incident,
-- nobody wants to compose SQL.

-- 1. Full actor timeline
SELECT at, event, target_type, target_id, outcome, ip
  FROM audit_log
 WHERE actor_id = :actor
   AND at BETWEEN :from AND :to
 ORDER BY at;

-- 2. What was actually accessed, summarised (this drives disclosure)
SELECT target_type, count(*) AS accesses, count(DISTINCT target_id) AS distinct_records,
       min(at) AS first, max(at) AS last
  FROM audit_log
 WHERE actor_id = :actor AND outcome = 'success'
   AND at BETWEEN :from AND :to
 GROUP BY target_type ORDER BY distinct_records DESC;

-- 3. The exact record ids — the difference between "some records" and a list
SELECT DISTINCT target_id
  FROM audit_log
 WHERE actor_id = :actor AND target_type = 'invoice' AND outcome = 'success'
   AND at BETWEEN :from AND :to;

-- 4. Which tenants are affected (per-customer notification)
SELECT tenant_id, count(DISTINCT target_id) AS records
  FROM audit_log
 WHERE actor_id = :actor AND outcome = 'success' AND at BETWEEN :from AND :to
 GROUP BY tenant_id ORDER BY records DESC;

-- 5. When did it really start? (usually earlier than the alert)
SELECT min(at)
  FROM audit_log a
 WHERE a.actor_id = :actor
   AND NOT EXISTS (SELECT 1 FROM known_ips k
                    WHERE k.user_id = a.actor_id AND k.ip = a.ip
                      AND k.first_seen < a.at - interval '7 days');

-- 6. Everyone who touched a specific record
SELECT at, actor_id, impersonated_by, event, outcome, ip
  FROM audit_log
 WHERE target_type = 'invoice' AND target_id = :id
 ORDER BY at;

-- 7. Lateral movement: other accounts from the same addresses
SELECT DISTINCT actor_id
  FROM audit_log
 WHERE ip IN (SELECT DISTINCT ip FROM audit_log WHERE actor_id = :actor)
   AND actor_id <> :actor;

-- 8. Denials before success — the shape of enumeration
SELECT date_trunc('minute', at) AS minute,
       count(*) FILTER (WHERE outcome = 'denied')  AS denied,
       count(*) FILTER (WHERE outcome = 'success') AS succeeded
  FROM audit_log
 WHERE actor_id = :actor AND at BETWEEN :from AND :to
 GROUP BY 1 ORDER BY 1;
-- Many denials then a success is the signature of finding a boundary.

-- 9. Verify the chain over the incident window
SELECT * FROM verify_audit_chain(:from_id);

// Logging every read is prohibitive. Logging none makes disclosure a guess.
// Log by data classification and by access pattern.

const READ_AUDIT_POLICY = {
  // Always — regulated or high-risk data
  always: ['patient_record', 'payment_method', 'national_id', 'medical_note'],

  // Bulk only — individual reads are normal, volume is the signal
  bulkOnly: ['invoice', 'customer', 'order'],
  bulkThreshold: 50,

  // Cross-boundary — reading something outside the actor's usual scope
  crossBoundary: true,

  // Privileged actors — admin and support reads are always interesting
  privilegedActors: ['admin', 'support'],

  // Never — public and non-sensitive
  never: ['product', 'plan', 'public_page'],
};

export async function auditRead(req, { type, ids }) {
  const list = [].concat(ids);
  const should =
    READ_AUDIT_POLICY.always.includes(type) ||
    (READ_AUDIT_POLICY.bulkOnly.includes(type) &&
     list.length >= READ_AUDIT_POLICY.bulkThreshold) ||
    READ_AUDIT_POLICY.privilegedActors.includes(req.user?.role) ||
    req.crossedBoundary;

  if (!should) return;

  // For bulk reads, record the SET rather than one row per record — the ids
  // are what matter, and one entry keeps the volume manageable.
  await audit.record({
    event: 'data.read',
    actor: { id: req.user.id, type: 'user', role: req.user.role,
             impersonatedBy: req.sessionData?.impersonatedBy,
             tenantId: req.user.tenant },
    target: { type, count: list.length },
    context: { ip: req.ip, userAgent: req.get('user-agent'), requestId: req.id },
    metadata: {
      // Cap the stored list; the count is always accurate.
      ids: list.slice(0, 1000),
      truncated: list.length > 1000,
      query: req.query,      // scrubbed — how they found these records
    },
  });
}

// Usage
app.get('/api/invoices', auth, async (req, res) => {
  const rows = await listInvoices(req.user, req.query);
  await auditRead(req, { type: 'invoice', ids: rows.map((r) => r.id) });
  res.json(rows.map(InvoiceSerializer.public));
});

// Storing the query alongside the ids is what lets an investigator reconstruct
// HOW the attacker found the records — which is usually how you find the
// vulnerability they used.