logus-lgpd 1.0.1

Privacy-by-Design para dados tabulares — LGPD compliance em Python.

logus

Privacy-by-Design para dados tabulares em Python.

Detecta, mascara e protege dados pessoais automaticamente — com suporte completo a LGPD, formato .lgs para transferência segura de bases de dados, e integração com bancos de dados relacionais.

import logus as lg

# Detecta PII em qualquer DataFrame
reports = lg.scan(df)

# Mascara automaticamente — CPF vira HMAC, nome vira REDACTED
df_safe = lg.mask(df, salt="chave-hmac")

# Salva cifrado com AES-256-GCM
lg.store(df_safe, "clientes.lgs", key="chave-aes")

# Lê com contexto (Pythônico)
with lg.open("clientes.lgs", key="chave-aes") as f:
    df = f.read()
    info = f.info()

---

Instalação

pip install logus-lgpd

# Com Polars (engine de alta performance)
pip install "logus-lgpd[polars]"

# Com suporte a dados sintéticos
pip install "logus-lgpd[synthetic]"

# Tudo
pip install "logus-lgpd[full]"

Dependências obrigatórias: pandas, pyarrow, cryptography, numpy

Opcionais: polars (performance), sqlalchemy (banco de dados), ctgan (dados sintéticos)

---

Índice

1. Detecção de PII
2. Mascaramento
3. Formato .lgs — arquivo seguro
4. LGSFile — context manager
5. Multi-frame
6. Análise e diagnóstico
7. Integração com bancos de dados
8. Streaming para big data
9. Engine Polars
10. Verificação e auditoria
11. Privacidade diferencial
12. Dados sintéticos
13. CLI
14. Segurança — posição e limites
15. Referência da API
16. Changelog

---

1. Detecção de PII

lg.scan() — detecta e classifica colunas

import logus as lg
import pandas as pd

df = pd.read_csv("clientes.csv")
reports = lg.scan(df)

# Também aceita caminhos de arquivo
reports = lg.scan("clientes.csv")
reports = lg.scan("clientes.lgs", key="chave-aes")
reports = lg.scan("clientes.parquet")

# Resultado por coluna
for col, r in reports.items():
    print(f"{col}: {r.pii_type.value} | risco={r.risk_level.value} | estratégia={r.mask_strategy.value}")
# cpf: cpf | risco=high | estratégia=hash
# email: email | risco=high | estratégia=hash
# nome: nome | risco=high | estratégia=redact
# cep: cep | risco=medium | estratégia=truncate

Tipos detectados automaticamente:

Tipo	Estratégia padrão	Exemplo
CPF	HMAC-SHA256	111.444.777-35 → 7c0a942e8c7919b6
CNPJ	HMAC-SHA256	12.345.678/0001-95 → 4a2f...
E-mail	HMAC-SHA256	[e-mail] → 9d1e...
Nome	REDACTED	Ana Silva → REDACTED
Telefone	Mantém DDD	(11) 99999-1234 → (11) XXXXX-XXXX
CEP	Trunca bairro	01310-100 → 01310-XXX
Data nasc.	Generaliza faixa	1985-03-15 → 1980-1989
RG	HMAC-SHA256	12.345.678-9 → bf3a...
IP	HMAC-SHA256	192.168.1.1 → c82e...
Cartão	REDACTED	4111 1111 1111 1111 → REDACTED
Quasi-ID	Mock categórico	SP → amostra da distribuição
Numérico	Mock numérico	5000.00 → perturbação ±5%

lg.profile() — diagnóstico completo

report = lg.profile(df)
# Também aceita arquivo:
report = lg.profile("clientes.lgs", key="chave-aes")

print(report["pii_columns"])       # ["cpf", "email", "nome"]
print(report["pii_risk_summary"])  # "3🔴 1🟡 0🟢"
print(report["null_pct"])          # 0.42  (% de nulos)
print(report["shape"])             # (50000, 12)

# JSON-serializable — pode ser logado, enviado para SIEM, etc.
import json
json.dumps(report)  # funciona sem erro

Detecção de dados sensíveis (Art. 11 LGPD)

reports, sensitive = lg.scan(df, sensitive=True)
# sensitive: achados de saúde, biometria, origem étnica, etc.

---

2. Mascaramento

lg.mask() — aplica mascaramento automático

# Mascara tudo detectado
df_safe = lg.mask(df, salt="chave-hmac-min-16-chars")

# Só colunas específicas
df_safe = lg.mask(df, salt="chave", columns=["cpf", "email"])

