Audit logging: rastreando mudanças nos dados

1. Fundamentos do Audit Logging em Bancos de Dados

Audit logging é o processo de registrar todas as operações que alteram dados em um banco de dados, criando um histórico imutável de quem fez o quê e quando. Os principais objetivos incluem rastreabilidade para investigação de incidentes, conformidade regulatória e segurança contra acessos não autorizados.

É crucial distinguir três tipos de logs:

  • Logs de transação: registram operações internas do banco para recovery (ex: WAL no PostgreSQL)
  • Logs de aplicação: gerados pelo software, focam em eventos de negócio
  • Logs de auditoria: capturam alterações nos dados com contexto completo (usuário, timestamp, valores antigos/novos)

Regulamentações como SOX (Sarbanes-Oxley), GDPR (General Data Protection Regulation) e HIPAA (Health Insurance Portability and Accountability Act) exigem que alterações em dados sensíveis sejam rastreáveis. Isso impacta o design do banco, exigindo tabelas de auditoria dedicadas e políticas de retenção específicas.

2. Estratégias de Implementação no SQL

A abordagem mais comum utiliza triggers para capturar operações INSERT, UPDATE e DELETE. Exemplo básico de trigger de auditoria:

CREATE OR REPLACE FUNCTION audit_trigger_function()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO audit_log(table_name, operation, old_data, new_data, user_name, timestamp)
        VALUES (TG_TABLE_NAME, 'INSERT', NULL, row_to_json(NEW), current_user, NOW());
        RETURN NEW;
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO audit_log(table_name, operation, old_data, new_data, user_name, timestamp)
        VALUES (TG_TABLE_NAME, 'UPDATE', row_to_json(OLD), row_to_json(NEW), current_user, NOW());
        RETURN NEW;
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO audit_log(table_name, operation, old_data, new_data, user_name, timestamp)
        VALUES (TG_TABLE_NAME, 'DELETE', row_to_json(OLD), NULL, current_user, NOW());
        RETURN OLD;
    END IF;
END;
$$ LANGUAGE plpgsql;

A estrutura da tabela de auditoria deve ser padronizada:

CREATE TABLE audit_log (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100) NOT NULL,
    operation VARCHAR(10) NOT NULL,
    old_data JSONB,
    new_data JSONB,
    user_name VARCHAR(100),
    timestamp TIMESTAMPTZ DEFAULT NOW(),
    ip_address INET,
    session_id VARCHAR(100)
);

Para bancos PostgreSQL, o tipo JSONB ou HSTORE é ideal para armazenar diffs de forma flexível, permitindo consultas sobre campos específicos.

3. Captura de Metadados Essenciais

Cada registro de auditoria deve conter:

  • Usuário: current_user captura o usuário da sessão atual
  • Timestamp preciso: NOW() retorna o timestamp do início da transação; clock_timestamp() retorna o tempo real, útil para operações concorrentes
  • Endereço IP: inet_client_addr() captura o IP da conexão
  • Tipo de operação: TG_OP dentro da trigger identifica INSERT/UPDATE/DELETE
  • Chave primária: essencial para rastrear qual registro foi alterado

Exemplo de trigger que captura a chave primária dinamicamente:

CREATE OR REPLACE FUNCTION audit_with_pk()
RETURNS TRIGGER AS $$
DECLARE
    pk_value TEXT;
BEGIN
    EXECUTE format('SELECT ($1).%I::TEXT', TG_ARGV[0]) INTO pk_value USING NEW;

    INSERT INTO audit_log(table_name, operation, pk_column, pk_value, old_data, new_data, user_name, timestamp, ip_address)
    VALUES (TG_TABLE_NAME, TG_OP, TG_ARGV[0], pk_value, 
            CASE WHEN TG_OP IN ('UPDATE','DELETE') THEN row_to_json(OLD) ELSE NULL END,
            CASE WHEN TG_OP IN ('INSERT','UPDATE') THEN row_to_json(NEW) ELSE NULL END,
            current_user, NOW(), inet_client_addr());
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

4. Técnicas de Versionamento de Linhas (Row Versioning)

Uma alternativa às triggers é usar tabelas temporais, onde cada linha possui validade temporal:

