Como implementar consent management para coleta de CPF em sites e apps

from flask import Flask, request, jsonify
import json
import hashlib
import sqlite3
from datetime import datetime, timezone
from typing import Optional, List

app = Flask(__name__)

class ConsentManagementPlatform:
    """Plataforma de gerenciamento de consentimento para coleta de CPF."""

    def __init__(self, db_path: str = "consent.db"):
    self.conn = sqlite3.connect(db_path, check_same_thread=False)
    self._criar_tabelas()

    def _criar_tabelas(self):
    self.conn.executescript("""
    CREATE TABLE IF NOT EXISTS consentimentos (
    id TEXT PRIMARY KEY,
    cpf_hash TEXT NOT NULL,
    finalidade TEXT NOT NULL,
    versao_termos TEXT NOT NULL,
    texto_consentimento TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'ativo',
    canal TEXT NOT NULL,
    ip_origem TEXT,
    user_agent TEXT,
    criado_em TEXT NOT NULL,
    revogado_em TEXT
    );

    CREATE INDEX IF NOT EXISTS idx_cpf_finalidade
    ON consentimentos(cpf_hash, finalidade, status);

    CREATE TABLE IF NOT EXISTS historico_consentimentos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    consentimento_id TEXT NOT NULL,
    acao TEXT NOT NULL,
    timestamp TEXT NOT NULL,
    detalhes TEXT
    );
    """)
    self.conn.commit()

    def registrar_consentimento(
    self,
    cpf: str,
    finalidade: str,
    versao_termos: str,
    texto_consentimento: str,
    canal: str,
    ip_origem: str,
    user_agent: str
    ) -> dict:
    """Registra um novo consentimento do titular."""

    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()
    consent_id = hashlib.sha256(
    f"{cpf_hash}:{finalidade}:{datetime.now().isoformat()}".encode()
    ).hexdigest()[:24]

    agora = datetime.now(timezone.utc).isoformat()

    self.conn.execute(
    """INSERT INTO consentimentos
    (id, cpf_hash, finalidade, versao_termos, texto_consentimento,
    status, canal, ip_origem, user_agent, criado_em)
    VALUES (?, ?, ?, ?, ?, 'ativo', ?, ?, ?, ?)""",
    (consent_id, cpf_hash, finalidade, versao_termos,
    texto_consentimento, canal, ip_origem, user_agent, agora)
    )

    self.conn.execute(
    "INSERT INTO historico_consentimentos (consentimento_id, acao, timestamp) VALUES (?, ?, ?)",
    (consent_id, "consentimento_registrado", agora)
    )

    self.conn.commit()

    return {
    "consent_id": consent_id,
    "status": "ativo",
    "finalidade": finalidade,
    "registrado_em": agora
    }

    def verificar_consentimento(self, cpf: str, finalidade: str) -> bool:
    """Verifica se existe consentimento ativo para a finalidade."""

    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()
    cursor = self.conn.execute(
    "SELECT id FROM consentimentos WHERE cpf_hash = ? AND finalidade = ? AND status = 'ativo'",
    (cpf_hash, finalidade)
    )
    return cursor.fetchone() is not None

    def revogar_consentimento(self, cpf: str, finalidade: str) -> bool:
    """Revoga consentimento do titular para a finalidade especificada."""

    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()
    agora = datetime.now(timezone.utc).isoformat()

    cursor = self.conn.execute(
    "UPDATE consentimentos SET status = 'revogado', revogado_em = ? WHERE cpf_hash = ? AND finalidade = ? AND status = 'ativo'",
    (agora, cpf_hash, finalidade)
    )
    self.conn.commit()

    if cursor.rowcount > 0:
    self.conn.execute(
    "INSERT INTO historico_consentimentos (consentimento_id, acao, timestamp) VALUES (?, ?, ?)",
    (f"{cpf_hash}:{finalidade}", "consentimento_revogado", agora)
    )
    self.conn.commit()
    return True

    return False

    def listar_consentimentos(self, cpf: str) -> List[dict]:
    """Lista todos os consentimentos de um titular."""

    cpf_hash = hashlib.sha256(cpf.encode()).hexdigest()
    cursor = self.conn.execute(
    "SELECT id, finalidade, status, criado_em, revogado_em FROM consentimentos WHERE cpf_hash = ?",
    (cpf_hash,)
    )
    return [
    {
    "consent_id": row[0],
    "finalidade": row[1],
    "status": row[2],
    "criado_em": row[3],
    "revogado_em": row[4]
    }
    for row in cursor.fetchall()
    ]

# Instanciar a CMP
cmp = ConsentManagementPlatform()

@app.route("/api/consent/registrar", methods=["POST"])
def registrar():
    dados = request.json
    resultado = cmp.registrar_consentimento(
    cpf=dados["cpf"],
    finalidade=dados["finalidade"],
    versao_termos=dados["versao_termos"],
    texto_consentimento=dados["texto_consentimento"],
    canal=dados.get("canal", "web"),
    ip_origem=request.remote_addr,
    user_agent=request.headers.get("User-Agent", "")
    )
    return jsonify(resultado), 201

@app.route("/api/consent/verificar", methods=["GET"])
def verificar():
    cpf = request.args.get("cpf")
    finalidade = request.args.get("finalidade")
    ativo = cmp.verificar_consentimento(cpf, finalidade)
    return jsonify({"consentimento_ativo": ativo})

@app.route("/api/consent/revogar", methods=["POST"])
def revogar():
    dados = request.json
    revogado = cmp.revogar_consentimento(dados["cpf"], dados["finalidade"])
    return jsonify({"revogado": revogado})

import requests

def consultar_cpf_com_consentimento(
    cpf: str, finalidade: str, cmp: ConsentManagementPlatform
) -> dict:
    """Consulta CPF somente se houver consentimento ativo."""

    if not cmp.verificar_consentimento(cpf, finalidade):
    return {"error": "Consentimento não encontrado ou revogado"}

    try:
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={
    "x-api-key": "SUA_API_KEY",
    "Accept": "application/json"
    },
    timeout=30
    )

    if response.status_code == 200:
    return response.json()

    except requests.exceptions.Timeout:
    return {"error": "Timeout na consulta"}

    return {"error": "CPF não encontrado"}

curl -X GET "https://api.cpfhub.io/cpf/12345678901" \
    -H "x-api-key: SUA_API_KEY" \
    -H "Accept: application/json" \
    --max-time 30