# Tudo exceto algumas colunas
df_safe = lg.mask(df, salt="chave", exclude=["nome"])

# Com relatório de colunas detectadas
df_safe = lg.mask(df, salt="chave", verbose=True)

Por que salt é obrigatório em produção?

O salt HMAC garante que o mesmo CPF em duas tabelas diferentes produza o mesmo token — preservando a integridade referencial para JOIN. Sem salt, o token é aleatório e os joins são impossíveis.

# Gera salt seguro (256 bits de entropia)
salt = lg.generate_salt()  # ex: "a3f8c2e1..."

Normalização automática: 111.444.777-35, 11144477735 e 111-444-777.35 geram o mesmo token. Isso é crítico para joins entre sistemas com formatações diferentes.

lg.join() — join seguro entre tabelas mascaradas

# Dados brutos — aplica mesmo mascaramento em ambos antes do join
result = lg.join(df_clientes, df_pedidos, on="cpf", salt="chave")

# Dados já mascarados — valida compatibilidade antes de fazer o merge
df_c = lg.mask(df_clientes, salt="chave")
df_p = lg.mask(df_pedidos,  salt="chave")
result = lg.join(df_c, df_p, on="cpf")

# Detecta automaticamente se salts diferentes foram usados (raises ValueError)
# join_result = lg.join(df_c_salt1, df_p_salt2, on="cpf")  # ← ValueError clara

lg.diff() — compara antes/depois

df_safe = lg.mask(df, salt="chave")
report = lg.diff(df, df_safe)

print(report["summary"])
# 3 coluna(s) mascarada(s):
#   hash: cpf, email
#   redact: nome
# 4 coluna(s) inalterada(s): uf, salario, data_admissao, produto

print(report["per_column"]["cpf"]["examples"])
# [{"before": "111.444.777-35", "after": "7c0a942e8c7919b6"}]

---

3. Formato .lgs — arquivo seguro

O .lgs é um contêiner binário para transferência e armazenamento seguro de DataFrames. Resolve um problema específico: como enviar uma base de dados entre ambientes com rastreabilidade, integridade verificável e criptografia forte, sem depender de infra de chave pública (GPG) nem de plataformas cloud.

Recurso	.lgs	GPG/age	7-Zip AES	Parquet+SSE
AES-256-GCM	✅	✅	✅	✅
HMAC-SHA256 (integridade)	✅	✅	❌	❌
Header com metadados LGPD	✅	❌	❌	❌
Multi-frame (várias tabelas)	✅	❌	❌	❌
Schema tabular nativo	✅	❌	❌	✅
Sem key (só integridade)	✅	❌	❌	❌
Metadata customizado	✅	❌	❌	❌

Estrutura binária

[5  bytes]  MAGIC = b"LOGUS"
[1  byte ]  VERSION (0x02=cifrado, 0x03=multi-frame, 0x04=aberto)
[1  byte ]  CIPHER (0x01=AES-256-GCM, 0x02=ChaCha20-Poly1305)
[32 bytes]  SALT_KDF    — salt HKDF único por arquivo
[12 bytes]  NONCE_HEADER
[4  bytes]  HEADER_CT_LEN
[N  bytes]  HEADER JSON cifrado (metadados LGPD: shape, schema, created_by...)
[12 bytes]  NONCE_PAYLOAD
[M  bytes]  PAYLOAD Parquet/zstd cifrado
[32 bytes]  FILE_HMAC — HMAC-SHA256 sobre tudo acima

Escrita e leitura

# --- Salvar ---
lg.store(df, "clientes.lgs", key="chave-aes")
lg.save(df, "clientes.lgs", key="chave-aes")  # alias

# Com metadados customizados
lg.store(df, "clientes.lgs", key="chave-aes", metadata={
    "origem":         "crm_v2",
    "squad":          "dados",
    "versao_schema":  "3",
    "data_referencia": "2024-01",
})

# Sem criptografia (dados já anonimizados)
lg.store(df_anonimo, "dados_dev.lgs")
lg.store(df_bruto,   "dados_dev.lgs", anonymize=True)  # mascara antes de gravar

# --- Ler ---
df = lg.read("clientes.lgs", key="chave-aes")
df = lg.load("clientes.lgs", key="chave-aes")  # alias
df = lg.read("clientes.lgs", key="chave-aes", raw=True)  # sem mascaramento adicional

# Arquivo aberto (sem criptografia)
df = lg.read("dados_dev.lgs")  # detecta v4 automaticamente

