HiTMS: Un marco de esteganografía lingüística multi-flujo de alto rendimiento

Código de investigación para esteganografía lingüística en diálogos de chat con LLM, con una línea base de flujo único y un protocolo multi-flujo (HiTMS) por lotes que oculta varios mensajes secretos independientes a la vez y aprovecha el procesamiento por lotes de la GPU para obtener un rendimiento mucho mayor.

Un modelo Bob hace preguntas; un modelo Alice responde, codificando secretamente la carga útil en sus elecciones de tokens mediante un codificador esteganográfico; Bob vuelve a ejecutar el modelo sobre la respuesta de Alice para recuperar los bits. La codificación y la decodificación son exactas a nivel de bit, por lo que el secreto se recupera sin pérdidas.

Características destacadas

• Dos codificadores, ambos con codificación/decodificación idéntica a nivel de bit:
  • Arithmetic coding (arithmetic*.py)
  • Discop (que preserva la distribución, discop*.py)
• Protocolo de flujo único (single_stream.py): sin sobrecarga de encuadre — cada bit del canal es carga útil (~100 % de utilización).
• Protocolo HiTMS multi-flujo (protocol.py): m flujos secretos fragmentados a lo largo de las rondas, un mapeo flujo→ranura impulsado por PRF, ranuras señuelo, cabeceras de longitud de 16 bits y un flujo de claves de relleno. Múltiples respuestas por ronda se generan en una única pasada por lotes, por lo que el rendimiento escala con el tamaño del lote.
• Controladores de experimento reproducibles con flujos de mensajes deterministas, checkpoint/reanudación por prueba y manejo de reintentos ante fallos/resultados no válidos.
• Evaluación de imperceptibilidad con LLM como juez (naturalidad + coherencia).

Modelos usados en los experimentos: meta-llama/Llama-3.2-3B-Instruct y google/gemma-3-4b-it. Conjuntos de datos: subconjuntos creativos/abiertos de databricks/databricks-dolly-15k y HuggingFaceH4/no_robots.

Estructura del repositorio

Ruta	Descripción
arithmetic.py, arithmetic_batch.py	Estego de codificación aritmética (codificador/descodificador de flujo único y por lotes).
discop.py, discop_batch.py	Codificador esteganográfico Discop + PRG compartido (flujo único y por lotes).
utils.py	Utilidades compartidas (conversión bit/entero, entropía, top-k, comprobaciones de fin de frase).
protocol.py	Protocolo HiTMS multi-flujo (mapeo de flujos, señuelos, cabeceras, relleno).
single_stream.py	Protocolo de flujo único (sin encuadre; truncar en el bit L-ésimo).
round_trip.py, batch_round_trip.py	Demostraciones/pruebas mínimas de ida y vuelta en una sola pasada.
multi_round_demo.py	Ejecutor de pruebas multi-flujo (run_trial) + demostración detallada de una sola prueba.
single_stream_demo.py	Ejecutor de pruebas de flujo único (run_single_trial) + demostración.
run_sweep.py	Barrido multi-flujo sobre {model}×{pool}×{coder}, reanudable.
run_single_sweep.py	Barrido equivalente de flujo único.
run_scaling_sweep.py	Barrido de escalado por número de flujos (x ∈ {1,2,4,8,16,32,64}), con cualquier modelo/pool/coder.
run_x1_shards.py, run_x1_finish.sh, merge_x1_shards.py	Ablación x=1: divide una celda entre varias GPUs y luego fusiona con comprobaciones de cobertura.
judge_quality.py	Puntuador de imperceptibilidad con LLM como juez (una QA por llamada, reanudable).
gen_cover.py	Generación de texto "cover" sin carga útil (referencia para estegaanálisis).
steganalysis_bert.py	Detector cover-vs-stego (BERT / RoBERTa / DeBERTa-v3 / ELECTRA).
export_data.py	Construye el espejo publicable de solo resumen en data/.
build_question_pool.py, build_norobots_pool.py	Construyen los pools de preguntas a partir de conjuntos de datos de HF.
*_creative_questions.json	Pools de preguntas preconstruidos.
legacy/	Scripts/pools anteriores, conservados como referencia.

Datos

data/ contiene los resultados de todos los experimentos del artículo — capacidad, rendimiento, utilización, puntuaciones del juez y AUROC del detector — como un objeto JSON por prueba. Consulta data/README.md para ver el esquema completo.

El stegotexto generado en sí no se incluye: los registros sin procesar incrustan la pregunta y la respuesta de cada fragmento, lo que los hace de ~1 GB, por lo que export_data.py elimina esos campos y conserva todas las mediciones (~24 MB). Todos los controladores tienen semilla fija, por lo que volver a ejecutar un barrido regenera el texto exactamente.

Los directorios de salida sin procesar (sweep_logs/, single_sweep_logs/, scaling_logs/, judge_logs/, cover_logs/, run_logs/, steganalysis_logs/) están en gitignore — son grandes (el último contiene checkpoints de detectores entrenados de varios GB) y totalmente regenerables.

Instalación

conda create -n ems python=3.10 -y && conda activate ems
pip install torch transformers datasets numpy
pip install openai          # only needed for judge_quality.py

Se requiere una GPU CUDA para ejecutar los LLM. Los checkpoints de Llama y Gemma están restringidos en el Hugging Face Hub, así que ejecuta huggingface-cli login (con acceso concedido a esos modelos) antes del primer uso.

Uso

Construye los pools de preguntas (una vez):

python build_question_pool.py        # -> dolly15k_creative_questions.json
python build_norobots_pool.py        # -> norobots_creative_questions.json

Prueba única detallada (comprobación de cordura):

# multi-stream
CUDA_VISIBLE_DEVICES=0 python multi_round_demo.py
# single-stream
CUDA_VISIBLE_DEVICES=0 python single_stream_demo.py
# override model / coder / pool via env
ROUND_TRIP_MODEL=google/gemma-3-4b-it STEGO_ALGORITHM=discop \
  STEGO_QUESTION_POOL=norobots_creative_questions.json \
  CUDA_VISIBLE_DEVICES=0 python multi_round_demo.py

Experimentos completos (reanudables: vuelve a ejecutar el mismo comando para continuar):

# multi-stream (8 streams x 1024 bits), all model/pool/coder combos, 500 trials
CUDA_VISIBLE_DEVICES=0 python run_sweep.py --trials 500
# single-stream baseline
CUDA_VISIBLE_DEVICES=0 python run_single_sweep.py --trials 500
# stream-count scaling (dolly + Llama + Discop)
CUDA_VISIBLE_DEVICES=0 python run_scaling_sweep.py --x-values 4 8 16 32 64

Evaluación de imperceptibilidad (necesita una clave de OpenAI en OPENAI_API_KEY o un OPENAI_API_key.txt local, ambos en gitignore):

python judge_quality.py --dry-run        # plan only, no API calls
python judge_quality.py --limit 2        # tiny live smoke test
python judge_quality.py                  # full run (resumable)
python judge_quality.py --aggregate-only # recompute the score table

Reproducibilidad

Cada controlador fija sus semillas (barajado de mensajes, muestreo de la carga útil, RNG de muestreo, torch.manual_seed) y consume los mensajes de un flujo de mensajes determinista y consciente de la pasada, por lo que un barrido completo es reproducible de principio a fin y se reanuda exactamente desde su checkpoint tras una interrupción.

Notas

• Este es código de investigación que acompaña a un artículo en preparación; las APIs pueden cambiar.
• Nunca hagas commit de secretos: OPENAI_API_key.txt, *.key y .env están ignorados.