CREATE TABLE orders_history (
    order_id INT,
    status VARCHAR(20),
    amount DECIMAL(10,2),
    valid_from TIMESTAMPTZ NOT NULL,
    valid_to TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (order_id, valid_from)
);

No SQL Server, SYSTEM_VERSIONING automatiza isso:

CREATE TABLE orders (
    order_id INT PRIMARY KEY,
    status VARCHAR(20),
    amount DECIMAL(10,2),
    valid_from DATETIME2 GENERATED ALWAYS AS ROW START,
    valid_to DATETIME2 GENERATED ALWAYS AS ROW END,
    PERIOD FOR SYSTEM_TIME (valid_from, valid_to)
) WITH (SYSTEM_VERSIONING = ON (HISTORY_TABLE = dbo.orders_history));

Para bancos com alto volume, o particionamento por período (ex: mensal) facilita a manutenção e consultas históricas.

5. Performance e Otimização

Índices são cruciais em tabelas de auditoria. Recomenda-se:

CREATE INDEX idx_audit_table_name ON audit_log(table_name);
CREATE INDEX idx_audit_timestamp ON audit_log(timestamp);
CREATE INDEX idx_audit_user ON audit_log(user_name);
CREATE INDEX idx_audit_operation ON audit_log(operation);

Para consultas por chave primária, um índice composto é eficiente:

CREATE INDEX idx_audit_pk ON audit_log(table_name, pk_value, timestamp);

Particionamento por data usando pg_partman (PostgreSQL) ou partitioned tables (SQL Server) permite descartar partições antigas sem DELETE pesado:

CREATE TABLE audit_log_2024_01 PARTITION OF audit_log
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

Trade-off entre batch e trigger: triggers garantem consistência imediata mas adicionam latência a cada operação. Para alta taxa de escrita, considere escrever logs em lote via fila (ex: RabbitMQ, Kafka) e processar assincronamente.

6. Consultas e Relatórios de Auditoria

Para reconstruir o estado de uma linha em um ponto no tempo:

SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY pk_value ORDER BY timestamp DESC) AS rn
    FROM audit_log
    WHERE table_name = 'orders' AND pk_value = '123' AND timestamp <= '2024-01-15 10:00:00'
) sub WHERE rn = 1;

Para identificar "quem fez o quê e quando":

SELECT user_name, operation, COUNT(*) AS total, MIN(timestamp) AS first_occurrence, MAX(timestamp) AS last_occurrence
FROM audit_log
WHERE table_name = 'orders' AND timestamp >= NOW() - INTERVAL '7 days'
GROUP BY user_name, operation
ORDER BY last_occurrence DESC;

Exemplo de consulta para deleções suspeitas:

SELECT user_name, COUNT(*) AS deletions, DATE(timestamp) AS deletion_date
FROM audit_log
WHERE operation = 'DELETE' AND table_name = 'orders'
GROUP BY user_name, DATE(timestamp)
HAVING COUNT(*) > 100
ORDER BY deletions DESC;

7. Considerações de Segurança e Manutenção

Proteção dos logs: a tabela de auditoria deve ter permissões restritas (apenas INSERT para triggers, SELECT para auditores). Considere imutabilidade usando triggers que impedem UPDATE/DELETE:

CREATE OR REPLACE FUNCTION prevent_audit_modification()
RETURNS TRIGGER AS $$
BEGIN
    RAISE EXCEPTION 'Audit logs cannot be modified';
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_prevent_modification
BEFORE UPDATE OR DELETE ON audit_log
FOR EACH ROW EXECUTE FUNCTION prevent_audit_modification();

Políticas de retenção: implemente um job agendado para mover logs antigos para cold storage ou descartar partições:

-- Excluir logs com mais de 1 ano (apenas se não houver particionamento)
DELETE FROM audit_log WHERE timestamp < NOW() - INTERVAL '1 year';

Monitoramento de anomalias: configure alertas para padrões suspeitos como múltiplas deleções em horário atípico ou acessos de IPs não autorizados. Ferramentas como pgBadger ou scripts personalizados podem analisar os logs em tempo real.

Referências