{{ t('help.run_cli_p1') }}
{{ t('help.run_cli_oneshot') }}
-uv run python main.py --config config.yaml
+ uv run python main.py --demo
+data-boar --demo
+uv run python main.py --config config.yaml
uv run python main.py --config config.yaml --validate-config
uv run python main.py --config config.yaml --diff <session_a> <session_b>
uv run python main.py --config config.yaml --diff <session_a> <session_b> --fail-on-new-high
diff --git a/core/demo/__init__.py b/core/demo/__init__.py
new file mode 100644
index 00000000..8dabc40c
--- /dev/null
+++ b/core/demo/__init__.py
@@ -0,0 +1,21 @@
+"""Installable demo corpus and workspace helpers (#1113)."""
+
+from core.demo.runtime import (
+ prepare_demo_workspace,
+ print_demo_banner,
+ register_demo_cleanup,
+)
+from core.demo.synthetic_corpus import (
+ ALL_SCENARIOS,
+ generate_corpus,
+ main as generate_corpus_cli,
+)
+
+__all__ = [
+ "ALL_SCENARIOS",
+ "generate_corpus",
+ "generate_corpus_cli",
+ "prepare_demo_workspace",
+ "print_demo_banner",
+ "register_demo_cleanup",
+]
diff --git a/core/demo/runtime.py b/core/demo/runtime.py
new file mode 100644
index 00000000..c0f6958a
--- /dev/null
+++ b/core/demo/runtime.py
@@ -0,0 +1,111 @@
+"""Demo workspace preparation for ``data-boar --demo`` (#1113, #834)."""
+
+from __future__ import annotations
+
+import atexit
+import os
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from core.demo.synthetic_corpus import ALL_SCENARIOS, generate_corpus
+
+_DEFAULT_SCENARIOS = "happy,unhappy,false_positive"
+_DEMO_DIRNAME = "data_boar_demo"
+_registered_cleanup: Path | None = None
+
+
+def _default_demo_root() -> Path:
+ return Path(tempfile.gettempdir()) / _DEMO_DIRNAME
+
+
+def _write_demo_config(demo_dir: Path, port: int) -> Path:
+ corpus = demo_dir / "corpus"
+ reports = demo_dir / "reports"
+ reports.mkdir(parents=True, exist_ok=True)
+ config_path = demo_dir / "demo.config.yaml"
+ config_path.write_text(
+ (
+ "targets:\n"
+ " - name: demo-corpus\n"
+ " type: filesystem\n"
+ f" path: {corpus}\n"
+ " recursive: true\n"
+ "\n"
+ "report:\n"
+ f" output_dir: {reports}\n"
+ "\n"
+ f"sqlite_path: {demo_dir / 'audit_results.db'}\n"
+ "\n"
+ "api:\n"
+ f" port: {port}\n"
+ " host: 127.0.0.1\n"
+ " allow_insecure_http: true\n"
+ ),
+ encoding="utf-8",
+ )
+ return config_path
+
+
+def _cleanup_demo_dir(demo_dir: Path) -> None:
+ import shutil
+
+ if demo_dir.exists():
+ shutil.rmtree(demo_dir, ignore_errors=True)
+
+
+def register_demo_cleanup(demo_dir: Path) -> None:
+ """Register atexit cleanup for a single-process ``--demo`` run."""
+ global _registered_cleanup
+ if _registered_cleanup is not None:
+ return
+ _registered_cleanup = demo_dir
+
+ def _on_exit() -> None:
+ _cleanup_demo_dir(demo_dir)
+
+ atexit.register(_on_exit)
+
+
+def print_demo_banner(port: int, demo_dir: Path) -> None:
+ print("")
+ print("╔══════════════════════════════════════════════════════════╗")
+ print("║ Data Boar — Demo (synthetic corpus, zero real data) ║")
+ print("╚══════════════════════════════════════════════════════════╝")
+ print(f"[demo] Workspace: {demo_dir}")
+ print(f"[demo] Dashboard: http://127.0.0.1:{port}/pt-br/")
+ print("[demo] Press Ctrl+C to stop (temp files removed on exit).")
+ print("")
+
+
+def prepare_demo_workspace(
+ *,
+ port: int = 8088,
+ scenarios: str = _DEFAULT_SCENARIOS,
+ demo_root: Path | None = None,
+ register_cleanup: bool = True,
+) -> tuple[Path, Path, dict[str, Any]]:
+ """
+ Generate synthetic corpus + minimal config under a temp directory.
+
+ Returns ``(demo_dir, config_path, config_dict)`` where ``config_dict`` is
+ ready to pass to ``load_config``-equivalent flows (after YAML load).
+ """
+ from config.loader import load_config
+
+ demo_dir = (demo_root or _default_demo_root()).resolve()
+ demo_dir.mkdir(parents=True, exist_ok=True)
+ corpus_dir = demo_dir / "corpus"
+ corpus_dir.mkdir(parents=True, exist_ok=True)
+
+ selected = [s.strip() for s in scenarios.split(",") if s.strip()]
+ generate_corpus(corpus_dir, selected or ALL_SCENARIOS[:3])
+
+ config_path = _write_demo_config(demo_dir, port)
+ os.environ["CONFIG_PATH"] = str(config_path)
+ config = load_config(str(config_path))
+
+ if register_cleanup:
+ register_demo_cleanup(demo_dir)
+
+ return demo_dir, config_path, config
diff --git a/core/demo/synthetic_corpus.py b/core/demo/synthetic_corpus.py
new file mode 100644
index 00000000..c5bb2f1b
--- /dev/null
+++ b/core/demo/synthetic_corpus.py
@@ -0,0 +1,939 @@
+#!/usr/bin/env python3
+"""
+generate_synthetic_poc_corpus.py
+================================
+Generates a synthetic test corpus for Data Boar POC validation.
+
+Covers seven test scenarios:
+ 1. happy -- clear PII in plain-text formats (should be found)
+ 2. unhappy -- PII with OCR noise, encoding quirks (should be found, harder)
+ 3. catastrophic -- nested archives, password-protected zips (may be missed)
+ 4. false_positive -- data that LOOKS like PII but is invalid (should NOT trigger)
+ 5. manual_review -- ambiguous / partial data (flag for human review)
+ 6. stego -- CPF/RG hidden in image LSB (NOT found without stego module)
+ 7. extensions -- one file per supported extension, all containing a CPF
+
+Usage:
+ uv run python scripts/generate_synthetic_poc_corpus.py
+ uv run python scripts/generate_synthetic_poc_corpus.py --scenario happy,stego
+ uv run python scripts/generate_synthetic_poc_corpus.py --output /tmp/poc_corpus
+
+Collaborator note:
+ After generating, point Data Boar at each sub-folder and compare findings
+ against the expected results in EXPECTED.txt (each sub-folder) and
+ docs/TESTING_POC_GUIDE.md (full validation checklist).
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import json
+import sqlite3
+import tarfile
+import textwrap
+import zipfile
+from pathlib import Path
+from typing import Callable
+
+# ---------------------------------------------------------------------------
+# Synthetic PII (deterministic, never real persons)
+# ---------------------------------------------------------------------------
+_CPFS = [
+ "123.456.789-09",
+ "987.654.321-00",
+ "111.222.333-96",
+ "000.000.001-91",
+ "529.982.247-25",
+]
+_CNPJS = ["11.222.333/0001-81", "00.000.000/0001-91", "12.345.678/0001-95"]
+_RGS = ["12.345.678-9", "98.765.432-1", "00.111.222-3"]
+_NAMES = [
+ "Ana Paula Souza",
+ "Carlos Eduardo Lima",
+ "Fernanda Beatriz Costa",
+ "Joao Roberto Colleague-E",
+ "Maria Oliveira Santos",
+]
+_EMAILS = [
+ "ana.souza@example-test.com",
+ "carlos.lima@demo.invalid",
+ "f.costa@poc-databoar.test",
+]
+_PHONES = ["(11) 99999-0001", "+55 21 98888-0002", "0800 123 4567"]
+_DATES = ["15/03/1985", "1990-07-22", "01/01/1970"]
+_ADDRS = [
+ "Rua das Flores, 123, Sao Paulo - SP, CEP 01234-567",
+ "Av. Brasil, 4500, Rio de Janeiro - RJ",
+]
+
+EXPECTED: dict[str, str] = {
+ "1_happy": "DEVE ENCONTRAR -- PII em claro, sem ofuscacao",
+ "2_unhappy": "DEVE ENCONTRAR -- mas pode requerer OCR ou tolerancia a ruido",
+ "3_catastrophic": "PODE NAO ENCONTRAR -- dados em arquivos aninhados ou com senha",
+ "4_false_positive": "NAO DEVE ENCONTRAR -- strings similares a PII mas invalidas",
+ "5_manual_review": "DEVE SINALIZAR PARA REVISAO MANUAL -- dados parcialmente mascarados",
+ "6_stego": "NAO DEVE ENCONTRAR sem modulo estego -- CPF em LSB de imagem PNG",
+ "7_extensions": "DEVE ENCONTRAR em todos os formatos suportados",
+}
+
+
+def _p(lst: list[str], i: int = 0) -> str:
+ return lst[i % len(lst)]
+
+
+def _w(path: Path, content: str | bytes, enc: str = "utf-8") -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ if isinstance(content, bytes):
+ path.write_bytes(content)
+ else:
+ path.write_text(content, encoding=enc)
+
+
+# ---------------------------------------------------------------------------
+# Scenario 1 - Happy path
+# ---------------------------------------------------------------------------
+def gen_scenario_1(base: Path) -> None:
+ out = base / "1_happy"
+
+ _w(
+ out / "employees.txt",
+ textwrap.dedent(f"""\
+ RELATORIO DE FUNCIONARIOS -- FICTICIO -- APENAS PARA TESTES POC
+ Nome: {_p(_NAMES, 0)} CPF: {_p(_CPFS, 0)} RG: {_p(_RGS, 0)}
+ Email: {_p(_EMAILS, 0)} Tel: {_p(_PHONES, 0)}
+ Nasc: {_p(_DATES, 0)} End: {_p(_ADDRS, 0)}
+ Nome: {_p(_NAMES, 1)} CPF: {_p(_CPFS, 1)} CNPJ: {_p(_CNPJS, 0)}
+ """),
+ )
+
+ buf = io.StringIO()
+ csv.writer(buf).writerows(
+ [["nome", "cpf", "rg", "email"]]
+ + [[_p(_NAMES, i), _p(_CPFS, i), _p(_RGS, i), _p(_EMAILS, i)] for i in range(5)]
+ )
+ _w(out / "employees.csv", buf.getvalue())
+
+ _w(
+ out / "employees.json",
+ json.dumps(
+ [
+ {"nome": _p(_NAMES, i), "cpf": _p(_CPFS, i), "rg": _p(_RGS, i)}
+ for i in range(4)
+ ],
+ ensure_ascii=False,
+ indent=2,
+ ),
+ )
+
+ try:
+ from reportlab.pdfgen import canvas as rc
+
+ c = rc.Canvas(str(out / "employees.pdf"))
+ y = 780
+ c.drawString(50, y, "DADOS FICTICIOS -- POC Data Boar")
+ y -= 20
+ for i in range(3):
+ for lbl, v in [
+ ("Nome", _p(_NAMES, i)),
+ ("CPF", _p(_CPFS, i)),
+ ("Email", _p(_EMAILS, i)),
+ ]:
+ c.drawString(50, y, f"{lbl}: {v}")
+ y -= 15
+ c.save()
+ except ImportError:
+ _w(out / "employees_pdf_fallback.txt", f"PDF nao gerado. CPF: {_p(_CPFS, 0)}")
+
+ try:
+ import docx as _d
+
+ doc = _d.Document()
+ doc.add_heading("Dados Ficticios POC", 0)
+ for i in range(3):
+ doc.add_paragraph(
+ f"Nome: {_p(_NAMES, i)}\nCPF: {_p(_CPFS, i)}\nRG: {_p(_RGS, i)}\n"
+ )
+ doc.save(str(out / "employees.docx"))
+ except ImportError:
+ _w(out / "employees_docx_fallback.txt", f"DOCX nao gerado. CPF: {_p(_CPFS, 1)}")
+
+ try:
+ import openpyxl
+
+ wb = openpyxl.Workbook()
+ ws = wb.active
+ ws.title = "Funcionarios"
+ ws.append(["Nome", "CPF", "RG", "Email", "Tel"])
+ for i in range(5):
+ ws.append(
+ [
+ _p(_NAMES, i),
+ _p(_CPFS, i),
+ _p(_RGS, i),
+ _p(_EMAILS, i),
+ _p(_PHONES, i),
+ ]
+ )
+ wb.save(str(out / "employees.xlsx"))
+ except ImportError:
+ _w(out / "employees_xlsx_fallback.txt", f"XLSX nao gerado. CPF: {_p(_CPFS, 2)}")
+
+ conn = sqlite3.connect(str(out / "employees.db"))
+ conn.execute(
+ "CREATE TABLE IF NOT EXISTS emp (id INTEGER PRIMARY KEY,nome TEXT,cpf TEXT,rg TEXT,email TEXT)"
+ )
+ for i in range(5):
+ conn.execute(
+ "INSERT INTO emp(nome,cpf,rg,email) VALUES(?,?,?,?)",
+ (_p(_NAMES, i), _p(_CPFS, i), _p(_RGS, i), _p(_EMAILS, i)),
+ )
+ conn.commit()
+ conn.close()
+
+ try:
+ from PIL import Image, ImageDraw
+
+ img = Image.new("RGB", (400, 120), (255, 255, 255))
+ draw = ImageDraw.Draw(img)
+ draw.text((10, 20), f"CPF: {_p(_CPFS, 0)}", (0, 0, 0))
+ draw.text((10, 50), f"Nome: {_p(_NAMES, 0)}", (0, 0, 0))
+ img.save(str(out / "id_card_visible.png"))
+ except ImportError:
+ _w(out / "image_fallback.txt", f"PNG nao gerado. CPF: {_p(_CPFS, 0)}")
+
+ _w(out / "EXPECTED.txt", EXPECTED["1_happy"])
+ print(f" v Scenario 1 (happy) -> {out}")
+
+
+# ---------------------------------------------------------------------------
+# Scenario 2 - Unhappy path
+# ---------------------------------------------------------------------------
+def gen_scenario_2(base: Path) -> None:
+ out = base / "2_unhappy"
+ import base64 as _b64
+
+ _w(
+ out / "ocr_noisy.txt",
+ f"N0me: {_p(_NAMES, 2).replace('a', '@').replace('e', '3')}\n"
+ f"CPF: {_p(_CPFS, 2).replace('.', ',')} (possivel ruido OCR)\n"
+ f"RG: {_p(_RGS, 2).replace('-', '_')}\n"
+ f"Email: {_p(_EMAILS, 2).replace('@', '[at]')}\n",
+ )
+
+ _w(
+ out / "latin1_encoded.txt",
+ f"Nome: {_p(_NAMES, 0)}\nCPF: {_p(_CPFS, 0)}\nObservacao: dado em latin-1\n",
+ enc="latin-1",
+ )
+
+ _w(
+ out / "bom_utf8.csv",
+ f"\ufeffNome;CPF;RG\n{_p(_NAMES, 1)};{_p(_CPFS, 1)};{_p(_RGS, 1)}\n",
+ enc="utf-8-sig",
+ )
+
+ _w(
+ out / "crlf_endings.txt",
+ f"CPF: {_p(_CPFS, 3)}\r\nTel: {_p(_PHONES, 0)}\r\nEnd: {_p(_ADDRS, 0)}\r\n",
+ )
+
+ _w(
+ out / "partial_redaction.txt",
+ f"Nome: {_p(_NAMES, 0)}\n"
+ f"CPF: ***.{_p(_CPFS, 0)[4:7]}.***-** (parcialmente redactado)\n"
+ f"Email: {_p(_EMAILS, 0)}\nRG: {_p(_RGS, 0)}\n",
+ )
+
+ blob = _b64.b64encode(f"CPF:{_p(_CPFS, 1)},Nome:{_p(_NAMES, 1)}".encode()).decode()
+ _w(out / "base64_embedded.txt", f"campo_documento: {blob}\n# dado acima e base64\n")
+
+ _w(out / "EXPECTED.txt", EXPECTED["2_unhappy"])
+ print(f" v Scenario 2 (unhappy) -> {out}")
+
+
+# ---------------------------------------------------------------------------
+# Scenario 3 - Catastrophic
+# ---------------------------------------------------------------------------
+def gen_scenario_3(base: Path) -> None:
+ out = base / "3_catastrophic"
+ out.mkdir(parents=True, exist_ok=True)
+ pii = f"DADOS FICTICIOS\nCPF: {_p(_CPFS, 4)}\nNome: {_p(_NAMES, 4)}\nRG: {_p(_RGS, 2)}\n".encode()
+
+ # nested zip
+ inner_buf = io.BytesIO()
+ with zipfile.ZipFile(inner_buf, "w", zipfile.ZIP_DEFLATED) as z:
+ z.writestr("pii.txt", pii)
+ with zipfile.ZipFile(out / "nested.zip", "w", zipfile.ZIP_DEFLATED) as z:
+ z.writestr("inner.zip", inner_buf.getvalue())
+
+ # password-protected zip
+ with zipfile.ZipFile(out / "password_protected.zip", "w", zipfile.ZIP_STORED) as z:
+ z.setpassword(b"poc-test-123")
+ z.writestr("secret.txt", pii)
+
+ # tar.gz
+ with tarfile.open(str(out / "archive.tar.gz"), "w:gz") as t:
+ info = tarfile.TarInfo("pii.txt")
+ info.size = len(pii)
+ t.addfile(info, io.BytesIO(pii))
+
+ # tar.bz2
+ with tarfile.open(str(out / "archive.tar.bz2"), "w:bz2") as t:
+ info = tarfile.TarInfo("pii.txt")
+ info.size = len(pii)
+ t.addfile(info, io.BytesIO(pii))
+
+ # disguised extension (text file named .jpg)
+ _w(
+ out / "report_2026.jpg",
+ pii.decode() + "\n# Arquivo de texto mascarado como .jpg\n",
+ )
+
+ # very long line stress test
+ _w(
+ out / "long_line_stress.txt",
+ "x" * 5000 + f" CPF: {_p(_CPFS, 0)} " + "y" * 5000 + "\n",
+ )
+
+ _w(out / "EXPECTED.txt", EXPECTED["3_catastrophic"])
+ _w(
+ out / "PASSWORD_HINT.txt",
+ "Senha: poc-test-123\nConfiguracao: zip_password no config.yaml\n",
+ )
+ print(f" v Scenario 3 (catastrophic) -> {out}")
+
+
+# ---------------------------------------------------------------------------
+# Scenario 4 - False positive pressure
+# ---------------------------------------------------------------------------
+def gen_scenario_4(base: Path) -> None:
+ import random as _r
+
+ out = base / "4_false_positive"
+
+ def _invalid_cpf_shaped() -> str:
+ d = [_r.randint(0, 9) for _ in range(9)]
+ return f"{d[0]}{d[1]}{d[2]}.{d[3]}{d[4]}{d[5]}.{d[6]}{d[7]}{d[8]}-{(d[0] + 1) % 10}{(d[1] + 1) % 10}"
+
+ _w(
+ out / "serial_numbers.txt",
+ "CATALOGO -- FICÇÃO\n"
+ + "\n".join(f"Serial: {_invalid_cpf_shaped()}" for _ in range(10)),
+ )
+ _w(out / "cnpj_shaped_refs.txt", "Ref: 00.111.222/0099-00\n" * 5) # invalid CNPJ
+ _w(
+ out / "random_codes.txt",
+ "\n".join(f"Cod: {_r.randint(10000000000, 99999999999)}" for _ in range(20)),
+ )
+ _w(
+ out / "ip_addresses.txt",
+ "\n".join(f"IP: 10.0.{i}.{j}" for i in range(5) for j in range(5)),
+ )
+ _w(
+ out / "version_strings.txt",
+ "\n".join(f"v: 1.{i}.{i + 1}-{i + 2}" for i in range(10)),
+ )
+
+ _w(out / "EXPECTED.txt", EXPECTED["4_false_positive"])
+ print(f" v Scenario 4 (false_positive) -> {out}")
+
+
+# ---------------------------------------------------------------------------
+# Scenario 5 - Manual review triggers
+# ---------------------------------------------------------------------------
+def gen_scenario_5(base: Path) -> None:
+ out = base / "5_manual_review"
+ _w(
+ out / "masked_pii.txt",
+ textwrap.dedent("""\
+ CPF: ***.456.789-** (mascarado -- padrao parcial visivel)
+ CPF: 123.***.***-09 (mascarado -- inicio e fim visiveis)
+ RG: 12.345.***-*
+ Email: a***.s***@example.com
+ Tel: (11) 9****-0001
+ Nome: Ana P. S. (iniciais -- identificacao possivel com contexto)
+ """),
+ )
+ _w(
+ out / "pii_in_prose.txt",
+ textwrap.dedent("""\
+ O documento de CPF terminado em 09 foi verificado.
+ O numero de registro e 123456789 (sem pontuacao -- validacao manual necessaria).
+ O titular nasceu em quinze de marco de 1985.
+ """),
+ )
+ _w(
+ out / "foreign_pii.txt",
+ textwrap.dedent("""\
+ DNI: 12345678A (Espanha -- nao e CPF brasileiro)
+ SSN: 123-45-6789 (EUA -- nao e CPF)
+ NIF: X1234567L (Espanha -- estrangeiro)
+ """),
+ )
+ _w(
+ out / "anonymized_columns.csv",
+ "cpf,nome,email\n[ANONIMIZADO],[ANONIMIZADO],[ANONIMIZADO]\n" * 5,
+ )
+
+ _w(out / "EXPECTED.txt", EXPECTED["5_manual_review"])
+ print(f" v Scenario 5 (manual_review) -> {out}")
+
+
+# ---------------------------------------------------------------------------
+# Scenario 6 - Steganography (LSB + EXIF metadata)
+# ---------------------------------------------------------------------------
+def _embed_lsb(img_path: Path, secret: str) -> None:
+ from PIL import Image
+
+ img = Image.new("RGB", (200, 200), (200, 200, 200))
+ pixels = list(img.getdata())
+ bits = "".join(f"{ord(c):08b}" for c in secret) + "00000000"
+ new_pixels = []
+ for i, (r, g, b) in enumerate(pixels):
+ if i < len(bits):
+ r = (r & 0xFE) | int(bits[i])
+ new_pixels.append((r, g, b))
+ out_img = Image.new("RGB", (200, 200))
+ out_img.putdata(new_pixels)
+ out_img.save(str(img_path), format="PNG")
+
+
+def _extract_lsb(img_path: Path) -> str:
+ from PIL import Image
+
+ pixels = list(Image.open(str(img_path)).getdata())
+ bits = [str(r & 1) for r, g, b in pixels]
+ chars = []
+ for i in range(0, len(bits) - 8, 8):
+ c = chr(int("".join(bits[i : i + 8]), 2))
+ if c == "\x00":
+ break
+ chars.append(c)
+ return "".join(chars)
+
+
+def gen_scenario_6(base: Path) -> None:
+ out = base / "6_stego"
+ out.mkdir(parents=True, exist_ok=True)
+ try:
+ from PIL import Image
+ from PIL.PngImagePlugin import PngInfo
+
+ secret = f"CPF:{_p(_CPFS, 0)};Nome:{_p(_NAMES, 0)}"
+ stego_path = out / "innocent_photo.png"
+ _embed_lsb(stego_path, secret)
+ recovered = _extract_lsb(stego_path)
+ assert recovered == secret, f"LSB mismatch: {recovered!r}"
+
+ _w(
+ out / "STEGO_KEY.txt",
+ f"Arquivo: innocent_photo.png\n"
+ f"Dado oculto (LSB canal R): {secret}\n"
+ f"Metodo: LSB -- canal R da imagem PNG (1 bit por pixel)\n"
+ f"Para extrair manualmente: use stegosuite, steghide, ou a funcao _extract_lsb() neste script.\n"
+ f"Verificacao OK: dado recuperado = {recovered!r}\n",
+ )
+
+ # EXIF / PNG metadata injection
+ img = Image.new("RGB", (200, 200), (180, 200, 220))
+ meta = PngInfo()
+ meta.add_text("Comment", f"CPF:{_p(_CPFS, 1)} Nome:{_p(_NAMES, 1)}")
+ meta.add_text("Author", _p(_NAMES, 1))
+ img.save(str(out / "photo_with_exif_pii.png"), pnginfo=meta)
+
+ _w(
+ out / "EXPECTED.txt",
+ EXPECTED["6_stego"] + "\n\n"
+ "VALIDACAO MANUAL:\n"
+ "1. innocent_photo.png -- CPF em LSB. Scanner padrao NAO detecta.\n"
+ ' Extrair: uv run python -c "'
+ "from scripts.generate_synthetic_poc_corpus import _extract_lsb;"
+ "from pathlib import Path; print(_extract_lsb(Path('tests/synthetic_corpus/6_stego/innocent_photo.png')))\"\n"
+ "2. photo_with_exif_pii.png -- CPF em metadado PNG (Comment). Scanner PODE detectar se ler metadata.\n",
+ )
+
+ print(f" v Scenario 6 (stego) -> {out} [LSB OK, recovered={recovered!r}]")
+
+ except ImportError:
+ _w(
+ out / "EXPECTED.txt",
+ "Pillow nao disponivel -- cenario 6 nao gerado.\nInstale: pip install pillow\n"
+ + EXPECTED["6_stego"],
+ )
+ print(
+ " ! Scenario 6 (stego) -> Pillow indisponivel, documentado sem gerar imagem"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 7 - Extension coverage (one file per supported extension)
+# ---------------------------------------------------------------------------
+def gen_all_extensions(base: Path) -> None:
+ out = base / "7_extensions"
+ out.mkdir(parents=True, exist_ok=True)
+ pii = f"CPF: {_p(_CPFS, 0)}\nNome: {_p(_NAMES, 0)}\n"
+ pii_b = pii.encode()
+
+ for ext in [
+ ".txt",
+ ".log",
+ ".md",
+ ".rst",
+ ".cfg",
+ ".ini",
+ ".env",
+ ".yml",
+ ".yaml",
+ ".sql",
+ ]:
+ _w(out / f"sample{ext}", pii)
+
+ _w(
+ out / "sample.json",
+ json.dumps({"cpf": _p(_CPFS, 0), "nome": _p(_NAMES, 0)}, ensure_ascii=False),
+ )
+ _w(
+ out / "sample.xml",
+ f'{_p(_CPFS, 0)} {_p(_NAMES, 0)} str:
- return lst[i % len(lst)]
-
-
-def _w(path: Path, content: str | bytes, enc: str = "utf-8") -> None:
- path.parent.mkdir(parents=True, exist_ok=True)
- if isinstance(content, bytes):
- path.write_bytes(content)
- else:
- path.write_text(content, encoding=enc)
-
-
-# ---------------------------------------------------------------------------
-# Scenario 1 - Happy path
-# ---------------------------------------------------------------------------
-def gen_scenario_1(base: Path) -> None:
- out = base / "1_happy"
-
- _w(
- out / "employees.txt",
- textwrap.dedent(f"""\
- RELATORIO DE FUNCIONARIOS -- FICTICIO -- APENAS PARA TESTES POC
- Nome: {_p(_NAMES, 0)} CPF: {_p(_CPFS, 0)} RG: {_p(_RGS, 0)}
- Email: {_p(_EMAILS, 0)} Tel: {_p(_PHONES, 0)}
- Nasc: {_p(_DATES, 0)} End: {_p(_ADDRS, 0)}
- Nome: {_p(_NAMES, 1)} CPF: {_p(_CPFS, 1)} CNPJ: {_p(_CNPJS, 0)}
- """),
- )
-
- buf = io.StringIO()
- csv.writer(buf).writerows(
- [["nome", "cpf", "rg", "email"]]
- + [[_p(_NAMES, i), _p(_CPFS, i), _p(_RGS, i), _p(_EMAILS, i)] for i in range(5)]
- )
- _w(out / "employees.csv", buf.getvalue())
-
- _w(
- out / "employees.json",
- json.dumps(
- [
- {"nome": _p(_NAMES, i), "cpf": _p(_CPFS, i), "rg": _p(_RGS, i)}
- for i in range(4)
- ],
- ensure_ascii=False,
- indent=2,
- ),
- )
-
- try:
- from reportlab.pdfgen import canvas as rc
-
- c = rc.Canvas(str(out / "employees.pdf"))
- y = 780
- c.drawString(50, y, "DADOS FICTICIOS -- POC Data Boar")
- y -= 20
- for i in range(3):
- for lbl, v in [
- ("Nome", _p(_NAMES, i)),
- ("CPF", _p(_CPFS, i)),
- ("Email", _p(_EMAILS, i)),
- ]:
- c.drawString(50, y, f"{lbl}: {v}")
- y -= 15
- c.save()
- except ImportError:
- _w(out / "employees_pdf_fallback.txt", f"PDF nao gerado. CPF: {_p(_CPFS, 0)}")
-
- try:
- import docx as _d
-
- doc = _d.Document()
- doc.add_heading("Dados Ficticios POC", 0)
- for i in range(3):
- doc.add_paragraph(
- f"Nome: {_p(_NAMES, i)}\nCPF: {_p(_CPFS, i)}\nRG: {_p(_RGS, i)}\n"
- )
- doc.save(str(out / "employees.docx"))
- except ImportError:
- _w(out / "employees_docx_fallback.txt", f"DOCX nao gerado. CPF: {_p(_CPFS, 1)}")
-
- try:
- import openpyxl
-
- wb = openpyxl.Workbook()
- ws = wb.active
- ws.title = "Funcionarios"
- ws.append(["Nome", "CPF", "RG", "Email", "Tel"])
- for i in range(5):
- ws.append(
- [
- _p(_NAMES, i),
- _p(_CPFS, i),
- _p(_RGS, i),
- _p(_EMAILS, i),
- _p(_PHONES, i),
- ]
- )
- wb.save(str(out / "employees.xlsx"))
- except ImportError:
- _w(out / "employees_xlsx_fallback.txt", f"XLSX nao gerado. CPF: {_p(_CPFS, 2)}")
-
- conn = sqlite3.connect(str(out / "employees.db"))
- conn.execute(
- "CREATE TABLE IF NOT EXISTS emp (id INTEGER PRIMARY KEY,nome TEXT,cpf TEXT,rg TEXT,email TEXT)"
- )
- for i in range(5):
- conn.execute(
- "INSERT INTO emp(nome,cpf,rg,email) VALUES(?,?,?,?)",
- (_p(_NAMES, i), _p(_CPFS, i), _p(_RGS, i), _p(_EMAILS, i)),
- )
- conn.commit()
- conn.close()
-
- try:
- from PIL import Image, ImageDraw
-
- img = Image.new("RGB", (400, 120), (255, 255, 255))
- draw = ImageDraw.Draw(img)
- draw.text((10, 20), f"CPF: {_p(_CPFS, 0)}", (0, 0, 0))
- draw.text((10, 50), f"Nome: {_p(_NAMES, 0)}", (0, 0, 0))
- img.save(str(out / "id_card_visible.png"))
- except ImportError:
- _w(out / "image_fallback.txt", f"PNG nao gerado. CPF: {_p(_CPFS, 0)}")
-
- _w(out / "EXPECTED.txt", EXPECTED["1_happy"])
- print(f" v Scenario 1 (happy) -> {out}")
-
-
-# ---------------------------------------------------------------------------
-# Scenario 2 - Unhappy path
-# ---------------------------------------------------------------------------
-def gen_scenario_2(base: Path) -> None:
- out = base / "2_unhappy"
- import base64 as _b64
-
- _w(
- out / "ocr_noisy.txt",
- f"N0me: {_p(_NAMES, 2).replace('a', '@').replace('e', '3')}\n"
- f"CPF: {_p(_CPFS, 2).replace('.', ',')} (possivel ruido OCR)\n"
- f"RG: {_p(_RGS, 2).replace('-', '_')}\n"
- f"Email: {_p(_EMAILS, 2).replace('@', '[at]')}\n",
- )
-
- _w(
- out / "latin1_encoded.txt",
- f"Nome: {_p(_NAMES, 0)}\nCPF: {_p(_CPFS, 0)}\nObservacao: dado em latin-1\n",
- enc="latin-1",
- )
-
- _w(
- out / "bom_utf8.csv",
- f"\ufeffNome;CPF;RG\n{_p(_NAMES, 1)};{_p(_CPFS, 1)};{_p(_RGS, 1)}\n",
- enc="utf-8-sig",
- )
-
- _w(
- out / "crlf_endings.txt",
- f"CPF: {_p(_CPFS, 3)}\r\nTel: {_p(_PHONES, 0)}\r\nEnd: {_p(_ADDRS, 0)}\r\n",
- )
-
- _w(
- out / "partial_redaction.txt",
- f"Nome: {_p(_NAMES, 0)}\n"
- f"CPF: ***.{_p(_CPFS, 0)[4:7]}.***-** (parcialmente redactado)\n"
- f"Email: {_p(_EMAILS, 0)}\nRG: {_p(_RGS, 0)}\n",
- )
-
- blob = _b64.b64encode(f"CPF:{_p(_CPFS, 1)},Nome:{_p(_NAMES, 1)}".encode()).decode()
- _w(out / "base64_embedded.txt", f"campo_documento: {blob}\n# dado acima e base64\n")
-
- _w(out / "EXPECTED.txt", EXPECTED["2_unhappy"])
- print(f" v Scenario 2 (unhappy) -> {out}")
-
-
-# ---------------------------------------------------------------------------
-# Scenario 3 - Catastrophic
-# ---------------------------------------------------------------------------
-def gen_scenario_3(base: Path) -> None:
- out = base / "3_catastrophic"
- out.mkdir(parents=True, exist_ok=True)
- pii = f"DADOS FICTICIOS\nCPF: {_p(_CPFS, 4)}\nNome: {_p(_NAMES, 4)}\nRG: {_p(_RGS, 2)}\n".encode()
-
- # nested zip
- inner_buf = io.BytesIO()
- with zipfile.ZipFile(inner_buf, "w", zipfile.ZIP_DEFLATED) as z:
- z.writestr("pii.txt", pii)
- with zipfile.ZipFile(out / "nested.zip", "w", zipfile.ZIP_DEFLATED) as z:
- z.writestr("inner.zip", inner_buf.getvalue())
-
- # password-protected zip
- with zipfile.ZipFile(out / "password_protected.zip", "w", zipfile.ZIP_STORED) as z:
- z.setpassword(b"poc-test-123")
- z.writestr("secret.txt", pii)
-
- # tar.gz
- with tarfile.open(str(out / "archive.tar.gz"), "w:gz") as t:
- info = tarfile.TarInfo("pii.txt")
- info.size = len(pii)
- t.addfile(info, io.BytesIO(pii))
-
- # tar.bz2
- with tarfile.open(str(out / "archive.tar.bz2"), "w:bz2") as t:
- info = tarfile.TarInfo("pii.txt")
- info.size = len(pii)
- t.addfile(info, io.BytesIO(pii))
-
- # disguised extension (text file named .jpg)
- _w(
- out / "report_2026.jpg",
- pii.decode() + "\n# Arquivo de texto mascarado como .jpg\n",
- )
-
- # very long line stress test
- _w(
- out / "long_line_stress.txt",
- "x" * 5000 + f" CPF: {_p(_CPFS, 0)} " + "y" * 5000 + "\n",
- )
-
- _w(out / "EXPECTED.txt", EXPECTED["3_catastrophic"])
- _w(
- out / "PASSWORD_HINT.txt",
- "Senha: poc-test-123\nConfiguracao: zip_password no config.yaml\n",
- )
- print(f" v Scenario 3 (catastrophic) -> {out}")
-
-
-# ---------------------------------------------------------------------------
-# Scenario 4 - False positive pressure
-# ---------------------------------------------------------------------------
-def gen_scenario_4(base: Path) -> None:
- import random as _r
-
- out = base / "4_false_positive"
-
- def _invalid_cpf_shaped() -> str:
- d = [_r.randint(0, 9) for _ in range(9)]
- return f"{d[0]}{d[1]}{d[2]}.{d[3]}{d[4]}{d[5]}.{d[6]}{d[7]}{d[8]}-{(d[0] + 1) % 10}{(d[1] + 1) % 10}"
-
- _w(
- out / "serial_numbers.txt",
- "CATALOGO -- FICÇÃO\n"
- + "\n".join(f"Serial: {_invalid_cpf_shaped()}" for _ in range(10)),
- )
- _w(out / "cnpj_shaped_refs.txt", "Ref: 00.111.222/0099-00\n" * 5) # invalid CNPJ
- _w(
- out / "random_codes.txt",
- "\n".join(f"Cod: {_r.randint(10000000000, 99999999999)}" for _ in range(20)),
- )
- _w(
- out / "ip_addresses.txt",
- "\n".join(f"IP: 10.0.{i}.{j}" for i in range(5) for j in range(5)),
- )
- _w(
- out / "version_strings.txt",
- "\n".join(f"v: 1.{i}.{i + 1}-{i + 2}" for i in range(10)),
- )
-
- _w(out / "EXPECTED.txt", EXPECTED["4_false_positive"])
- print(f" v Scenario 4 (false_positive) -> {out}")
-
-
-# ---------------------------------------------------------------------------
-# Scenario 5 - Manual review triggers
-# ---------------------------------------------------------------------------
-def gen_scenario_5(base: Path) -> None:
- out = base / "5_manual_review"
- _w(
- out / "masked_pii.txt",
- textwrap.dedent("""\
- CPF: ***.456.789-** (mascarado -- padrao parcial visivel)
- CPF: 123.***.***-09 (mascarado -- inicio e fim visiveis)
- RG: 12.345.***-*
- Email: a***.s***@example.com
- Tel: (11) 9****-0001
- Nome: Ana P. S. (iniciais -- identificacao possivel com contexto)
- """),
- )
- _w(
- out / "pii_in_prose.txt",
- textwrap.dedent("""\
- O documento de CPF terminado em 09 foi verificado.
- O numero de registro e 123456789 (sem pontuacao -- validacao manual necessaria).
- O titular nasceu em quinze de marco de 1985.
- """),
- )
- _w(
- out / "foreign_pii.txt",
- textwrap.dedent("""\
- DNI: 12345678A (Espanha -- nao e CPF brasileiro)
- SSN: 123-45-6789 (EUA -- nao e CPF)
- NIF: X1234567L (Espanha -- estrangeiro)
- """),
- )
- _w(
- out / "anonymized_columns.csv",
- "cpf,nome,email\n[ANONIMIZADO],[ANONIMIZADO],[ANONIMIZADO]\n" * 5,
- )
-
- _w(out / "EXPECTED.txt", EXPECTED["5_manual_review"])
- print(f" v Scenario 5 (manual_review) -> {out}")
-
-
-# ---------------------------------------------------------------------------
-# Scenario 6 - Steganography (LSB + EXIF metadata)
-# ---------------------------------------------------------------------------
-def _embed_lsb(img_path: Path, secret: str) -> None:
- from PIL import Image
-
- img = Image.new("RGB", (200, 200), (200, 200, 200))
- pixels = list(img.getdata())
- bits = "".join(f"{ord(c):08b}" for c in secret) + "00000000"
- new_pixels = []
- for i, (r, g, b) in enumerate(pixels):
- if i < len(bits):
- r = (r & 0xFE) | int(bits[i])
- new_pixels.append((r, g, b))
- out_img = Image.new("RGB", (200, 200))
- out_img.putdata(new_pixels)
- out_img.save(str(img_path), format="PNG")
-
-
-def _extract_lsb(img_path: Path) -> str:
- from PIL import Image
-
- pixels = list(Image.open(str(img_path)).getdata())
- bits = [str(r & 1) for r, g, b in pixels]
- chars = []
- for i in range(0, len(bits) - 8, 8):
- c = chr(int("".join(bits[i : i + 8]), 2))
- if c == "\x00":
- break
- chars.append(c)
- return "".join(chars)
-
-
-def gen_scenario_6(base: Path) -> None:
- out = base / "6_stego"
- out.mkdir(parents=True, exist_ok=True)
- try:
- from PIL import Image
- from PIL.PngImagePlugin import PngInfo
-
- secret = f"CPF:{_p(_CPFS, 0)};Nome:{_p(_NAMES, 0)}"
- stego_path = out / "innocent_photo.png"
- _embed_lsb(stego_path, secret)
- recovered = _extract_lsb(stego_path)
- assert recovered == secret, f"LSB mismatch: {recovered!r}"
-
- _w(
- out / "STEGO_KEY.txt",
- f"Arquivo: innocent_photo.png\n"
- f"Dado oculto (LSB canal R): {secret}\n"
- f"Metodo: LSB -- canal R da imagem PNG (1 bit por pixel)\n"
- f"Para extrair manualmente: use stegosuite, steghide, ou a funcao _extract_lsb() neste script.\n"
- f"Verificacao OK: dado recuperado = {recovered!r}\n",
- )
-
- # EXIF / PNG metadata injection
- img = Image.new("RGB", (200, 200), (180, 200, 220))
- meta = PngInfo()
- meta.add_text("Comment", f"CPF:{_p(_CPFS, 1)} Nome:{_p(_NAMES, 1)}")
- meta.add_text("Author", _p(_NAMES, 1))
- img.save(str(out / "photo_with_exif_pii.png"), pnginfo=meta)
-
- _w(
- out / "EXPECTED.txt",
- EXPECTED["6_stego"] + "\n\n"
- "VALIDACAO MANUAL:\n"
- "1. innocent_photo.png -- CPF em LSB. Scanner padrao NAO detecta.\n"
- ' Extrair: uv run python -c "'
- "from scripts.generate_synthetic_poc_corpus import _extract_lsb;"
- "from pathlib import Path; print(_extract_lsb(Path('tests/synthetic_corpus/6_stego/innocent_photo.png')))\"\n"
- "2. photo_with_exif_pii.png -- CPF em metadado PNG (Comment). Scanner PODE detectar se ler metadata.\n",
- )
-
- print(f" v Scenario 6 (stego) -> {out} [LSB OK, recovered={recovered!r}]")
-
- except ImportError:
- _w(
- out / "EXPECTED.txt",
- "Pillow nao disponivel -- cenario 6 nao gerado.\nInstale: pip install pillow\n"
- + EXPECTED["6_stego"],
- )
- print(
- " ! Scenario 6 (stego) -> Pillow indisponivel, documentado sem gerar imagem"
- )
-
-
-# ---------------------------------------------------------------------------
-# Scenario 7 - Extension coverage (one file per supported extension)
-# ---------------------------------------------------------------------------
-def gen_all_extensions(base: Path) -> None:
- out = base / "7_extensions"
- out.mkdir(parents=True, exist_ok=True)
- pii = f"CPF: {_p(_CPFS, 0)}\nNome: {_p(_NAMES, 0)}\n"
- pii_b = pii.encode()
-
- for ext in [
- ".txt",
- ".log",
- ".md",
- ".rst",
- ".cfg",
- ".ini",
- ".env",
- ".yml",
- ".yaml",
- ".sql",
- ]:
- _w(out / f"sample{ext}", pii)
-
- _w(
- out / "sample.json",
- json.dumps({"cpf": _p(_CPFS, 0), "nome": _p(_NAMES, 0)}, ensure_ascii=False),
- )
- _w(
- out / "sample.xml",
- f'{_p(_CPFS, 0)} {_p(_NAMES, 0)}