# --- Inspecionar sem decifrar ---
info = lg.inspect("clientes.lgs", key="chave-aes")
print(info["shape"])       # (50000, 12)
print(info["created_at"])  # "2024-01-15T10:30:00+00:00"
print(info["metadata"])    # {"origem": "crm_v2", ...}
print(info["encryption"])  # "AES256GCM"

# --- Verificar integridade ---
result = lg.inspect("clientes.lgs", key="chave-aes")  # raises ValueError se corrompido

# Arquivo aberto
lg.inspect("dados_dev.lgs")  # funciona sem key

Rotação de chave

lg.rekey(
    "clientes.lgs",
    old_key="chave-antiga-compromissada",
    new_key="nova-chave-segura",
)
# Operação atômica — write-then-rename, nunca deixa arquivo em estado inconsistente

---

4. LGSFile — context manager

A interface orientada a objeto para trabalhar com arquivos .lgs:

import logus as lg

# Context manager (pattern Pythônico)
with lg.open("clientes.lgs", key="chave") as f:
    df   = f.read()                      # decifra e retorna DataFrame
    info = f.info()                      # metadados sem decifrar payload
    s    = f.shape()                     # (50000, 12) sem decifrar
    f.write(df_novo)                     # sobrescreve
    f.add_frame("pedidos", df_pedidos)   # converte para multi-frame
    f2 = f.copy_to("backup.lgs")        # copia sem decifrar

# Fluent API
df = lg.open("clientes.lgs", key="chave").read()

# Boolean check
if not lg.open("arquivo.lgs", key="chave"):
    raise RuntimeError("Arquivo corrompido ou não existe!")

# Verificação simples
f = lg.open("arquivo.lgs", key="chave")
f.valid()       # True/False
f.exists()      # True/False
f.size_kb()     # 142.3
f.delete()      # remove o arquivo

LGSInfo — resultado rico de verify()

from logus.secure_file import SecureFile

# Bool direto — sem tuple unpacking obrigatório
info = SecureFile.verify("clientes.lgs", key="chave")

if not info:
    raise RuntimeError("Corrompido!")

# Acesso por atributo
print(info.content_type)   # "raw_dataframe"
print(info.shape)          # [50000, 12]
print(info.encryption)     # "AES256GCM"
print(info.created_at)     # "2024-01-15T10:30:00+00:00"

# Retrocompatível: tuple unpacking ainda funciona
ok, data = SecureFile.verify("clientes.lgs", key="chave")
assert ok is True
assert data["content_type"] == "raw_dataframe"

# Acesso por chave (dict-like)
print(info["shape"])
print(info.get("metadata", {}))
info.to_dict()  # serializa para dict puro

---

5. Multi-frame

Um único arquivo .lgs pode conter múltiplas tabelas — útil para transferir uma base de dados completa entre ambientes:

# Escrita
lg.store({
    "clientes":  df_clientes,
    "pedidos":   df_pedidos,
    "produtos":  df_produtos,
    "estoque":   df_estoque,
}, "base_producao.lgs", key="chave")

# Leitura — todas as tabelas
frames = lg.read("base_producao.lgs", key="chave")
df_c = frames["clientes"]
df_p = frames["pedidos"]

# Leitura — uma tabela (sem carregar as outras)
df_c = lg.read("base_producao.lgs", key="chave", frame="clientes")

# Via LGSFile
with lg.open("base_producao.lgs", key="chave") as f:
    print(f.frame_names())     # ["clientes", "pedidos", "produtos", "estoque"]
    df = f.frame("clientes")   # carrega só esta tabela

# Adiciona frame a arquivo existente
with lg.open("base_producao.lgs", key="chave") as f:
    f.add_frame("logs", df_logs)

# Inspeciona sem decifrar
info = lg.inspect("base_producao.lgs", key="chave")
print(info["n_frames"])     # 4
print(info["frame_names"])  # ["clientes", "pedidos", ...]

Implementação interna: o payload é um ZIP em memória de Parquets com índice JSON. O byte de versão 0x03 identifica o formato. Retrocompatível — leitores v1/v2 recebem TypeError descritivo.

---

6. Análise e diagnóstico

Funções analíticas — API pandas, engine Polars

Todas as funções aceitam pd.DataFrame e pl.DataFrame:

import logus as lg

