Día 3 -- Audit trail inmutable de accesos a datos

1Rosa Martinez  08:47 AM  #data-privacy
2@equipo-datos @laura.sanchez
3
4Ayer tuve reunion con la abogada externa (Marta Serrano de Garrigues).
5Nos ha informado de una nueva directriz de la AEPD para plataformas
6de salud digital: a partir de abril, debemos poder demostrar QUIEN
7accedio a CUALQUIER dato de paciente, CUANDO y POR QUE, con un SLA
8de respuesta inferior a 72 horas.
9
10Esto significa audit trail COMPLETO e INMUTABLE.
11
12Requisitos:
131. Registrar cada SELECT/INSERT/UPDATE/DELETE sobre tablas de pacientes
142. Campos mínimos: usuario, timestamp, tabla, operación, query, IP,
15   número de registros afectados
163. La tabla de audit debe ser APPEND-ONLY (nadie puede borrar ni
17   modificar registros de auditoría)
184. Busqueda indexada por paciente_id y por usuario
195. Retencion mínima: 24 meses
20
21Prioridad: MAXIMA. Necesitamos esto operativo antes de abril.
22
23Laura, esto es para @tu y Javier. Que empiecen hoy.

Rosa no pide -- informa de obligaciones legales con deadlines

1-- Tabla de audit trail: APPEND-ONLY, inmutable
2-- Nadie puede hacer DELETE ni UPDATE sobre está tabla
3
4CREATE TABLE audit_accesos_pacientes (
5    id BIGSERIAL PRIMARY KEY,
6    timestamp_utc TIMESTAMP NOT NULL DEFAULT NOW(),
7    
8    -- QUIEN accedio
9    usuario_id VARCHAR(50) NOT NULL,
10    usuario_nombre VARCHAR(100),
11    usuario_rol VARCHAR(30) NOT NULL,      -- medico/enfermero/admin/analista/investigador
12    ip_origen INET,
13    sesion_id VARCHAR(64),
14    
15    -- QUE hizo
16    operación VARCHAR(10) NOT NULL,         -- SELECT/INSERT/UPDATE/DELETE
17    tabla_afectada VARCHAR(50) NOT NULL,
18    query_ejecutada TEXT,                   -- La query completa (censurada si contiene PII)
19    
20    -- SOBRE QUIEN
21    paciente_ids TEXT[],                    -- Array de pacientes afectados
22    num_registros INTEGER DEFAULT 0,        -- Cuantos registros se leyeron/modificaron
23    
24    -- CONTEXTO
25    finalidad VARCHAR(50),                  -- asistencial/investigacion/operaciones/admin
26    justificacion TEXT,                     -- Motivo libre (obligatorio para investigacion)
27    
28    -- METADATA
29    aplicacion VARCHAR(30),                 -- webapp/api/etl/adhoc
30    duracion_ms INTEGER,                    -- Tiempo de ejecucion de la query
31    
32    -- Inmutabilidad: hash de la fila anterior para detectar manipulacion
33    hash_anterior VARCHAR(64)
34);
35
36-- Indices para busqueda rápida (SLA < 72h pero queremos < 5 segundos)
37CREATE INDEX idx_audit_paciente ON audit_accesos_pacientes USING GIN (paciente_ids);
38CREATE INDEX idx_audit_usuario ON audit_accesos_pacientes (usuario_id, timestamp_utc DESC);
39CREATE INDEX idx_audit_timestamp ON audit_accesos_pacientes (timestamp_utc DESC);
40CREATE INDEX idx_audit_tabla ON audit_accesos_pacientes (tabla_afectada, operación);
41
42-- Particionado por mes para rendimiento (24 meses de retención)
43-- En producción usariamos particionado nativo de PostgreSQL 15
44-- CREATE TABLE audit_accesos_pacientes (...) PARTITION BY RANGE (timestamp_utc);
45
46-- INMUTABILIDAD: revocar DELETE y UPDATE a TODOS los usuarios
47REVOKE DELETE, UPDATE ON audit_accesos_pacientes FROM PUBLIC;
48REVOKE DELETE, UPDATE ON audit_accesos_pacientes FROM app_user;
49REVOKE DELETE, UPDATE ON audit_accesos_pacientes FROM data_engineer;
50-- Solo el superuser de emergencia puede tocar está tabla (y queda logueado en pg_log)

