Breach notification: como notificar a ANPD em caso de vazamento de CPFs

import json
import logging
from datetime import datetime, timezone, timedelta
from collections import defaultdict
from typing import List, Dict

logging.basicConfig(
    filename="incident_response.log",
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)

class DetectorVazamentoCPF:
    """Detecta padrões anômalos que podem indicar vazamento de dados de CPF."""

    def __init__(self):
    self.acessos_por_usuario = defaultdict(list)
    self.acessos_por_ip = defaultdict(list)
    self.alertas = []

    def registrar_acesso(self, usuario: str, ip: str, cpf_masked: str):
    """Registra um acesso e verifica anomalias."""

    agora = datetime.now(timezone.utc)
    registro = {"timestamp": agora, "cpf": cpf_masked, "ip": ip}

    self.acessos_por_usuario[usuario].append(registro)
    self.acessos_por_ip[ip].append(registro)

    # Verificar anomalias
    self._verificar_volume_excessivo(usuario, agora)
    self._verificar_ip_anomalo(usuario, ip)
    self._verificar_horario_incomum(usuario, agora)

    def _verificar_volume_excessivo(self, usuario: str, agora: datetime):
    """Detecta volume anormal de consultas."""

    uma_hora_atras = agora - timedelta(hours=1)
    acessos_recentes = [
    a for a in self.acessos_por_usuario[usuario]
    if a["timestamp"] > uma_hora_atras
    ]

    if len(acessos_recentes) > 100:
    alerta = {
    "tipo": "VOLUME_EXCESSIVO",
    "severidade": "CRITICA",
    "usuario": usuario,
    "consultas_ultima_hora": len(acessos_recentes),
    "timestamp": agora.isoformat()
    }
    self.alertas.append(alerta)
    logging.critical(json.dumps(alerta))

    def _verificar_ip_anomalo(self, usuario: str, ip: str):
    """Detecta acesso de IP não usual."""

    ips_conhecidos = set(
    a["ip"] for a in self.acessos_por_usuario[usuario][:-1]
    )

    if ips_conhecidos and ip not in ips_conhecidos:
    alerta = {
    "tipo": "IP_ANOMALO",
    "severidade": "ALTA",
    "usuario": usuario,
    "ip_novo": ip,
    "ips_conhecidos": list(ips_conhecidos)[:5],
    "timestamp": datetime.now(timezone.utc).isoformat()
    }
    self.alertas.append(alerta)
    logging.warning(json.dumps(alerta))

    def _verificar_horario_incomum(self, usuario: str, agora: datetime):
    """Detecta acesso fora do horário comercial."""

    hora = agora.hour
    if hora < 6 or hora > 23:
    alerta = {
    "tipo": "HORARIO_INCOMUM",
    "severidade": "MEDIA",
    "usuario": usuario,
    "hora": hora,
    "timestamp": agora.isoformat()
    }
    self.alertas.append(alerta)
    logging.warning(json.dumps(alerta))

    def obter_alertas_criticos(self) -> List[Dict]:
    """Retorna alertas que podem indicar vazamento."""
    return [a for a in self.alertas if a["severidade"] == "CRITICA"]

from dataclasses import dataclass, field
from typing import List

@dataclass
class NotificacaoANPD:
    """Gera a notificação formal para a ANPD conforme exigências da LGPD."""

    controlador: str
    cnpj: str
    dpo_nome: str
    dpo_email: str
    data_incidente: str
    data_deteccao: str
    dados_afetados: List[str]
    titulares_afetados: int
    descricao_incidente: str
    medidas_seguranca: List[str]
    medidas_mitigacao: List[str]
    riscos: List[str]

    def gerar_documento(self) -> dict:
    """Gera o documento de notificação estruturado."""

    return {
    "comunicacao_incidente_seguranca": {
    "controlador": {
    "razao_social": self.controlador,
    "cnpj": self.cnpj,
    "encarregado": {
    "nome": self.dpo_nome,
    "email": self.dpo_email
    }
    },
    "incidente": {
    "data_ocorrencia": self.data_incidente,
    "data_conhecimento": self.data_deteccao,
    "descricao": self.descricao_incidente,
    "dados_pessoais_afetados": self.dados_afetados,
    "numero_titulares_afetados": self.titulares_afetados,
    "categoria_titulares": "clientes"
    },
    "medidas": {
    "seguranca_preexistentes": self.medidas_seguranca,
    "mitigacao_adotadas": self.medidas_mitigacao
    },
    "riscos": self.riscos,
    "comunicacao_titulares": {
    "realizada": True,
    "meio": "e-mail individual",
    "data": self.data_deteccao
    }
    }
    }

# Exemplo de uso
notificacao = NotificacaoANPD(
    controlador="Empresa XYZ Ltda",
    cnpj="12.345.678/0001-00",
    dpo_nome="Carlos Souza",
    dpo_email="dpo@empresa.com.br",
    data_incidente="2025-08-08",
    data_deteccao="2025-08-09",
    dados_afetados=["CPF", "Nome completo", "Data de nascimento"],
    titulares_afetados=1500,
    descricao_incidente="Acesso não autorizado ao banco de dados de clientes via credencial comprometida.",
    medidas_seguranca=[
    "Criptografia AES-256 em repouso",
    "Controle de acesso RBAC",
    "Logs de auditoria ativos"
    ],
    medidas_mitigacao=[
    "Credencial comprometida revogada imediatamente",
    "Forçado reset de senha para todos os administradores",
    "Bloqueio do IP de origem do ataque",
    "Notificação aos titulares afetados via e-mail"
    ],
    riscos=[
    "Possibilidade de fraude identitária com dados de CPF expostos",
    "Risco de engenharia social usando nome e data de nascimento"
    ]
)

documento = notificacao.gerar_documento()
print(json.dumps(documento, indent=2, ensure_ascii=False))

# Verificar se um CPF específico foi afetado
curl -X GET "https://api.cpfhub.io/cpf/12345678901" \
    -H "x-api-key: SUA_API_KEY" \
    -H "Accept: application/json" \
    --max-time 30