lg.describe(df)                                    # df.describe() — 2–4x mais rápido
lg.value_counts(df, "uf", normalize=True, n=10)   # df["uf"].value_counts()
lg.null_counts(df)                                 # df.isnull().sum() — 5–10x mais rápido
lg.nunique(df)                                     # df.nunique()
lg.corr(df)                                        # df.corr() — 2–5x mais rápido
lg.head(df, 10)                                    # df.head(10)
lg.tail(df, 5)                                     # df.tail(5)
lg.shape(df)                                       # df.shape
lg.dtypes(df)                                      # {col: tipo_str}

# Filtro — sem shadow do builtin filter
lg.where(df, {"uf": "SP"})                         # dict → igualdade
lg.where(df, {"salario": (5000, 9000)})            # dict → range
lg.where(df, 'uf == "SP" and salario > 5000')      # query string
lg.where(df, lambda d: d["uf"] == "SP")            # callable
lg.where(df, pl.col("uf") == "SP")                 # pl.Expr nativo
lg.query(df, {"uf": "SP"})                         # alias de where()

# Transformações
lg.sort(df, ["uf", "salario"], ascending=[True, False])
lg.groupby(df, "uf", {"salario": ["sum", "mean"], "v": "count"})
lg.select(df, ["cpf", "uf"])
lg.drop(df, "coluna_inutil")
lg.rename(df, {"cpf": "documento"})
lg.cast(df, {"idade": "int", "preco": "float32"})
lg.fillna(df, {"salario": 0, "uf": "NA"})
lg.sample(df, n=1000, random_state=42)
lg.unique(df, "uf")                                # lista de valores únicos

---

7. Integração com bancos de dados

Leitura com mascaramento automático

from logus import link

# Conecta (aceita URL string ou SQLAlchemy Engine)
adapter = link.db("postgresql://user:pass@host:5432/db", salt="chave-hmac")
# MySQL
adapter = link.db("mysql+pymysql://user:pass@host/db", salt="chave")
# SQLite
adapter = link.db("sqlite:///dados.db", salt="chave")
# SQL Server
adapter = link.db(
    "mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+17+for+SQL+Server",
    salt="chave",
)

# Pull & Mask — lê localmente e mascara
df = adapter.query("SELECT * FROM clientes WHERE uf = %s", params=("SP",))
df = adapter.query_table("clientes", where="ativo = true", limit=10_000)
df = adapter.query_table("clientes", mask_columns=["cpf", "email"])  # só estas
df = adapter.query_chunked("SELECT * FROM eventos", chunksize=50_000)

Escrita no banco

# Escreve DataFrame mascarado de volta ao banco
df_safe = lg.mask(df, salt="chave")
n = adapter.write(df_safe, "clientes_masked", if_exists="replace")

# Pipeline completo: lê prod, mascara, grava em dev
adapter.read_and_write_masked("clientes_prod", "clientes_dev")

In-DB masking — dados nunca saem do banco

O modo mais seguro: só uma amostra de 500 linhas sai para detecção PII, o mascaramento é executado via SQL no banco.

# Revisa SQLs antes de executar
result = adapter.in_db_mask("clientes", dry_run=True)
for sql in result["sql_statements"]:
    print(sql)
# UPDATE "clientes" SET "cpf" = encode(hmac("cpf"::text::bytea, 'chave'::bytea, 'sha256'), 'hex');
# UPDATE "clientes" SET "nome" = 'REDACTED';
# UPDATE "clientes" SET "cep" = substring(regexp_replace("cep", '[^0-9]', '', 'g'), 1, 5) || '-XXX';

# Executa (modifica dados IRREVERSIVELMENTE — faça backup antes)
result = adapter.in_db_mask("clientes")
print(result["columns_masked"])  # ["cpf", "email", "nome", "cep"]

# Cria VIEW mascarada — dado original intacto
result = adapter.create_masked_view("clientes")
# Agora: SELECT * FROM clientes_masked
print(result["sql"])  # CREATE OR REPLACE VIEW clientes_masked AS SELECT ...

# Listar tabelas
adapter.tables()       # ["clientes", "pedidos", "produtos"]
adapter.columns("clientes")  # [{name, type, nullable}, ...]

Suporte de SQL por banco:

Banco	Hash	REDACT	CEP	Data	Numeric
PostgreSQL	HMAC via pgcrypto ✅	✅	✅	✅	✅
MySQL/MariaDB	SHA2(CONCAT) ⚠️	✅	✅	✅	✅
SQL Server	HASHBYTES ⚠️	✅	✅	✅	✅
SQLite	randomblob ⚠️	✅	✅	—	✅
BigQuery	SHA256 ⚠️	✅	✅	—	✅

