ryehr/hitms_steganographyBetten Sie mehrere geheime Nachrichten in LLM-Chat-Token-Auswahlen ein, mithilfe arithmetischer/Discop-steganografischer Kodierer, mit bitexakter Dekodierung und Steganalyse-Auswertung.
Forschungscode für linguistische Steganografie in LLM-Chat-Dialogen, mit einem Einzelstrom-Baseline und einem batch-basierten Mehrfachstrom-Protokoll (HiTMS), das mehrere unabhängige Geheimnachrichten gleichzeitig verbirgt und GPU-Batching für einen deutlich höheren Durchsatz nutzt.
Ein Bob-Modell stellt Fragen; ein Alice-Modell antwortet und kodiert die Nutzlast heimlich über einen steganografischen Kodierer in seine Token-Wahlen; Bob führt das Modell erneut auf Alices Antwort aus, um die Bits wiederherzustellen. Kodierung und Dekodierung sind bitexakt, sodass das Geheimnis verlustfrei wiederhergestellt wird.
arithmetic*.py)discop*.py)single_stream.py): kein Frame-Overhead – jedes Kanalbit ist Nutzlast (~100 % Auslastung).protocol.py): m geheime Streams, fragmentiert über Runden, eine PRF-gesteuerte Stream→Slot-Zuordnung, Köder-Slots, 16-Bit-Längen-Header und ein Füller-Keystream. Mehrere Antworten pro Runde werden in einem gebatchten Forward-Pass erzeugt, sodass der Durchsatz mit der Batchgröße skaliert.In den Experimenten verwendete Modelle: meta-llama/Llama-3.2-3B-Instruct und google/gemma-3-4b-it. Datensätze: kreative/offene Teilmengen von databricks/databricks-dolly-15k und HuggingFaceH4/no_robots.
| Pfad | Beschreibung |
|---|---|
arithmetic.py, arithmetic_batch.py | Arithmetikkodierungs-Stego (Einzel- + Batch-Stream-Encoder/-Decoder). |
discop.py, discop_batch.py | Discop-Stego-Kodierer + gemeinsamer PRG (einzeln + gebatcht). |
utils.py | Gemeinsame Hilfsfunktionen (Bit/Int-Konvertierung, Entropie, Top-k, Satzende-Prüfungen). |
protocol.py | Mehrfachstrom-HiTMS-Protokoll (Stream-Zuordnung, Köder, Header, Füller). |
single_stream.py | Einzelstrom-Protokoll (kein Framing; Abbruch am L-ten Bit). |
round_trip.py, batch_round_trip.py | Minimale One-Shot-Roundtrip-Demos/Tests. |
multi_round_demo.py | Mehrfachstrom-Trial-Runner (run_trial) + ausführliche Einzeltrial-Demo. |
single_stream_demo.py | Einzelstrom-Trial-Runner (run_single_trial) + Demo. |
run_sweep.py | Mehrfachstrom-Sweep über {model}×{pool}×{coder}, fortsetzbar. |
run_single_sweep.py | Einzelstrom-Pendant-Sweep. |
run_scaling_sweep.py | Stream-Anzahl-Skalierungssweep (x ∈ {1,2,4,8,16,32,64}), beliebiges Modell/Pool/Kodierer. |
run_x1_shards.py, run_x1_finish.sh, merge_x1_shards.py | x=1-Ablation: eine Zelle über GPUs aufteilen (Sharding), dann mit Abdeckungsprüfungen zusammenführen. |
judge_quality.py | LLM-as-a-Judge-Unauffälligkeits-Scorer (ein QA pro Aufruf, fortsetzbar). |
gen_cover.py | Nutzdatenfreie „Cover“-Texterzeugung (Steganalyse-Referenz). |
steganalysis_bert.py | Cover-gegen-Stego-Erkenner (BERT / RoBERTa / DeBERTa-v3 / ELECTRA). |
export_data.py | Erstellt den veröffentlichbaren, nur Zusammenfassungen enthaltenden Spiegel in data/. |
build_question_pool.py, build_norobots_pool.py | Erstellt die Fragenpools aus HF-Datensätzen. |
*_creative_questions.json | Vorgefertigte Fragenpools. |
legacy/ | Frühere Skripte/Pools, zur Referenz aufbewahrt. |
data/ enthält die Ergebnisse jedes Experiments des Papers – Kapazität, Durchsatz, Auslastung, Judge-Scores und Detektor-AUROC-Werte – als ein JSON-Objekt pro Trial. Das vollständige Schema finden Sie in data/README.md.
Der erzeugte Stegotext selbst ist nicht enthalten: Die Roh-Logs betten Frage und Antwort für jedes Fragment ein, was sie auf ~1 GB bringt; daher entfernt export_data.py diese Felder und behält alle Messwerte (~24 MB). Alle Treiber sind mit festen Seeds versehen, sodass ein erneuter Sweep den Text exakt neu erzeugt.
Die rohen Ausgabeverzeichnisse (
sweep_logs/,single_sweep_logs/,scaling_logs/,judge_logs/,cover_logs/,run_logs/,steganalysis_logs/) sind gitignored – sie sind groß (das letzte enthält mehrere GB große trainierte Detektor-Checkpoints) und vollständig regenerierbar.
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
Für den Betrieb der LLMs ist eine CUDA-GPU erforderlich. Die Llama- und Gemma-Checkpoints sind auf dem Hugging-Face-Hub gated; führen Sie daher vor der ersten Verwendung huggingface-cli login aus (mit Zugriff auf diese Modelle).
Erstellen Sie die Fragenpools (einmal):
python build_question_pool.py # -> dolly15k_creative_questions.json
python build_norobots_pool.py # -> norobots_creative_questions.json
Einzelner ausführlicher Testlauf (Sanity-Check):
# 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
Vollständige Experimente (fortsetzbar – führen Sie denselben Befehl erneut aus, um fortzufahren):
# 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
Evaluierung der Unauffälligkeit (benötigt einen OpenAI-Schlüssel in OPENAI_API_KEY oder eine lokale OPENAI_API_key.txt, beide gitignored):
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
Jeder Treiber legt seine Seeds fest (Prompts-Mischen, Nutzlast-Sampling, Sampling-RNG, torch.manual_seed) und bezieht Prompts aus einem deterministischen, durchgangsbewussten Prompt-Stream, sodass ein vollständiger Sweep Ende-zu-Ende reproduzierbar ist und nach einer Unterbrechung exakt an seinem Checkpoint fortsetzt.
OPENAI_API_key.txt, *.key und .env werden ignoriert.