Tabla append-only con índices para busqueda en < 5 segundos

1-- Funcion de auditoría generica para triggers
2CREATE OR REPLACE FUNCTION fn_audit_pacientes()
3RETURNS TRIGGER AS $$
4DECLARE
5    paciente_id_valor TEXT;
6    usuario_actual TEXT;
7    ip_actual INET;
8BEGIN
9    -- Obtener usuario y contexto de sesion
10    usuario_actual := COALESCE(
11        current_setting('app.current_user_id', true),
12        current_user
13    );
14    ip_actual := COALESCE(
15        current_setting('app.client_ip', true)::INET,
16        inet_client_addr()
17    );
18
19    -- Extraer paciente_id segun la operación
20    IF TG_OP = 'DELETE' THEN
21        paciente_id_valor := OLD.paciente_id;
22    ELSE
23        paciente_id_valor := NEW.paciente_id;
24    END IF;
25
26    -- Si la tabla es 'pacientes', el id ES el paciente_id
27    IF TG_TABLE_NAME = 'pacientes' THEN
28        IF TG_OP = 'DELETE' THEN
29            paciente_id_valor := OLD.id;
30        ELSE
31            paciente_id_valor := NEW.id;
32        END IF;
33    END IF;
34
35    -- Insertar registro de auditoría
36    INSERT INTO audit_accesos_pacientes (
37        usuario_id, usuario_rol, ip_origen,
38        operación, tabla_afectada, 
39        paciente_ids, num_registros,
40        finalidad, aplicacion
41    ) VALUES (
42        usuario_actual,
43        COALESCE(current_setting('app.current_user_role', true), 'sistema'),
44        ip_actual,
45        TG_OP,
46        TG_TABLE_NAME,
47        ARRAY[paciente_id_valor],
48        1,
49        COALESCE(current_setting('app.finalidad', true), 'operacional'),
50        COALESCE(current_setting('app.nombre_aplicacion', true), 'trigger_auto')
51    );
52
53    -- Los triggers AFTER no modifican la fila
54    IF TG_OP = 'DELETE' THEN
55        RETURN OLD;
56    END IF;
57    RETURN NEW;
58END;
59$$ LANGUAGE plpgsql SECURITY DEFINER;
60
61-- Instalar triggers en TODAS las tablas con datos de pacientes
62CREATE TRIGGER trg_audit_pacientes
63    AFTER INSERT OR UPDATE OR DELETE ON pacientes
64    FOR EACH ROW EXECUTE FUNCTION fn_audit_pacientes();
65
66CREATE TRIGGER trg_audit_historiales
67    AFTER INSERT OR UPDATE OR DELETE ON historiales
68    FOR EACH ROW EXECUTE FUNCTION fn_audit_pacientes();
69
70CREATE TRIGGER trg_audit_prescripciones
71    AFTER INSERT OR UPDATE OR DELETE ON prescripciones
72    FOR EACH ROW EXECUTE FUNCTION fn_audit_pacientes();
73
74CREATE TRIGGER trg_audit_consultas
75    AFTER INSERT OR UPDATE OR DELETE ON consultas
76    FOR EACH ROW EXECUTE FUNCTION fn_audit_pacientes();
77
78CREATE TRIGGER trg_audit_wearables
79    AFTER INSERT OR UPDATE OR DELETE ON wearables
80    FOR EACH ROW EXECUTE FUNCTION fn_audit_pacientes();

Un trigger por tabla -- cualquier DML queda registrado automáticamente