⚠️ Sem HMAC nativo: SHA256 sem chave secreta — pseudonimização mais fraca. Use pull-and-mask para dados críticos nesses bancos.

Geração de script SQL (sem conexão)

reports = lg.scan(df)
script = link.sql(df, reports, table="clientes", dialect="postgresql")
print(script)
# CREATE OR REPLACE VIEW clientes_masked AS SELECT
#   encode(hmac("cpf"::text::bytea, '${LOGUS_SALT}'::bytea, 'sha256'), 'hex') AS "cpf",
#   'REDACTED' AS "nome",
#   ...
# FROM "clientes";

Atalho de alto nível

# Lê direto sem criar adapter
df = lg.read_db(
    "postgresql://user:pass@host/db",
    "SELECT * FROM clientes WHERE uf = %s",
    salt="chave",
    params=("SP",),
)

# Lê tabela inteira
df = lg.read_db(
    "postgresql://user:pass@host/db",
    "clientes",
    salt="chave",
    table=True,
    limit=100_000,
)

---

8. Streaming para big data

Para arquivos maiores que a memória disponível:

# Streaming básico — CSV e Parquet
for df_chunk in lg.stream("grande.csv", salt="chave", chunksize=50_000):
    processar(df_chunk)

# Com progresso
from tqdm import tqdm
with tqdm(unit=" linhas", total=5_000_000) as bar:
    for chunk in lg.stream(
        "grande.csv",
        salt="chave",
        chunksize=50_000,
        on_progress=lambda n, done, total: bar.update(len(chunk)),
    ):
        processar(chunk)

# Com callback customizado
def progresso(chunk_n, feitas, total_estimado):
    print(f"Chunk {chunk_n}: {feitas:,}/{total_estimado:,}")

for chunk in lg.stream("grande.parquet", salt="chave", on_progress=progresso):
    processar(chunk)

# Streaming de banco de dados
adapter = link.db("postgresql://...", salt="chave")
df = adapter.query_chunked("SELECT * FROM eventos", chunksize=100_000)

# Pipeline Polars — grava chunks mascarados em Parquet (sem acumular em memória)
from logus.adapters.polars_adapter import load_secure_dataframe_chunked
load_secure_dataframe_chunked(
    "grande.csv",
    salt="chave",
    chunksize=100_000,
    output_path="grande_mascarado.parquet",
)

Performance: 500k linhas × 7 colunas em ~3.5s; 100k linhas em ~0.73s.

---

9. Engine Polars

Quando polars está instalado, o engine de alta performance é ativado automaticamente:

# pd.DataFrame → detecta Polars, usa engine nativo
df_safe = lg.mask(df_pandas, salt="chave")

# pl.DataFrame — namespace dedicado
import polars as pl
df_pl = pl.read_csv("clientes.csv")

df_pl_safe = lg.pl.mask(df_pl, salt="chave")           # → pl.DataFrame
df_pl_safe = lg.pl.read("clientes.csv", salt="chave")   # → pl.DataFrame
df_pl_safe = lg.pl.read("clientes.lgs", key="k")        # → pl.DataFrame

# LazyFrame — arquivo não é carregado em memória
lf = lg.pl.lazy("grande.parquet", salt="chave")         # → pl.LazyFrame
result = lf.filter(pl.col("uf") == "SP").collect()

# lg.pl espelha toda a API de alto nível
lg.pl.scan(df_pl)
lg.pl.store(df_pl, "f.lgs", key="k")

Ganhos mensurados:

Operação	Pandas	Polars	Ganho
Regex scan (CPF 100k)	~34ms	~8ms	4×
CPF normalize	~160ms	—	str.replace vetorizado: 4×
DateMasker string	~190ms	LUT numpy: ~80ms	2.3×
null_counts()	bitmask pandas	bitmask Arrow	5–10×
groupby()	pandas hash	Polars hash-join	5–20×
value_counts()	sort+count	Arrow SIMD	3–8×
corr()	produto matricial	Polars nativo	2–5×

---

10. Verificação e auditoria

Verificação de integridade

from logus.secure_file import SecureFile

# Bool direto
if not SecureFile.verify("clientes.lgs", key="chave"):
    raise RuntimeError("Arquivo adulterado!")

# Metadados detalhados
info = SecureFile.verify("clientes.lgs", key="chave")
print(info.shape)       # [50000, 12]
print(info.created_at)  # "2024-01-15T10:30:00+00:00"
print(info.encryption)  # "AES256GCM"
print(info.metadata)    # {"origem": "crm_v2"}

# Tuple unpacking retrocompatível
ok, data = SecureFile.verify("clientes.lgs", key="chave")

# Arquivo sem criptografia
SecureFile.verify("dados_dev.lgs")  # sem key

Auditoria automática (LGPD Art. 50)

from logus.reports.audit_report import AuditReport

# Ativa auditoria global — toda chamada a lg.mask() registra automaticamente
audit = AuditReport()
lg.configure(audit=audit)

# Todas as operações são registradas a partir daqui
df_safe = lg.mask(df, salt="chave")

# Exporta trilha de auditoria
audit.save("audit/lgpd_2024_01.json")
audit.print()
# [2024-01-15T10:30:00Z] cpf | technique=hash | rows=50000 | status=success
# [2024-01-15T10:30:00Z] nome | technique=redact | rows=50000 | status=success

# Desativa
lg.configure(audit=None)

Métricas de privacidade

# k-anonimato (ANPD recomenda k >= 5)
report = lg.check.kanon(
    df,
    quasi_identifiers=["uf", "faixa_etaria", "escolaridade"],
    target_k=5,
)
print(f"k={report.k_anonymity.k_value} | ANPD: {report.compliant_anpd}")

# t-closeness
report = lg.check.tcloseness(
    df,
    quasi_identifiers=["uf", "idade"],
    sensitive_attribute="diagnostico",
    target_t=0.2,
)

# Risk score de re-identificação (0 a 1)
report = lg.check.risk(
    df_safe,
    quasi_identifiers=["uf", "faixa_etaria"],
    masked_columns=["cpf", "email"],
)
print(f"Risk: {report.risk_score:.2f} | {report.risk_level}")

# Utilidade preservada após mascaramento
report = lg.check.utility(df_original, df_masked)
print(f"Utilidade: {report.overall_score:.0%}")

---

11. Privacidade diferencial

# Cria mecanismo DP configurado
dp = lg.check.dp(epsilon=1.0)

# Adiciona ruído de Laplace a estatísticas
noisy_mean = dp.laplace(df["salario"].mean(), sensitivity=df["salario"].max())
noisy_count = dp.laplace(len(df), sensitivity=1.0)

# Mecanismo Gaussiano (recomendado para ML)
noisy_value = dp.gaussian(df["renda"].mean(), sensitivity=1000.0)

# Randomized Response (variáveis binárias)
resposta_privada = dp.randomized_response(resposta_real=True)

# Rastreamento de budget
dp.budget.report()
# ε restante: 0.72 / 1.0
# Operações: 2

---

12. Dados sintéticos

# Treina modelo generativo nos dados mascarados
modelo = lg.train(df_masked, epochs=100)

# Gera dataset sintético preservando distribuição
df_synth = lg.clone(df_masked, n=50_000)

# Pipeline completo: mask → clone → avalia fidelidade
resultado = lg.sandbox(df,
    salt="chave",
    n_synth=10_000,
)
print(f"Fidelidade: {resultado.fidelity_score:.2%}")
df_sintetico = resultado.df_synthetic

# Avalia fidelidade estatística
report = lg.check.fidelity(df_original, df_synth)
report.print_report()

---

13. CLI

# Instala e testa
pip install logus-lgpd
logus --help

# Detecta PII em arquivo
logus scan clientes.csv
logus scan clientes.csv --threshold 0.7 --json > relatorio.json

# Mascara e salva
logus mask clientes.csv --salt $LOGUS_SALT --output clientes_masked.csv

# Empacota em .lgs cifrado
logus pack clientes.csv --key $LOGUS_KEY --output clientes.lgs

# Inspeciona .lgs sem decifrar payload
logus inspect clientes.lgs --key $LOGUS_KEY

# Extrai .lgs para CSV
logus unpack clientes.lgs --key $LOGUS_KEY --output clientes.csv

# Diagnóstico rápido
logus profile clientes.csv

# Variáveis de ambiente (recomendado em CI/CD)
export LOGUS_SALT="$(cat /run/secrets/logus_salt)"
export LOGUS_KEY="$(cat /run/secrets/logus_key)"
logus mask clientes.csv --output mascarado.csv

---

14. Segurança — posição e limites

Criptografia