1# middleware_auditoria.py -- Registra SELECTs sobre tablas de pacientes
2# Este middleware se interpone entre la app y PostgreSQL
3
4import re
5import time
6import logging
7from datetime import datetime
8from typing import Optional, List
9from contextlib import contextmanager
10
11logger = logging.getLogger('audit_middleware')
12
13# Tablas que requieren auditoría en SELECT
14TABLAS_AUDITADAS = {
15    'pacientes', 'historiales', 'prescripciones',
16    'consultas', 'wearables', 'consentimientos'
17}
18
19# Patron para detectar tablas en una query SELECT
20PATRON_TABLAS = re.compile(
21    r'\bFROM\s+([\w.]+)|\bJOIN\s+([\w.]+)',
22    re.IGNORECASE
23)
24
25
26class AuditMiddleware:
27    """Intercepta queries y registra accesos a tablas sensibles."""
28
29    def __init__(self, db_connection, usuario_id: str, usuario_rol: str, ip: str):
30        self.db = db_connection
31        self.usuario_id = usuario_id
32        self.usuario_rol = usuario_rol
33        self.ip = ip
34
35    @contextmanager
36    def query_auditada(self, finalidad: str, justificacion: str = ''):
37        """Context manager que audita la query ejecutada."""
38        # Setear variables de sesion para los triggers DML
39        cursor = self.db.cursor()
40        cursor.execute(f"SET LOCAL app.current_user_id = '{self.usuario_id}'")
41        cursor.execute(f"SET LOCAL app.current_user_role = '{self.usuario_rol}'")
42        cursor.execute(f"SET LOCAL app.client_ip = '{self.ip}'")
43        cursor.execute(f"SET LOCAL app.finalidad = '{finalidad}'")
44
45        yield AuditedCursor(
46            cursor, self.db, self.usuario_id, self.usuario_rol,
47            self.ip, finalidad, justificacion
48        )
49
50        cursor.close()
51
52
53class AuditedCursor:
54    """Cursor wrapper que registra automaticamente SELECTs auditables."""
55
56    def __init__(self, cursor, db, usuario_id, usuario_rol, ip, finalidad, justificacion):
57        self._cursor = cursor
58        self._db = db
59        self._usuario_id = usuario_id
60        self._usuario_rol = usuario_rol
61        self._ip = ip
62        self._finalidad = finalidad
63        self._justificacion = justificacion
64
65    def execute(self, query: str, params=None):
66        """Ejecuta query y registra si toca tablas auditadas."""
67        inicio = time.time()
68
69        # Ejecutar la query real
70        self._cursor.execute(query, params)
71        duracion_ms = int((time.time() - inicio) * 1000)
72
73        # Determinar si es SELECT sobre tabla auditada
74        if query.strip().upper().startswith('SELECT'):
75            tablas_tocadas = self._extraer_tablas(query)
76            tablas_sensibles = tablas_tocadas & TABLAS_AUDITADAS
77
78            if tablas_sensibles:
79                num_registros = self._cursor.rowcount
80                self._registrar_acceso(
81                    query, tablas_sensibles, num_registros, duracion_ms
82                )
83
84    def _extraer_tablas(self, query: str) -> set:
85        """Extrae nombres de tablas de una query SQL."""
86        matches = PATRON_TABLAS.findall(query)
87        tablas = set()
88        for match in matches:
89            tabla = match[0] or match[1]
90            # Quitar schema si existe (public.pacientes -> pacientes)
91            tablas.add(tabla.split('.')[-1])
92        return tablas
93
94    def _registrar_acceso(
95        self, query: str, tablas: set, num_registros: int, duracion_ms: int
96    ):
97        """Inserta registro de auditoría para el SELECT."""
98        # Censurar la query si contiene datos PII en WHERE
99        query_censurada = self._censurar_query(query)
100
101        for tabla in tablas:
102            self._db.execute(
103                """INSERT INTO audit_accesos_pacientes (
104                    usuario_id, usuario_rol, ip_origen,
105                    operación, tabla_afectada, query_ejecutada,
106                    num_registros, finalidad, justificacion,
107                    aplicacion, duracion_ms
108                ) VALUES (%s, %s, %s, 'SELECT', %s, %s, %s, %s, %s, 'middleware', %s)""",
109                (
110                    self._usuario_id, self._usuario_rol, self._ip,
111                    tabla, query_censurada, num_registros,
112                    self._finalidad, self._justificacion, duracion_ms
113                )
114            )
115
116    def _censurar_query(self, query: str) -> str:
117        """Reemplaza valores PII en la query por [REDACTED]."""
118        # Censurar DNIs (8 digitos + letra)
119        query = re.sub(r"'\d{8}[A-Z]'", "'[REDACTED_DNI]'", query)
120        # Censurar emails
121        query = re.sub(r"'[\w.]+@[\w.]+'", "'[REDACTED_EMAIL]'", query)
122        # Censurar nombres entre comillas
123        query = re.sub(r"nombre\s*=\s*'[^']+'", "nombre='[REDACTED]'", query)
124        return query
125
126    def fetchall(self):
127        return self._cursor.fetchall()
128
129    def fetchone(self):
130        return self._cursor.fetchone()

Middleware que audita SELECTs sobre tablas sensibles automáticamente