• AES-256-GCM (FIPS 140-3 approved) — confidencialidade + integridade em uma operação
• HKDF-SHA256 (RFC 5869) — deriva DEK e HEK separadas do mesmo key + salt único por arquivo
• HMAC-SHA256 — MAC sobre o arquivo completo (Verify-then-Decrypt)
• Cipher auto-negociado: AES-NI disponível → AES-256-GCM; caso contrário → ChaCha20-Poly1305

O que logus protege

Ameaça	Protegido?	Mecanismo
Arquivo interceptado em trânsito	✅	AES-256-GCM
Arquivo adulterado (bit-flip)	✅	HMAC-SHA256 + GCM auth tag
Metadados vazados sem a key	✅	Header cifrado separadamente com HEK
Re-identificação por CPF bruto	✅	HMAC-SHA256 com salt
Re-identificação por combinação	✅ (parcial)	k-anonimato, quasi-IDs
Força bruta no hash de CPF	✅	Salt de ≥ 16 bytes obrigatório

O que logus NÃO protege

Ameaça	Motivo
Key vazada pelo usuário	Responsabilidade do usuário (use vault, env var)
Dados em memória RAM	GC Python não é determinístico — janela de exposição minimizada, não eliminada
SQL injection em in_db_mask()	Table/column names não são parametrizáveis — use nomes confiáveis
Re-identificação por quasi-IDs	Requer check.kanon() adicional

Boas práticas em produção

# ✅ Keys de fontes seguras
import os
key  = os.environ["LOGUS_KEY"]   # ou vault.get("lgs_key")
salt = os.environ["LOGUS_SALT"]  # diferente da key

# ✅ Key e salt nunca iguais
lg.store(df, "f.lgs", key=key, salt=salt)  # ValueError se key==salt

# ✅ Salt com entropia suficiente
salt = lg.generate_salt()  # 256 bits — hex string de 64 chars

# ✅ Verificação antes de processar
with lg.open("recebido.lgs", key=key) as f:
    if not f.valid():
        raise SecurityError("Arquivo corrompido ou adulterado")
    df = f.read()

# ✅ Auditoria em produção
from logus.reports.audit_report import AuditReport
lg.configure(audit=AuditReport(output_dir="/var/log/logus/"))

# ✅ In-DB mask: revisa antes de executar
result = adapter.in_db_mask("clientes", dry_run=True)
if not all_sql_approved(result["sql_statements"]):
    raise Exception("SQLs não aprovados")
result = adapter.in_db_mask("clientes")

---

15. Referência da API

Funções principais

Função	Descrição
lg.scan(source, *, key, sample_size, threshold)	Detecta colunas PII
lg.mask(df, *, salt, columns, exclude, verbose)	Aplica mascaramento
lg.profile(source, *, key, sample_size)	Diagnóstico completo (JSON-serializable)
lg.diff(original, masked)	Compara antes/depois
lg.join(left, right, on, *, salt, how)	Join seguro com validação de tokens
lg.read(source, *, key, salt, raw, frame)	Lê arquivo (qualquer formato)
lg.store(source, path, *, key, salt, metadata, anonymize)	Salva como .lgs
lg.open(path, *, key, salt)	Retorna LGSFile (context manager)
lg.inspect(path, *, key)	Metadados sem decifrar payload
lg.rekey(path, *, old_key, new_key)	Rotação de chave atômica
lg.stream(source, *, salt, chunksize, on_progress)	Streaming com progresso
lg.configure(*, audit, audit_path)	Configuração global
lg.generate_salt()	Gera salt seguro (256 bits)
lg.read_db(url, sql, *, salt, ...)	Lê banco com mascaramento
lg.load	Alias de lg.read
lg.save	Alias de lg.store

Analytics (API pandas, engine Polars)

describe, value_counts, head, tail, shape, dtypes, nunique, isnull, null_counts, corr, groupby, sort, where, query, select, drop, rename, cast, fillna, sample, unique

LGSFile

Método	Descrição
f.read(*, raw, frame)	Decifra e retorna DataFrame
f.write(df, *, label, metadata)	Sobrescreve arquivo
f.frames()	Retorna todos os frames (multi-frame)
f.frame(name)	Retorna um frame específico
f.add_frame(name, df)	Adiciona frame (converte para multi-frame)
f.info(*)	Metadados sem decifrar payload
f.valid()	True se arquivo existe e tem HMAC correto
f.shape()	(linhas, colunas) sem decifrar
f.frame_names()	Lista de nomes de frames (multi-frame)
f.size_kb()	Tamanho em KB
f.copy_to(dest)	Copia sem decifrar
f.delete()	Remove o arquivo
bool(f)	True se existe e é válido

SecureFile (API de baixo nível)

Método	Parâmetros principais
pack(source_path, output_path, key, ...)	Empacota arquivo de dados
pack_dataframe(df, output_path, key, ...)	Empacota DataFrame
pack_frames(frames, output_path, key, ...)	Empacota dict[str, DataFrame]
pack_bytes(payload, output_path, key, ...)	Empacota bytes arbitrários
pack_open(df, output_path, *, anonymize, ...)	Sem criptografia
load(path, key, *, salt_masking, ...)	Lê com mascaramento
load_raw(path, key)	Lê sem mascaramento adicional
load_frames(path, key, ...)	Lê multi-frame
load_frame(path, *, frame, key, ...)	Lê um frame
load_bytes(path, key)	Lê payload binário
load_open(path, *, anonymize, ...)	Lê arquivo sem criptografia
verify(path, key)	Retorna LGSInfo com __bool__

---

16. Changelog

v1.9.0 (2026-05) — API polida para produção

Bugs corrigidos:

• FileNotFoundError não era mais swallowed como ValueError em inspect(), read(), scan() e profile()
• store() marcava dados brutos como masked_dataframe (header mentia)
• raw=True em CSV era ignorado silenciosamente — agora emite UserWarning
• stacklevel errado no aviso de salt fraco — agora aponta para o código do usuário
• key == salt passa silenciosamente — agora ValueError com mensagem clara

Inconsistências de API corrigidas:

• master_key= renomeado para key= em toda a SecureFile (alias deprecated preservado)
• verify() retorna LGSInfo com __bool__, acesso por atributo e tuple unpacking retrocompatível
• load_frame() agora usa frame= como keyword-only

Novas funcionalidades:

• LGSFile + lg.open() — context manager pythônico para .lgs
• lg.diff() — compara DataFrame antes/depois do mascaramento
• lg.scan() e lg.profile() aceitam caminhos de arquivo (.csv, .parquet, .lgs)
• metadata= em store() — metadados customizados preservados em inspect()
• stream() com on_progress= callback (suporte a tqdm)
• lg.save / lg.load — aliases pythônicos
• profile()['pii_reports'] é JSON-serializable

v1.8.0 — DB + CLI + correções de performance

• CLI: logus scan/mask/inspect/pack/unpack/profile
• lg.mask(columns=, exclude=) — mascaramento seletivo
• lg.configure(audit=) — auditoria global automática
• lg.profile() — diagnóstico integrado
• lg.join() — join seguro com validação de tokens
• link.db() — integração com PostgreSQL, MySQL, SQLite, SQL Server
• adapter.in_db_mask() — mascaramento sem puxar dados para Python
• adapter.create_masked_view() — VIEW mascarada sem alterar dado original
• adapter.write() — escreve DataFrame mascarado no banco
• lg.read_db() — atalho de alto nível para leitura de banco
• Fix: _normalize_identifier vetorizado (3.7× mais rápido)
• Fix: DateMasker com LUT numpy (2.3× mais rápido)
• Fix: PIIDetector converte DataFrame inteiro para Polars uma vez

v1.7.0 — Polars no core + analytics

• Engine Polars ativo no PIIDetector, pandas_adapter e DateMasker
• 20 funções analíticas com API pandas, engine Polars
• lg.where() (substitui lg.filter que shadoweava builtin)
• __dir__() customizado — módulos internos não vazam no namespace
• _detect_best_cipher() lazy — não executa no import

v1.6.0 — Multi-frame + polars_adapter

• Formato .lgs v3: multi-frame (ZIP de Parquets com índice JSON)
• lg.store(dict, ...), lg.read(path, frame=...), LGSFile.frames()
• polars_adapter.py com API espelhada ao pandas_adapter
• lg.pl namespace
• Fix: VERSION_V3 corretamente gravado no byte de versão

v1.5.0 — Formato .lgs v4 + banco de dados

• .lgs v4: sem criptografia (pack_open, load_open)
• SecureDBAdapter com cache de schema e chunked query
• SQLAdapter: geração de scripts SQL com views mascaradas
• AuditReport como parâmetro opcional do _MaskingEngine

---

Licença

GNU Affero General Public License v3 (AGPLv3)

Copyright (C) 2024 Leonardo Borges

Este programa é software livre: você pode redistribuí-lo e/ou modificá-lo
sob os termos da GNU Affero General Public License conforme publicada pela
Free Software Foundation, versão 3.