From 2fe1d68bdc92b678fbfd99e8053673a59d23e753 Mon Sep 17 00:00:00 2001 From: Bouletto <118677264+orion4d@users.noreply.github.com> Date: Sat, 6 Sep 2025 14:11:05 +0200 Subject: [PATCH] Initial commit --- .gitattributes | 2 + LICENSE | 21 +++ README.md | 247 ++++++++++++++++++++++++++ __init__.py | 48 +++++ generate_passphrase.py | 208 ++++++++++++++++++++++ image_cipher.py | 222 +++++++++++++++++++++++ jpeg_stego.py | 380 ++++++++++++++++++++++++++++++++++++++++ stego_capacity.py | 59 +++++++ stego_image_in_image.py | 350 ++++++++++++++++++++++++++++++++++++ stego_text.py | 329 ++++++++++++++++++++++++++++++++++ text_cipher.py | 125 +++++++++++++ 11 files changed, 1991 insertions(+) create mode 100644 .gitattributes create mode 100644 LICENSE create mode 100644 README.md create mode 100644 __init__.py create mode 100644 generate_passphrase.py create mode 100644 image_cipher.py create mode 100644 jpeg_stego.py create mode 100644 stego_capacity.py create mode 100644 stego_image_in_image.py create mode 100644 stego_text.py create mode 100644 text_cipher.py diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..dfe0770 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +# Auto detect text files and perform LF normalization +* text=auto diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..5a57b1f --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Philippe Joye (Orion4D) + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..7b82158 --- /dev/null +++ b/README.md @@ -0,0 +1,247 @@ +# 🔐 ComfyUI EncryptMaster + +Pack **ComfyUI** de nƓuds pour **chiffrer** et **cacher** du texte ou des images. +SĂ©curitĂ© : **AES-256-GCM** (authentifiĂ©) avec dĂ©rivation de clĂ© **scrypt**. +StĂ©ganographie : **LSB** (PNG/TIFF) et **DCT/JPEG** (plus robuste aux recompressions). + +--- + +## 📁 Contenu + +Dossier : `ComfyUI/custom_nodes/Comfyui_EncryptMaster/` + +- `text_cipher.py` — Text Cipher (chiffrer/dĂ©chiffrer une chaĂźne) +- `stego_text.py` — Stego Text (cacher/extraire du texte dans une image, LSB) +- `stego_capacity.py` — Stego Capacity Estimator (capacitĂ© d’une image en LSB) +- `image_cipher.py` — Image ⇄ Noise (image chiffrĂ©e rendue comme “bruit”) +- `jpeg_stego.py` — JPEG Stego (texte en DCT, robuste aux recompressions) +- `stego_image_in_image.py` — Stego Image-in-Image (cacher une image dans une autre, compression avant chiffrement) +- `generate_passphrase.py` — Generate Passphrase (phrases de passe solides) +- `__init__.py` — enregistrement des nƓuds + +--- + +## ⚙ Installation + +1) Copier ce dĂ©pĂŽt dans : + + ComfyUI/custom_nodes/Comfyui_EncryptMaster + +2) Activer votre venv puis installer les dĂ©pendances : + + pip install cryptography pillow numpy + +3) Relancer ComfyUI. Les nƓuds « EncryptMaster — 
 » apparaissent dans la recherche. + +TestĂ© avec Python 3.10–3.12, ComfyUI ≄ 0.3.x, Windows / Linux. + +--- + +## 🔑 SĂ©curitĂ© (rĂ©sumĂ©) + +- **Chiffrement** : AES-256-GCM (authentifiĂ©). +- **DĂ©rivation** : scrypt (N=2^14, r=8, p=1) depuis une passphrase humaine. +- **Associated Data (AAD)** : champ `associated_data` optionnel, non chiffrĂ© mais **authentifiĂ©**. + Si l’AAD ne correspond pas entre chiffrement et dĂ©chiffrement, le dĂ©chiffrement Ă©choue. + +--- + +## đŸ§© NƓuds et usages + +### 1) EncryptMaster — Text Cipher (AES-GCM) + +- **EntrĂ©es** + - `text` (STRING) : texte clair ou blob chiffrĂ©. + - `passphrase` (STRING) : mot/phrase de passe. + - `mode` (STRING) : "encrypt" ou "decrypt". + - `associated_data` (STRING, optionnel). + +- **Sorties** + - `text` (STRING) : texte chiffrĂ© « armored » ou texte dĂ©chiffrĂ©. + +- **Notes** + - Parfait pour sĂ©curiser prompts, JSON ou credentials avant stockage. + +--- + +### 2) EncryptMaster — Stego Embed Text (AES-GCM) / Stego Extract Text + +- **But** : cacher un texte chiffrĂ© dans les **bits de poids faible (LSB)** d’une image. + +- **EntrĂ©es (Embed)** + - `image` (IMAGE), `text` (STRING), `passphrase` (STRING) + - `bits_per_channel` : 1 (discret) ou 2 (×2 capacitĂ©) + - `associated_data` (STRING, optionnel) + +- **Sorties (Embed)** + - `image` (IMAGE) : mĂȘme visuel, message cachĂ©. + +- **EntrĂ©es (Extract)** + - `image` (IMAGE), `passphrase` (STRING), `associated_data` (STRING, optionnel) + +- **Sorties (Extract)** + - `text` (STRING) : texte en clair. + +- **Conseils** + - Sauvegarder en **PNG/TIFF** (sans perte). + - Éviter toute recompression/redimensionnement aprĂšs l’embed. + +--- + +### 3) EncryptMaster — Stego Capacity Estimator + +- **But** : estimer la **capacitĂ© maximale** (octets et nombre de caractĂšres) d’une image pour l’embed LSB. + +- **EntrĂ©es** + - `image` (IMAGE), `bits_per_channel` (1/2) + +- **Sorties** + - `capacity_bytes` (INT), `capacity_chars` (INT), `report` (STRING) + +- **Rappel rapide** + - CapacitĂ© brute ≈ `largeur × hauteur × 3 × bpc / 8` octets. + +--- + +### 4) EncryptMaster — Image Cipher → Noise (AES-GCM) +### EncryptMaster — Image Decipher ← Noise (AES-GCM) + +- **But** : chiffrer **tous les pixels** d’une image et produire une image de **bruit alĂ©atoire** (rĂ©versible). + +- **EntrĂ©es (Cipher)** + - `image` (IMAGE), `passphrase` (STRING) + - `associated_data` (STRING, optionnel) + - `preserve_width` (BOOLEAN) : conserve la largeur; la hauteur s’ajuste si nĂ©cessaire. + +- **Sorties (Cipher)** + - `image` (IMAGE bruitĂ©e), `out_width` (INT), `out_height` (INT) + +- **EntrĂ©es (Decipher)** + - `image` (IMAGE bruitĂ©e), `passphrase` (STRING), `associated_data` (STRING, optionnel) + +- **Sorties (Decipher)** + - `image` (IMAGE originale), `report` (STRING) + +- **Important** + - Toujours **sauver en PNG/TIFF**. Le JPEG dĂ©truit des octets → dĂ©chiffrement impossible. + +--- + +### 5) EncryptMaster — Jpeg Stego Embed Text (DCT/AES-GCM) / Extract + +- **But** : cacher un texte dans les **coefficients DCT** (canal Y) d’un JPEG — plus robuste aux recompressions (FB/IG). + +- **EntrĂ©es (Embed)** + - `image` (IMAGE), `text` (STRING) + - `passphrase` (STRING), `quality` (INT 10–95) + - `associated_data` (STRING, optionnel) + +- **Sorties (Embed)** + - `image` (IMAGE) : JPEG rĂ©encodĂ© Ă  la qualitĂ© choisie. + +- **EntrĂ©es (Extract)** + - `image` (IMAGE), `passphrase` (STRING), `associated_data` (STRING, optionnel) + +- **Sorties (Extract)** + - `text` (STRING) + +- **Limites** + - CapacitĂ© : **quelques kilo-octets**. + - TrĂšs fortes dĂ©gradations (Q ≀ 60), resize massif ou filtres agressifs peuvent dĂ©truire le message. + +--- + +### 6) EncryptMaster — Stego Embed Image (LSB/AES-GCM) / Stego Extract Image + +- **But** : cacher une **image secrĂšte** B dans une **image porteuse** A (LSB), en **compressant** le secret **avant chiffrement**. + +- **EntrĂ©es (Embed)** + - `cover_image` (IMAGE), `secret_image` (IMAGE), `passphrase` (STRING) + - `bits_per_channel` : 1 ou 2 + - `secret_encode` : "jpeg" (recommandĂ©) ou "png" + - `jpeg_quality` (50–95) : qualitĂ© du secret si "jpeg" + - `resize_secret_to_cover` (BOOLEAN) : redimensionner le secret Ă  la cover + - `associated_data` (STRING, optionnel) + +- **Sorties (Embed)** + - `image` (IMAGE), `report` (STRING) + +- **EntrĂ©es (Extract)** + - `image` (IMAGE), `passphrase` (STRING), `associated_data` (STRING, optionnel) + +- **Sorties (Extract)** + - `image` (secret reconstruit), `report` (STRING) + +- **CapacitĂ© (ordre de grandeur)** + - Cover 1216×1600 @ 1 bpc ≈ ~712 KiB ; @ 2 bpc ≈ ~1.42 MiB. + - Secret 1216×1600 en **JPEG q85** ≈ 200–400 Ko → OK avec 2 bpc. + - Mode **PNG** lossless souvent plus volumineux → nĂ©cessitera une cover plus grande ou 2 bpc. + +- **Format de sortie** + - Sauver la cover modifiĂ©e en **PNG/TIFF** (pas de recompression JPEG aprĂšs embed). + +--- + +### 7) EncryptMaster — Generate Passphrase + +- **But** : gĂ©nĂ©rer une phrase de passe robuste (type Diceware). +- **Astuce** : viser ≄ 24 caractĂšres ou 8–9 mots alĂ©atoires. + +--- + +## đŸ§Ș Workflows types + +### A) Cacher un texte chiffrĂ© dans une image (PNG) + + [Text Box] -> [Text Cipher (encrypt)] -> [Stego Embed Text] -> [Save Image (PNG)] + +### B) Partage rĂ©seaux sociaux (rĂ©sistant Ă  la recompression) + + [Text Box] -> [Jpeg Stego Embed Text (quality ~90)] -> [Save Image (JPEG)] + +### C) Cacher une image dans une autre + + [Load Cover] + [Load Secret] -> [Stego Embed Image (secret_encode=jpeg, bpc=2)] -> [Save Image (PNG)] + +### D) Chiffrer une image en « bruit » (rĂ©versible) + + [Load Image] -> [Image Cipher → Noise] -> [Save Image (PNG)] + [Load Noise] -> [Image Decipher ← Noise] -> [Preview] + +--- + +## 📏 Rappels de capacitĂ© (LSB) + +- Formule : `capacitĂ©_octets ≈ largeur × hauteur × 3 × bpc / 8` +- Overhead crypto ≈ 40–60 octets (entĂȘtes + tag GCM) +- Plus `bpc` est Ă©levĂ©, plus la capacitĂ© augmente
 et le risque de dĂ©tection statistique aussi. + +--- + +## ✅ Bonnes pratiques + +- Utiliser **Generate Passphrase** pour des passphrases solides. +- Renseigner un **`associated_data`** stable (ex. `project=EncryptMaster;v=1`) et le rĂ©utiliser Ă  l’extraction. +- Éviter les filtres/redimensionnements aprĂšs stĂ©ganographie LSB. +- Pour diffusion sur rĂ©seaux sociaux, prĂ©fĂ©rer le **mode JPEG/DCT**. + +--- + +## 🐞 DĂ©pannage + +- **Capacity too small** + Augmenter `bits_per_channel`, choisir `secret_encode="jpeg"` (q 80–85), rĂ©duire le secret, ou utiliser une cover plus grande. + +- **Decrypt failed / wrong tag** + Passphrase ou `associated_data` incorrects, ou image altĂ©rĂ©e (recompression/resize pour LSB). + +- **Image « bruit » illisible** + Sauvegarde en JPEG → rĂ©essayer en PNG/TIFF. + +--- + +## 📜 Licence + +Ce projet est sous licence **MIT**. + +--- diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..a32f386 --- /dev/null +++ b/__init__.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +# ComfyUI_EncryptMaster / __init__.py + +NODE_ID = "EncryptMaster" + +from .text_cipher import TextCipher +from .stego_text import Stego_EmbedText as StegoEmbedText, Stego_ExtractText as StegoExtractText +from .generate_passphrase import GeneratePassphrase +from .stego_capacity import Stego_CapacityEstimator +from .image_cipher import Image_CipherToNoise, Image_DecipherFromNoise +from .jpeg_stego import Jpeg_Stego_EmbedText, Jpeg_Stego_ExtractText +from .stego_image_in_image import Stego_EmbedImage, Stego_ExtractImage + +# Dictionnaires de mapping +NODE_CLASS_MAPPINGS = { + "Text Cipher": TextCipher, + "Stego Embed Text": StegoEmbedText, + "Stego Extract Text": StegoExtractText, + "Generate Passphrase": GeneratePassphrase, + "Stego Capacity Estimator": Stego_CapacityEstimator, + "Image Cipher To Noise": Image_CipherToNoise, + "Image Decipher From Noise": Image_DecipherFromNoise, + "Jpeg Stego Embed Text": Jpeg_Stego_EmbedText, + "Jpeg Stego Extract Text": Jpeg_Stego_ExtractText, + "Stego Embed Image": Stego_EmbedImage, + "Stego Extract Image": Stego_ExtractImage, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Text Cipher": "EncryptMaster — Text Cipher (AES-GCM)", + "Stego Embed Text": "EncryptMaster — Stego Embed Text (AES-GCM)", + "Stego Extract Text": "EncryptMaster — Stego Extract Text (AES-GCM)", + "Generate Passphrase": "EncryptMaster — Generate Passphrase", + "Stego Capacity Estimator": "EncryptMaster — Stego Capacity Estimator", + "Image Cipher To Noise": "EncryptMaster — Image Cipher → Noise (AES-GCM)", + "Image Decipher From Noise": "EncryptMaster — Image Decipher ← Noise (AES-GCM)", + "Jpeg Stego Embed Text": "EncryptMaster — Jpeg Stego Embed Text (DCT/AES-GCM)", + "Jpeg Stego Extract Text": "EncryptMaster — Jpeg Stego Extract Text (DCT/AES-GCM)", + "Stego Embed Image": "EncryptMaster — Stego Embed Image (LSB/AES-GCM)", + "Stego Extract Image": "EncryptMaster — Stego Extract Image (LSB/AES-GCM)", +} + +WEB_DIRECTORY = "./web" + +__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"] + +print(f"### Loading: {NODE_ID}") +print(f" - Mapped {len(NODE_CLASS_MAPPINGS)} nodes") diff --git a/generate_passphrase.py b/generate_passphrase.py new file mode 100644 index 0000000..546b681 --- /dev/null +++ b/generate_passphrase.py @@ -0,0 +1,208 @@ +# generate_passphrase.py +# Generate Passphrase +# Deux modes: random_chars (cryptographique) et diceware (wordlist externe) +# +# EntrĂ©es (random_chars): +# length:int, use_lowercase:bool, use_uppercase:bool, use_digits:bool, use_symbols:bool, +# exclude_ambiguous:bool, extra_digits_suffix:int (0..6), deterministic_seed:int (=-1 pour off) +# +# EntrĂ©es (diceware): +# diceware_wordlist_path:str, num_words:int, separator:str, extra_digits_suffix:int, +# deterministic_seed:int (=-1 pour off) +# +# Sorties: +# passphrase: STRING +# entropy_bits: FLOAT +# recipe: STRING +# +# Notes: +# - RNG par defaut: secrets (cryptographiquement sĂ»r) +# - Si deterministic_seed >= 0: bascule sur random.Random(seed) (DEV UNIQUEMENT) + +import os +import math +import string +import random +from typing import List, Tuple + +try: + import secrets + _HAS_SECRETS = True +except Exception: + _HAS_SECRETS = False + +CATEGORY = "EncryptMaster" + +AMBIGUOUS = set("Il1O0oB8S5Z2") +SYMBOLS_DEFAULT = "!@#$%^&*()_-+=[]{};:,.?/\\|~`" + +def _safe_choice(seq, rng_secrets: bool, rnd: random.Random | None): + if rng_secrets and _HAS_SECRETS: + return secrets.choice(seq) + return rnd.choice(seq) # type: ignore + +def _safe_randint(a: int, b: int, rng_secrets: bool, rnd: random.Random | None) -> int: + if rng_secrets and _HAS_SECRETS: + # secrets.randbelow pour bornes inclusives + return a + secrets.randbelow(b - a + 1) + return rnd.randint(a, b) # type: ignore + +def _build_charset(use_lowercase: bool, use_uppercase: bool, use_digits: bool, use_symbols: bool, exclude_ambiguous: bool) -> List[str]: + pool = "" + if use_lowercase: + pool += string.ascii_lowercase + if use_uppercase: + pool += string.ascii_uppercase + if use_digits: + pool += string.digits + if use_symbols: + pool += SYMBOLS_DEFAULT + + if exclude_ambiguous: + pool = "".join([c for c in pool if c not in AMBIGUOUS]) + + # sĂ©curitĂ©: Ă©viter pool vide + if not pool: + # minimal fallback : minuscules + pool = string.ascii_lowercase + return list(pool) + +def _entropy_random_chars(length: int, alphabet_size: int, extra_digits_suffix: int) -> float: + # H = length*log2(S) + extra_digits_suffix*log2(10) + return length * math.log2(max(2, alphabet_size)) + extra_digits_suffix * math.log2(10) + +def _entropy_diceware(num_words: int, vocab_size: int, extra_digits_suffix: int) -> float: + # H = num_words*log2(vocab_size) + extra_digits_suffix*log2(10) + return num_words * math.log2(max(2, vocab_size)) + extra_digits_suffix * math.log2(10) + +def _load_wordlist(path: str) -> List[str]: + if not path or not os.path.exists(path): + raise FileNotFoundError(f"Diceware wordlist not found: {path}") + words = [] + with open(path, "r", encoding="utf-8", errors="ignore") as f: + for line in f: + w = line.strip() + if w: + words.append(w) + # Ă©viter doublons vides + words = [w for w in words if w] + if len(words) < 1000: + # 1000+ recommandĂ© pour une entropie correcte + raise ValueError(f"Wordlist too small ({len(words)} entries). Provide a larger list (e.g., 2048–7776+).") + return words + +def _gen_random_chars(length: int, charset: List[str], extra_digits_suffix: int, rng_secrets: bool, rnd: random.Random | None) -> str: + # assure au moins 1 caractĂšre de chaque classe activĂ©e ? Non: on laisse alĂ©atoire pur (entropie max) + out = "".join(_safe_choice(charset, rng_secrets, rnd) for _ in range(max(1, length))) + if extra_digits_suffix > 0: + digits = "".join(str(_safe_randint(0, 9, rng_secrets, rnd)) for _ in range(extra_digits_suffix)) + out = out + digits + return out + +def _gen_diceware(num_words: int, words: List[str], separator: str, extra_digits_suffix: int, rng_secrets: bool, rnd: random.Random | None) -> str: + picks = [] + for _ in range(max(1, num_words)): + if rng_secrets and _HAS_SECRETS: + idx = secrets.randbelow(len(words)) + else: + idx = rnd.randrange(len(words)) # type: ignore + picks.append(words[idx]) + out = separator.join(picks) + if extra_digits_suffix > 0: + digits = "".join(str(_safe_randint(0, 9, rng_secrets, rnd)) for _ in range(extra_digits_suffix)) + out = out + digits + return out + +class GeneratePassphrase: + """ + GĂ©nĂ©rateur de passphrase sĂ©curisĂ©. + - random_chars : jeu de caractĂšres configurable + - diceware : mots alĂ©atoires depuis un fichier wordlist (un mot par ligne) + """ + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "mode": (["random_chars", "diceware"], {"default": "random_chars"}), + }, + "optional": { + # random_chars + "length": ("INT", {"default": 20, "min": 4, "max": 256}), + "use_lowercase": ("BOOLEAN", {"default": True}), + "use_uppercase": ("BOOLEAN", {"default": True}), + "use_digits": ("BOOLEAN", {"default": True}), + "use_symbols": ("BOOLEAN", {"default": True}), + "exclude_ambiguous": ("BOOLEAN", {"default": True}), + # diceware + "diceware_wordlist_path": ("STRING", {"default": "", "multiline": False}), + "num_words": ("INT", {"default": 8, "min": 2, "max": 32}), + "separator": ("STRING", {"default": "-", "multiline": False}), + # commun + "extra_digits_suffix": ("INT", {"default": 0, "min": 0, "max": 10}), + "deterministic_seed": ("INT", {"default": -1, "min": -1, "max": 2**31-1}), + } + } + + RETURN_TYPES = ("STRING", "FLOAT", "STRING") + RETURN_NAMES = ("passphrase", "entropy_bits", "recipe") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run( + self, + mode="random_chars", + length=20, + use_lowercase=True, + use_uppercase=True, + use_digits=True, + use_symbols=True, + exclude_ambiguous=True, + diceware_wordlist_path="", + num_words=8, + separator="-", + extra_digits_suffix=0, + deterministic_seed=-1, + ): + # SĂ©lection RNG + rng_secrets = deterministic_seed < 0 + rnd = None + if not rng_secrets: + rnd = random.Random(deterministic_seed) + + if mode == "diceware": + # Charge wordlist + try: + words = _load_wordlist(diceware_wordlist_path) + except Exception as e: + return (f"[GeneratePassphrase ERROR] {type(e).__name__}: {e}", 0.0, "load_wordlist_failed") + + passphrase = _gen_diceware(num_words, words, separator, extra_digits_suffix, rng_secrets, rnd) + entropy = _entropy_diceware(num_words, len(words), extra_digits_suffix) + recipe = f"mode=diceware; words={num_words}; vocab={len(words)}; sep={repr(separator)}; digits+={extra_digits_suffix}; rng={'secrets' if rng_secrets else 'deterministic'}" + if not rng_secrets: + recipe += " [WARNING: deterministic RNG — dev/test only]" + return (passphrase, float(entropy), recipe) + + # random_chars + charset = _build_charset(use_lowercase, use_uppercase, use_digits, use_symbols, exclude_ambiguous) + passphrase = _gen_random_chars(length, charset, extra_digits_suffix, rng_secrets, rnd) + entropy = _entropy_random_chars(length, len(charset), extra_digits_suffix) + recipe = ( + f"mode=random_chars; length={length}; alphabet={len(charset)};" + f" lower={use_lowercase}; upper={use_uppercase}; digits={use_digits}; symbols={use_symbols};" + f" exclude_ambiguous={exclude_ambiguous}; digits+={extra_digits_suffix};" + f" rng={'secrets' if rng_secrets else 'deterministic'}" + ) + if not rng_secrets: + recipe += " [WARNING: deterministic RNG — dev/test only]" + return (passphrase, float(entropy), recipe) + + +NODE_CLASS_MAPPINGS = { + "GeneratePassphrase": GeneratePassphrase, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "GeneratePassphrase": "Generate Passphrase", +} diff --git a/image_cipher.py b/image_cipher.py new file mode 100644 index 0000000..826d054 --- /dev/null +++ b/image_cipher.py @@ -0,0 +1,222 @@ +# image_cipher.py +# EncryptMaster — Image <-> Noise (AES-256-GCM + scrypt) +# - Image_CipherToNoise : chiffre une image en "bruit" (PNG/TIFF conseillĂ©) +# - Image_DecipherFromNoise : reconstruit l'image originale depuis le "bruit" +# +# Remarques: +# - La sortie peut ĂȘtre +haute que l'entrĂ©e (quelques lignes) pour stocker l'overhead (sel+nonce+header+tag). +# - Sauvegarder en PNG/TIFF (sans perte). Le JPEG casserait les octets. + +import os +import math +import struct +from typing import Tuple + +import numpy as np +from PIL import Image +import torch + +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +CATEGORY = "EncryptMaster" + +# ======== Crypto helpers ======== +def _derive_key(passphrase: str, salt: bytes, length: int = 32) -> bytes: + if not passphrase: + raise ValueError("passphrase cannot be empty") + kdf = Scrypt(salt=salt, length=length, n=2**14, r=8, p=1) + return kdf.derive(passphrase.encode("utf-8")) + +def _encrypt(plaintext: bytes, passphrase: str, aad: str = "") -> Tuple[bytes, bytes, bytes]: + salt = os.urandom(16) + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + nonce = os.urandom(12) + ad = aad.encode("utf-8") if aad else None + cipher = aes.encrypt(nonce, plaintext, ad) # = plaintext + 16-byte tag + return salt, nonce, cipher + +def _decrypt(salt: bytes, nonce: bytes, cipher: bytes, passphrase: str, aad: str = "") -> bytes: + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + ad = aad.encode("utf-8") if aad else None + return aes.decrypt(nonce, cipher, ad) + +# ======== Tensor <-> PIL helpers ======== +def tensor2rgb_u8(img_tensor: torch.Tensor) -> np.ndarray: + """ + torch.Tensor [B,H,W,C] float32 in [0,1] -> np.uint8 [H,W,3] + """ + t = img_tensor[0].detach().cpu().numpy() # [H,W,C] + t = np.clip(t, 0.0, 1.0) + arr = (t * 255.0).round().astype(np.uint8) + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + if arr.shape[2] < 3: + arr = np.tile(arr[:, :, :1], (1, 1, 3)) + if arr.shape[2] > 3: + arr = arr[:, :, :3] + return arr + +def rgb_u8_to_tensor(arr: np.ndarray): + """ + np.uint8 [H,W,3] -> (torch.Tensor [B,H,W,C] float32 in [0,1],) + """ + arr = arr.astype(np.float32) / 255.0 + t = torch.from_numpy(arr).unsqueeze(0) + return (t,) + +# ======== Payload header ======== +# MAGIC(6) | VER(1) | H(4) | W(4) | C(1) | CIPHER_LEN(8) | RSRV(4) +# = 6 + 1 + 4 + 4 + 1 + 8 + 4 = 28 bytes +MAGIC = b"EMIMG1" +HEADER_FMT = ">6sBIIBQ I" # Espace ignorĂ© par struct; Ă©quiv. ">6sBII BQI" +# Pour Ă©viter toute ambiguĂŻtĂ©, on calcule la taille explicitement: +HEADER_SIZE = 6 + 1 + 4 + 4 + 1 + 8 + 4 # 28 + +def _build_header(h: int, w: int, c: int, cipher_len: int, ver: int = 1) -> bytes: + # RSRV = 0 (rĂ©servĂ© pour futures versions) + # Comme struct ne gĂšre pas bien les espaces dans le fmt, pack manuellement: + return ( + MAGIC + + bytes([ver & 0xFF]) + + struct.pack(">I", h) + + struct.pack(">I", w) + + bytes([c & 0xFF]) + + struct.pack(">Q", cipher_len) + + struct.pack(">I", 0) + ) + +def _parse_header(buf: bytes): + if len(buf) < HEADER_SIZE: + raise ValueError("Header too small") + if buf[:6] != MAGIC: + raise ValueError("Invalid magic") + ver = buf[6] + h = struct.unpack(">I", buf[7:11])[0] + w = struct.unpack(">I", buf[11:15])[0] + c = buf[15] + cipher_len = struct.unpack(">Q", buf[16:24])[0] + # rsrv = struct.unpack(">I", buf[24:28])[0] + return ver, h, w, c, cipher_len + +# ======== Nodes ======== +class Image_CipherToNoise: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + "preserve_width": ("BOOLEAN", {"default": True}), + }, + } + + RETURN_TYPES = ("IMAGE", "INT", "INT") + RETURN_NAMES = ("image", "out_width", "out_height") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, passphrase, associated_data="", preserve_width=True): + # 1) Image -> bytes + rgb = tensor2rgb_u8(image) # [H,W,3] + h, w, c = rgb.shape + plain = rgb.tobytes(order="C") # len = h*w*3 + + # 2) Encrypt + salt, nonce, cipher = _encrypt(plain, passphrase, associated_data) + + # 3) Build payload: header + salt + nonce + cipher + header = _build_header(h, w, 3, len(cipher), ver=1) + payload = header + salt + nonce + cipher + total = len(payload) + + # 4) Decide output size (capacity = H*W*3 bytes). We enlarge height if needed. + if preserve_width and w > 0: + out_w = w + pixels = math.ceil(total / 3) + out_h = math.ceil(pixels / out_w) + else: + # make it near-square if free width + pixels = math.ceil(total / 3) + side = int(math.sqrt(pixels)) + out_w = max(1, side) + out_h = math.ceil(pixels / out_w) + + # 5) Fill noise buffer + out = np.frombuffer(os.urandom(out_w * out_h * 3), dtype=np.uint8).copy() + out[:total] = np.frombuffer(payload, dtype=np.uint8) + out = out.reshape(out_h, out_w, 3) + + return rgb_u8_to_tensor(out) + (int(out_w), int(out_h)) + + +class Image_DecipherFromNoise: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + }, + } + + RETURN_TYPES = ("IMAGE", "STRING") + RETURN_NAMES = ("image", "report") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, passphrase, associated_data=""): + # 1) Read raw bytes from noise image + arr = tensor2rgb_u8(image) # [H,W,3] + raw = arr.reshape(-1).tobytes(order="C") + + # 2) Parse header + if len(raw) < HEADER_SIZE + 16 + 12: + return (rgb_u8_to_tensor(np.zeros((1,1,3), dtype=np.uint8))[0], + "[ImageCipher ERROR] Not enough data for header") + ver, orig_h, orig_w, orig_c, cipher_len = _parse_header(raw[:HEADER_SIZE]) + + # 3) Read salt, nonce, cipher + ofs = HEADER_SIZE + salt = raw[ofs:ofs+16]; ofs += 16 + nonce = raw[ofs:ofs+12]; ofs += 12 + end = ofs + cipher_len + if end > len(raw): + return (rgb_u8_to_tensor(np.zeros((1,1,3), dtype=np.uint8))[0], + "[ImageCipher ERROR] Truncated payload") + cipher = raw[ofs:end] + + # 4) Decrypt + try: + plain = _decrypt(salt, nonce, cipher, passphrase, associated_data) + except Exception as e: + return (rgb_u8_to_tensor(np.zeros((1,1,3), dtype=np.uint8))[0], + f"[ImageCipher ERROR] {type(e).__name__}: {e}") + + # 5) Rebuild original image + needed = orig_h * orig_w * orig_c + if len(plain) != needed or orig_c != 3: + return (rgb_u8_to_tensor(np.zeros((1,1,3), dtype=np.uint8))[0], + "[ImageCipher ERROR] Invalid plaintext size/channels") + rgb = np.frombuffer(plain, dtype=np.uint8).reshape(orig_h, orig_w, 3) + return rgb_u8_to_tensor(rgb)[0], f"Recovered {orig_w}x{orig_h} RGB image (v{ver})." + + +# ======== ComfyUI mappings ======== +NODE_CLASS_MAPPINGS = { + "Image Cipher To Noise": Image_CipherToNoise, + "Image Decipher From Noise": Image_DecipherFromNoise, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Image Cipher To Noise": "EncryptMaster — Image Cipher → Noise (AES-GCM)", + "Image Decipher From Noise": "EncryptMaster — Image Decipher ← Noise (AES-GCM)", +} diff --git a/jpeg_stego.py b/jpeg_stego.py new file mode 100644 index 0000000..b1e21a7 --- /dev/null +++ b/jpeg_stego.py @@ -0,0 +1,380 @@ +# jpeg_stego.py +# EncryptMaster — JPEG Stego (DCT, mid-frequencies, AES-GCM) +# Nodes: +# - Jpeg_Stego_EmbedText : IMAGE, text, passphrase, quality -> IMAGE +# - Jpeg_Stego_ExtractText : IMAGE, passphrase -> STRING +# +# NOTE (prototype robuste FB): +# - Encode dans les AC "mid-frequency" du canal Y aprĂšs quantification. +# - Plus rĂ©sistant Ă  une recompression (Q~90->80) qu'un LSB spatial. +# - CapacitĂ© modĂ©rĂ©e (quelques Ko). +# +# Limites: +# - Recompresssions trĂšs agressives (Q<=60), gros redimensionnements/filters -> perte possible. + +import io +import os +import math +import struct +from typing import Tuple + +import numpy as np +from PIL import Image +import torch + +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +CATEGORY = "EncryptMaster" + +# ---------- Crypto helpers (mĂȘmes fondations que le pack) ---------- +def _derive_key(passphrase: str, salt: bytes, length: int = 32) -> bytes: + if not passphrase: + raise ValueError("passphrase cannot be empty") + kdf = Scrypt(salt=salt, length=length, n=2**14, r=8, p=1) + return kdf.derive(passphrase.encode("utf-8")) + +def encrypt_bytes(plaintext: bytes, passphrase: str, associated_data: str = "") -> Tuple[bytes, bytes, bytes]: + salt = os.urandom(16) + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + nonce = os.urandom(12) + aad = associated_data.encode("utf-8") if associated_data else None + cipher = aes.encrypt(nonce, plaintext, aad) + return salt, nonce, cipher + +def decrypt_bytes(salt: bytes, nonce: bytes, cipher: bytes, passphrase: str, associated_data: str = "") -> bytes: + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + aad = associated_data.encode("utf-8") if associated_data else None + return aes.decrypt(nonce, cipher, aad) + +# ---------- Tensor <-> PIL ---------- +def tensor2pil(img_tensor: torch.Tensor) -> Image.Image: + t = img_tensor[0].detach().cpu() + arr = t.numpy() + arr = np.clip(arr, 0.0, 1.0) + arr = (arr * 255.0).round().astype(np.uint8) + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + if arr.ndim == 3 and arr.shape[2] > 3: + arr = arr[:, :, :3] + return Image.fromarray(arr, mode="RGB") + +def pil2tensor(img: Image.Image): + if img.mode != "RGB": + img = img.convert("RGB") + arr = np.array(img, dtype=np.float32) / 255.0 + t = torch.from_numpy(arr).unsqueeze(0) # [1,H,W,C] + return (t,) + +# ---------- JPEG DCT helpers (8x8, canal Y) ---------- +# Matrice DCT 8x8 (orthonormĂ©e) +def _dct_matrix(n=8): + A = np.zeros((n, n), dtype=np.float64) + for u in range(n): + for x in range(n): + alpha = math.sqrt(1.0/n) if u == 0 else math.sqrt(2.0/n) + A[u, x] = alpha * math.cos(((2*x + 1) * u * math.pi) / (2 * n)) + return A + +A8 = _dct_matrix(8) +A8T = A8.T + +# Table luminance standard (base JPEG, sera "scalĂ©e" selon quality) +_QY_STD = np.array([ + [16,11,10,16,24,40,51,61], + [12,12,14,19,26,58,60,55], + [14,13,16,24,40,57,69,56], + [14,17,22,29,51,87,80,62], + [18,22,37,56,68,109,103,77], + [24,35,55,64,81,104,113,92], + [49,64,78,87,103,121,120,101], + [72,92,95,98,112,100,103,99], +], dtype=np.float64) + +# QualitĂ© JPEG -> scale quantization (approximation standard) +def _scale_qtable(q: int) -> np.ndarray: + q = max(10, min(95, int(q))) + if q < 50: + s = 5000 / q + else: + s = 200 - 2 * q + Q = np.floor((_QY_STD * s + 50) / 100) + Q[Q < 1] = 1 + Q[Q > 255] = 255 + return Q + +# Zigzag order pour ordonner (u,v) +_ZZ = np.array([ + (0,0),(0,1),(1,0),(2,0),(1,1),(0,2),(0,3),(1,2), + (2,1),(3,0),(4,0),(3,1),(2,2),(1,3),(0,4),(0,5), + (1,4),(2,3),(3,2),(4,1),(5,0),(6,0),(5,1),(4,2), + (3,3),(2,4),(1,5),(0,6),(0,7),(1,6),(2,5),(3,4), + (4,3),(5,2),(6,1),(7,0),(7,1),(6,2),(5,3),(4,4), + (3,5),(2,6),(1,7),(2,7),(3,6),(4,5),(5,4),(6,3), + (7,2),(7,3),(6,4),(5,5),(4,6),(3,7),(4,7),(5,6), + (6,5),(7,4),(7,5),(6,6),(5,7),(6,7),(7,6),(7,7) +], dtype=np.int32) + +# Indices AC "mid-frequency" (on Ă©vite DC et hautes frĂ©quences trop fragiles) +_MID_AC_IDX = [i for i in range(10, 40)] # empirique, fonctionne bien + +def _blockify(arr: np.ndarray, bs=8): + h, w = arr.shape + H = (h + bs - 1) // bs * bs + W = (w + bs - 1) // bs * bs + pad = np.zeros((H, W), dtype=arr.dtype) + pad[:h, :w] = arr + blocks = (pad.reshape(H//bs, bs, W//bs, bs) + .transpose(0, 2, 1, 3)) # [nH, nW, 8, 8] + return blocks, h, w + +def _unblockify(blocks: np.ndarray, orig_h: int, orig_w: int): + nH, nW, bs, _ = blocks.shape + arr = blocks.transpose(0, 2, 1, 3).reshape(nH*bs, nW*bs) + return arr[:orig_h, :orig_w] + +def _dct2(block): + return A8 @ block @ A8T + +def _idct2(coeff): + return A8T @ coeff @ A8 + +def _rgb_to_y(image: Image.Image) -> np.ndarray: + y, _, _ = image.convert("YCbCr").split() + return np.array(y, dtype=np.float64) + +def _y_to_rgb(y: np.ndarray, base_rgb: Image.Image) -> Image.Image: + # on remplace le Y de base + y_img = Image.fromarray(np.clip(y, 0, 255).astype(np.uint8), mode="L") + Y, Cb, Cr = base_rgb.convert("YCbCr").split() + out = Image.merge("YCbCr", (y_img, Cb, Cr)).convert("RGB") + return out + +def _quantize(c: np.ndarray, Q: np.ndarray) -> np.ndarray: + return np.round(c / Q).astype(np.int32) + +def _dequantize(qc: np.ndarray, Q: np.ndarray) -> np.ndarray: + return (qc.astype(np.float64) * Q) + +def _bytes_to_bits(data: bytes) -> np.ndarray: + arr = np.frombuffer(data, dtype=np.uint8) + return np.unpackbits(arr) + +def _bits_to_bytes(bits: np.ndarray) -> bytes: + if bits.size % 8 != 0: + bits = np.pad(bits, (0, 8 - (bits.size % 8)), constant_values=0) + return np.packbits(bits).tobytes() + +# Header “armored” pour le stego payload +MAGIC = b"EMJPG1" # 6 bytes +HEADER_FMT = ">6sB16s12sI" # MAGIC(6) | QUAL(1) | SALT(16) | NONCE(12) | LEN(4) +HEADER_SIZE = struct.calcsize(HEADER_FMT) # 39 bytes + +def _build_header(quality: int, salt: bytes, nonce: bytes, payload_len: int) -> bytes: + q = max(10, min(95, int(quality))) + return struct.pack(HEADER_FMT, MAGIC, q, salt, nonce, payload_len) + +def _parse_header(blob: bytes): + magic, qual, salt, nonce, payload_len = struct.unpack(HEADER_FMT, blob[:HEADER_SIZE]) + if magic != MAGIC: + raise ValueError("Invalid JPEG stego header") + return qual, salt, nonce, payload_len + +# ---------- Embedding / Extraction dans les AC ---------- +def _embed_bits_in_qcoeff(qblocks: np.ndarray, bits: np.ndarray) -> int: + """ + qblocks: [nH, nW, 8, 8] int32 (quantized DCT for Y) + bits : flat {0,1} + Retourne: nombre de bits Ă©crits + """ + nH, nW, _, _ = qblocks.shape + bit_idx = 0 + for i in range(nH): + for j in range(nW): + block = qblocks[i, j] + # zigzag order + for k in _MID_AC_IDX: + if bit_idx >= bits.size: + return bit_idx + u, v = _ZZ[k] + val = block[u, v] + if val == 0: + continue + # on Ă©vite |val| == 1 (fragile) + if abs(val) == 1: + continue + b = bits[bit_idx] + # impose paritĂ© sur |val| + if (abs(val) & 1) != int(b): + # ajuste d'une unitĂ© vers +/- + if val > 0: + val -= 1 + else: + val += 1 + block[u, v] = val + bit_idx += 1 + return bit_idx + +def _extract_bits_from_qcoeff(qblocks: np.ndarray, nbits: int) -> np.ndarray: + nH, nW, _, _ = qblocks.shape + out = np.zeros(nbits, dtype=np.uint8) + bit_idx = 0 + for i in range(nH): + for j in range(nW): + block = qblocks[i, j] + for k in _MID_AC_IDX: + if bit_idx >= nbits: + return out + u, v = _ZZ[k] + val = block[u, v] + if val == 0 or abs(val) == 1: + continue + out[bit_idx] = (abs(val) & 1) + bit_idx += 1 + if bit_idx >= nbits: + return out + return out + +# ---------- ComfyUI Nodes ---------- +class Jpeg_Stego_EmbedText: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "text": ("STRING", {"multiline": True, "default": ""}), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + "quality": ("INT", {"default": 90, "min": 10, "max": 95}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + } + } + + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, text, passphrase, quality, associated_data=""): + # 1) PIL + pil = tensor2pil(image) + if pil.mode != "RGB": + pil = pil.convert("RGB") + # 2) Y channel + Y = _rgb_to_y(pil).astype(np.float64) - 128.0 # centrĂ© + blocks, h, w = _blockify(Y, 8) + # 3) DCT + quantization + Q = _scale_qtable(quality) + nH, nW, _, _ = blocks.shape + dct_blocks = np.zeros_like(blocks, dtype=np.float64) + for i in range(nH): + for j in range(nW): + dct_blocks[i, j] = _dct2(blocks[i, j]) + qblocks = _quantize(dct_blocks, Q) + + # 4) payload chiffrĂ© + salt, nonce, cipher = encrypt_bytes(text.encode("utf-8"), passphrase, associated_data) + header = _build_header(quality, salt, nonce, len(cipher)) + payload = header + cipher + bits = _bytes_to_bits(payload) + + # 5) embed bits -> qblocks + written = _embed_bits_in_qcoeff(qblocks, bits) + if written < bits.size: + need = bits.size - written + return (image,) # capacitĂ© insuffisante (on pourrait renvoyer un message d'erreur) + + # 6) dequantize + IDCT + rec_dct = _dequantize(qblocks, Q) + rec_blocks = np.zeros_like(rec_dct, dtype=np.float64) + for i in range(nH): + for j in range(nW): + rec_blocks[i, j] = _idct2(rec_dct[i, j]) + Yrec = _unblockify(rec_blocks, h, w) + 128.0 + + # 7) remixer en RGB et rĂ©-encoder JPEG (qualitĂ© choisie) + out_rgb = _y_to_rgb(Yrec, pil) + # on encode en JPEG (qualitĂ©), puis on relit -> sortie IMAGE + buf = io.BytesIO() + out_rgb.save(buf, format="JPEG", quality=quality, subsampling=1, optimize=False) + buf.seek(0) + out_jpeg = Image.open(buf).convert("RGB") + return pil2tensor(out_jpeg) + +class Jpeg_Stego_ExtractText: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("text",) + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, passphrase, associated_data=""): + pil = tensor2pil(image) + if pil.mode != "RGB": + pil = pil.convert("RGB") + Y = _rgb_to_y(pil).astype(np.float64) - 128.0 + blocks, h, w = _blockify(Y, 8) + nH, nW, _, _ = blocks.shape + + # Heuristique: essayer qualitĂ©s communes pour rĂ©cupĂ©rer l'entĂȘte + for try_q in (90, 85, 80, 75): + Q = _scale_qtable(try_q) + dct_blocks = np.zeros_like(blocks, dtype=np.float64) + for i in range(nH): + for j in range(nW): + dct_blocks[i, j] = _dct2(blocks[i, j]) + qblocks = _quantize(dct_blocks, Q) + + # Extraire juste l'entĂȘte (39 bytes) + header_bits = _extract_bits_from_qcoeff(qblocks, HEADER_SIZE * 8) + header_bytes = _bits_to_bytes(header_bits)[:HEADER_SIZE] + try: + qual, salt, nonce, payload_len = _parse_header(header_bytes) + except Exception: + continue # essaie qualitĂ© suivante + + # Maintenant extraire header + payload + total_bits = (HEADER_SIZE + payload_len) * 8 + bits = _extract_bits_from_qcoeff(qblocks, total_bits) + blob = _bits_to_bytes(bits)[:HEADER_SIZE + payload_len] + try: + _, q2, salt2, nonce2, plen2 = struct.unpack(HEADER_FMT, blob[:HEADER_SIZE]) + if q2 != qual or salt2 != salt or nonce2 != nonce or plen2 != payload_len: + continue + except Exception: + continue + + cipher = blob[HEADER_SIZE:HEADER_SIZE + payload_len] + try: + pt = decrypt_bytes(salt, nonce, cipher, passphrase, associated_data) + return (pt.decode("utf-8"),) + except Exception: + # mauvaise passphrase / mauvais AAD / mauvaise qualitĂ© + continue + + return ("[Jpeg Stego ERROR] Unable to recover payload (quality/transform too strong or wrong passphrase/AAD).",) + + +NODE_CLASS_MAPPINGS = { + "Jpeg Stego Embed Text": Jpeg_Stego_EmbedText, + "Jpeg Stego Extract Text": Jpeg_Stego_ExtractText, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Jpeg Stego Embed Text": "EncryptMaster — Jpeg Stego Embed Text (DCT/AES-GCM)", + "Jpeg Stego Extract Text": "EncryptMaster — Jpeg Stego Extract Text (DCT/AES-GCM)", +} diff --git a/stego_capacity.py b/stego_capacity.py new file mode 100644 index 0000000..e1c25dd --- /dev/null +++ b/stego_capacity.py @@ -0,0 +1,59 @@ +# stego_capacity.py +# EncryptMaster — Stego Capacity Estimator +# Calcule combien d'octets et de caractĂšres max on peut cacher dans une image + +import numpy as np +import torch + +CATEGORY = "EncryptMaster" + +def _capacity_bytes(h: int, w: int, bpc: int) -> int: + cap_bits = h * w * 3 * bpc + return cap_bits // 8 + +class Stego_CapacityEstimator: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "bits_per_channel": ([1, 2], {"default": 1}), + } + } + + RETURN_TYPES = ("INT", "INT", "STRING") + RETURN_NAMES = ("capacity_bytes", "capacity_chars", "report") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, bits_per_channel): + # tensor -> shape + t = image[0].detach().cpu().numpy() # [H,W,C], float32 [0,1] + h, w, _ = t.shape + capacity = _capacity_bytes(h, w, bits_per_channel) + + # rĂ©serve 40 octets pour l'entĂȘte (salt, nonce, taille
) + usable_bytes = max(0, capacity - 40) + + # estimation caractĂšres = octets + usable_chars = usable_bytes + + report = ( + f"Image: {w}x{h}px\n" + f"Bits per channel: {bits_per_channel}\n" + f"CapacitĂ© totale: {capacity} octets (≈ {capacity/1024:.2f} KiB)\n" + f"RĂ©servĂ© entĂȘte: 40 octets\n" + f"Utilisable: {usable_bytes} octets ≈ {usable_chars} caractĂšres (UTF-8 simple)\n" + f"⚠ Attention: les caractĂšres spĂ©ciaux/emoji peuvent prendre plus d’octets." + ) + + return int(capacity), int(usable_chars), report + + +NODE_CLASS_MAPPINGS = { + "Stego Capacity Estimator": Stego_CapacityEstimator, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Stego Capacity Estimator": "EncryptMaster — Stego Capacity Estimator", +} diff --git a/stego_image_in_image.py b/stego_image_in_image.py new file mode 100644 index 0000000..255155c --- /dev/null +++ b/stego_image_in_image.py @@ -0,0 +1,350 @@ +# stego_image_in_image.py +# EncryptMaster — Stego Image-in-Image (AES-256-GCM + scrypt + LSB) +# Cache une image "secrĂšte" dans une image "porteuse" (PNG/TIFF recommandĂ©). +# AmĂ©liorations : +# - Quand resize_secret_to_cover=True, on chiffre la version redimensionnĂ©e (fix). +# - secret_encode: "png" (defaut) ou "jpeg" (avec jpeg_quality) -> compression AVANT chiffrement. +# - Header inclut le "enc_type" (0=raw, 1=png, 2=jpeg). + +import io +import os +import struct +from typing import Tuple + +import numpy as np +from PIL import Image +import torch +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +CATEGORY = "EncryptMaster" + +# ========= Crypto ========= +def _derive_key(passphrase: str, salt: bytes, length: int = 32) -> bytes: + if not passphrase: + raise ValueError("passphrase cannot be empty") + kdf = Scrypt(salt=salt, length=length, n=2**14, r=8, p=1) + return kdf.derive(passphrase.encode("utf-8")) + +def encrypt_bytes(plaintext: bytes, passphrase: str, associated_data: str = "") -> Tuple[bytes, bytes, bytes]: + salt = os.urandom(16) + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + nonce = os.urandom(12) + aad = associated_data.encode("utf-8") if associated_data else None + cipher = aes.encrypt(nonce, plaintext, aad) + return salt, nonce, cipher + +def decrypt_bytes(salt: bytes, nonce: bytes, cipher: bytes, passphrase: str, associated_data: str = "") -> bytes: + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + aad = associated_data.encode("utf-8") if associated_data else None + return aes.decrypt(nonce, cipher, aad) + +# ========= Tensor <-> RGB uint8 ========= +def tensor2rgb_u8(img_tensor: torch.Tensor) -> np.ndarray: + t = img_tensor[0].detach().cpu().numpy() + t = np.clip(t, 0.0, 1.0) + arr = (t * 255.0).round().astype(np.uint8) + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + if arr.shape[2] < 3: + arr = np.tile(arr[:, :, :1], (1, 1, 3)) + if arr.shape[2] > 3: + arr = arr[:, :, :3] + return arr + +def rgb_u8_to_tensor(arr: np.ndarray): + arr = arr.astype(np.float32) / 255.0 + t = torch.from_numpy(arr).unsqueeze(0) + return (t,) + +# ========= LSB helpers ========= +MAGIC = b"EMIIMG2" # 7 bytes (v2 header) + +# Header v2 (50 octets): +# 0..6 MAGIC (7) +# 7 BPC (1) +# 8..11 SH (4, >I) +# 12..15 SW (4, >I) +# 16 SC (1) (toujours 3 = RGB aprĂšs dĂ©codage) +# 17 ENC (1) (0=raw, 1=png, 2=jpeg) +# 18..33 SALT (16) +# 34..45 NONCE (12) +# 46..49 LEN (4, >I) +HEADER_SIZE = 50 + +def _bytes_to_bits(data: bytes) -> np.ndarray: + arr = np.frombuffer(data, dtype=np.uint8) + return np.unpackbits(arr) + +def _bits_to_bytes(bits: np.ndarray) -> bytes: + if bits.size % 8 != 0: + bits = np.pad(bits, (0, 8 - (bits.size % 8)), constant_values=0) + return np.packbits(bits).tobytes() + +def _capacity_bits(rgb: np.ndarray, bpc: int) -> int: + h, w, _ = rgb.shape + return h * w * 3 * bpc + +def _embed_bits_into_image(rgb: np.ndarray, bits: np.ndarray, bpc: int) -> np.ndarray: + H, W, _ = rgb.shape + cap = _capacity_bits(rgb, bpc) + if bits.size > cap: + raise ValueError(f"Message too large. Capacity={cap} bits, need={bits.size} bits.") + + flat = rgb.reshape(-1, 3).astype(np.uint8) + channels = [flat[:, 0], flat[:, 1], flat[:, 2]] + bit_idx = 0 + mask = 0xFF ^ ((1 << bpc) - 1) + + for c in range(3): + ch = channels[c] + if bpc == 1: + take = min(bits.size - bit_idx, ch.size) + if take > 0: + vals = ch & mask + vals[:take] |= bits[bit_idx:bit_idx + take].astype(np.uint8) + ch[:take] = vals[:take] + bit_idx += take + else: + remaining = bits.size - bit_idx + pairs = min(ch.size, remaining // 2) + if pairs > 0: + vals = ch & mask + b = bits[bit_idx:bit_idx + pairs * 2].reshape(-1, 2) + twob = (b[:, 0].astype(np.uint8) << 1) | b[:, 1].astype(np.uint8) + vals[:pairs] |= twob + ch[:pairs] = vals[:pairs] + bit_idx += pairs * 2 + channels[c] = ch + if bit_idx >= bits.size: + break + + out = np.stack(channels, axis=1).reshape(H, W, 3) + return out + +def _extract_bits_from_image(rgb: np.ndarray, nbits: int, bpc: int) -> np.ndarray: + H, W, _ = rgb.shape + cap = _capacity_bits(rgb, bpc) + if nbits > cap: + raise ValueError("Requested more bits than image capacity.") + + flat = rgb.reshape(-1, 3).astype(np.uint8) + channels = [flat[:, 0], flat[:, 1], flat[:, 2]] + out_bits = np.zeros(nbits, dtype=np.uint8) + bit_idx = 0 + + for c in range(3): + ch = channels[c] + if bpc == 1: + take = min(nbits - bit_idx, ch.size) + if take > 0: + out_bits[bit_idx:bit_idx + take] = (ch[:take] & 0x01).astype(np.uint8) + bit_idx += take + else: + remaining = nbits - bit_idx + pairs = min(ch.size, (remaining + 1) // 2) + if pairs > 0: + vals = ch[:pairs] & 0x03 + b1 = (vals >> 1) & 1 + b2 = vals & 1 + merged = np.column_stack([b1, b2]).reshape(-1) + take = min(remaining, merged.size) + out_bits[bit_idx:bit_idx + take] = merged[:take] + bit_idx += take + if bit_idx >= nbits: + break + return out_bits + +def _build_header(bpc: int, sh: int, sw: int, sc: int, enc_type: int, salt: bytes, nonce: bytes, payload_len: int) -> bytes: + return ( + MAGIC + + bytes([bpc & 0xFF]) + + struct.pack(">I", sh) + + struct.pack(">I", sw) + + bytes([sc & 0xFF]) + + bytes([enc_type & 0xFF]) + + salt + + nonce + + struct.pack(">I", payload_len) + ) + +def _parse_header_bytes(hb: bytes): + if len(hb) < HEADER_SIZE: + raise ValueError("Header too small") + if hb[:7] != MAGIC: + raise ValueError("Invalid magic") + bpc = hb[7] + sh = struct.unpack(">I", hb[8:12])[0] + sw = struct.unpack(">I", hb[12:16])[0] + sc = hb[16] + enc_type = hb[17] + salt = hb[18:34] + nonce = hb[34:46] + payload_len = struct.unpack(">I", hb[46:50])[0] + if len(salt) != 16 or len(nonce) != 12: + raise ValueError("Invalid salt/nonce length") + return bpc, sh, sw, sc, enc_type, salt, nonce, payload_len + +def _read_header(rgb: np.ndarray): + for bpc_try in (1, 2): + bits = _extract_bits_from_image(rgb, HEADER_SIZE * 8, bpc_try) + hb = _bits_to_bytes(bits)[:HEADER_SIZE] + try: + return _parse_header_bytes(hb) + except Exception: + continue + raise ValueError("Header not found") + +# ========= Nodes ========= +class Stego_EmbedImage: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "cover_image": ("IMAGE",), + "secret_image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + "bits_per_channel": ([1, 2], {"default": 1}), + "secret_encode": (["png", "jpeg"], {"default": "png"}), + }, + "optional": { + "jpeg_quality": ("INT", {"default": 85, "min": 50, "max": 95}), + "associated_data": ("STRING", {"multiline": False, "default": ""}), + "resize_secret_to_cover": ("BOOLEAN", {"default": True}), + }, + } + + RETURN_TYPES = ("IMAGE", "STRING") + RETURN_NAMES = ("image", "report") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, cover_image, secret_image, passphrase, bits_per_channel, + secret_encode, jpeg_quality=85, associated_data="", resize_secret_to_cover=True): + + if bits_per_channel not in (1, 2): + bits_per_channel = 1 + + cover = tensor2rgb_u8(cover_image) + + # --- prĂ©parer l'image secrĂšte (et VRAIMENT la redimensionner si demandĂ©) --- + secret_arr = tensor2rgb_u8(secret_image) + if resize_secret_to_cover: + sw, sh = cover.shape[1], cover.shape[0] + secret_pil = Image.fromarray(secret_arr, mode="RGB").resize((sw, sh), Image.LANCZOS) + else: + sh, sw, _ = secret_arr.shape + secret_pil = Image.fromarray(secret_arr, mode="RGB") + + # --- encodage (compression) avant chiffrement --- + enc_type = 1 if secret_encode == "png" else 2 # 0=raw,1=png,2=jpeg + if enc_type == 1: # PNG + buf = io.BytesIO() + # PNG lossless -> peut faire 100–1500 Ko selon contenu + secret_pil.save(buf, format="PNG", optimize=True, compress_level=6) + plaintext = buf.getvalue() + elif enc_type == 2: # JPEG + buf = io.BytesIO() + # JPEG q85 -> ~200–400 Ko sur 1216x1600 typique + secret_pil.save(buf, format="JPEG", quality=int(jpeg_quality), subsampling=1, optimize=True) + plaintext = buf.getvalue() + else: # raw (rarement utile) + plaintext = np.array(secret_pil, dtype=np.uint8).tobytes(order="C") + enc_type = 0 + + salt, nonce, cipher = encrypt_bytes(plaintext, passphrase, associated_data) + + header = _build_header(bits_per_channel, sh, sw, 3, enc_type, salt, nonce, len(cipher)) + payload = header + cipher + bits = _bytes_to_bits(payload) + + cap_bits = _capacity_bits(cover, bits_per_channel) + if bits.size > cap_bits: + need_bytes = (bits.size + 7) // 8 + have_bytes = cap_bits // 8 + msg = (f"[StegoImage ERROR] Capacity too small: need ~{need_bytes} bytes, have ~{have_bytes} bytes. " + f"Tip: use secret_encode='jpeg' (q={jpeg_quality}) ou une cover plus grande / bpc=2.") + return rgb_u8_to_tensor(cover)[0], msg + + stego = _embed_bits_into_image(cover, bits, bits_per_channel) + return rgb_u8_to_tensor(stego)[0], f"Embedded secret {sw}x{sh} enc={secret_encode} bpc={bits_per_channel}" + +class Stego_ExtractImage: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + }, + } + + RETURN_TYPES = ("IMAGE", "STRING") + RETURN_NAMES = ("image", "report") + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, passphrase, associated_data=""): + cover = tensor2rgb_u8(image) + try: + bpc_enc, sh, sw, sc, enc_type, salt, nonce, payload_len = _read_header(cover) + except Exception as e: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], f"[StegoImage ERROR] {type(e).__name__}: {e}" + + total_bits = (HEADER_SIZE + payload_len) * 8 + try: + bits = _extract_bits_from_image(cover, total_bits, int(bpc_enc)) + except Exception as e: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], f"[StegoImage ERROR] {type(e).__name__}: {e}" + + blob = _bits_to_bytes(bits)[:HEADER_SIZE + payload_len] + try: + bpc2, sh2, sw2, sc2, enc2, salt2, nonce2, len2 = _parse_header_bytes(blob[:HEADER_SIZE]) + if not (bpc2 == bpc_enc and sh2 == sh and sw2 == sw and sc2 == sc and enc2 == enc_type and salt2 == salt and nonce2 == nonce and len2 == payload_len): + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], "[StegoImage ERROR] Header mismatch" + except Exception as e: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], f"[StegoImage ERROR] {type(e).__name__}: {e}" + + cipher = blob[HEADER_SIZE:HEADER_SIZE + payload_len] + try: + plaintext = decrypt_bytes(salt, nonce, cipher, passphrase, associated_data) + except Exception as e: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], f"[StegoImage ERROR] {type(e).__name__}: {e}" + + # dĂ©codage selon enc_type + if enc_type == 0: + # raw RGB + if len(plaintext) != sh * sw * 3: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], "[StegoImage ERROR] Invalid raw size" + secret = np.frombuffer(plaintext, dtype=np.uint8).reshape(sh, sw, 3) + else: + try: + img = Image.open(io.BytesIO(plaintext)).convert("RGB") + secret = np.array(img, dtype=np.uint8) + except Exception as e: + tiny = np.zeros((1,1,3), dtype=np.uint8) + return rgb_u8_to_tensor(tiny)[0], f"[StegoImage ERROR] Decode failed: {type(e).__name__}" + return rgb_u8_to_tensor(secret)[0], f"Recovered secret {secret.shape[1]}x{secret.shape[0]} (enc={'png' if enc_type==1 else 'jpeg' if enc_type==2 else 'raw'})" + + +# ========= ComfyUI mappings ========= +NODE_CLASS_MAPPINGS = { + "Stego Embed Image": Stego_EmbedImage, + "Stego Extract Image": Stego_ExtractImage, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Stego Embed Image": "EncryptMaster — Stego Embed Image (LSB/AES-GCM)", + "Stego Extract Image": "EncryptMaster — Stego Extract Image (LSB/AES-GCM)", +} diff --git a/stego_text.py b/stego_text.py new file mode 100644 index 0000000..1ebd1a2 --- /dev/null +++ b/stego_text.py @@ -0,0 +1,329 @@ +# stego_text.py +# EncryptMaster — Stego Text (AES-256-GCM + scrypt + LSB) +# Cache/extrais un texte chiffrĂ© dans une image (PNG/TIFF recommandĂ©s). +# Nodes : +# - Stego_EmbedText : IMAGE, STRING, passphrase -> IMAGE +# - Stego_ExtractText : IMAGE, passphrase -> STRING + +import os +import struct +from typing import Tuple + +import numpy as np +from PIL import Image +import torch + +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +# ===================== Crypto helpers ===================== + +def _derive_key(passphrase: str, salt: bytes, length: int = 32) -> bytes: + if not passphrase: + raise ValueError("passphrase cannot be empty") + kdf = Scrypt(salt=salt, length=length, n=2**14, r=8, p=1) + return kdf.derive(passphrase.encode("utf-8")) + +def encrypt_bytes(plaintext: bytes, passphrase: str, associated_data: str = "") -> Tuple[bytes, bytes, bytes]: + salt = os.urandom(16) + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + nonce = os.urandom(12) + aad = associated_data.encode("utf-8") if associated_data else None + cipher = aes.encrypt(nonce, plaintext, aad) # ciphertext || tag + return salt, nonce, cipher + +def decrypt_bytes(salt: bytes, nonce: bytes, cipher: bytes, passphrase: str, associated_data: str = "") -> bytes: + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + aad = associated_data.encode("utf-8") if associated_data else None + return aes.decrypt(nonce, cipher, aad) + + +# ===================== Tensor <-> PIL helpers (ComfyUI) ===================== + +def tensor2pil(img_tensor: torch.Tensor) -> Image.Image: + """ + img_tensor: torch.Tensor [B,H,W,C] float32 in [0,1] + -> PIL.Image (RGB/RGBA) + """ + t = img_tensor[0].detach().cpu() + arr = t.numpy() + arr = np.clip(arr, 0.0, 1.0) + arr = (arr * 255.0).round().astype(np.uint8) + + if arr.ndim == 3 and arr.shape[2] == 4: + mode = "RGBA" + return Image.fromarray(arr, mode=mode) + # force RGB + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + elif arr.ndim == 3 and arr.shape[2] > 3: + arr = arr[:, :, :3] + return Image.fromarray(arr, mode="RGB") + +def pil2tensor(img: Image.Image): + """ + PIL.Image -> (torch.Tensor [B,H,W,C] float32 in [0,1],) + """ + if img.mode not in ("RGB", "RGBA"): + img = img.convert("RGB") + arr = np.array(img) + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + if arr.ndim == 3 and arr.shape[2] == 4: + # On travaille en RGB pour la stĂ©ganographie (3 canaux) + arr = arr[:, :, :3] + arr = arr.astype(np.float32) / 255.0 + t = torch.from_numpy(arr).unsqueeze(0) # [1,H,W,C] + return (t,) + + +# ===================== LSB helpers ===================== + +MAGIC = b"EMSTEG1" # 7 bytes +# Header: MAGIC(7) | BPC(1) | SALT(16) | NONCE(12) | LEN(4, big-endian) +HEADER_FMT = ">7sB16s12sI" +HEADER_SIZE = struct.calcsize(HEADER_FMT) # 40 bytes + +def _bytes_to_bits(data: bytes) -> np.ndarray: + arr = np.frombuffer(data, dtype=np.uint8) + return np.unpackbits(arr) + +def _bits_to_bytes(bits: np.ndarray) -> bytes: + if bits.size % 8 != 0: + bits = np.pad(bits, (0, 8 - (bits.size % 8)), constant_values=0) + arr = np.packbits(bits) + return arr.tobytes() + +def _capacity_bits(rgb: np.ndarray, bpc: int) -> int: + h, w, _ = rgb.shape + return h * w * 3 * bpc + +def _embed_bits_into_image(rgb: np.ndarray, bits: np.ndarray, bpc: int) -> np.ndarray: + """ + rgb: HxWx3 uint8 + bits: flat array of {0,1} + bpc: 1 or 2 + """ + H, W, _ = rgb.shape + capacity_bits = _capacity_bits(rgb, bpc) + if bits.size > capacity_bits: + raise ValueError(f"Message too large. Capacity={capacity_bits} bits, need={bits.size} bits.") + + flat = rgb.reshape(-1, 3).astype(np.uint8) + channels = [flat[:, 0], flat[:, 1], flat[:, 2]] + + bit_idx = 0 + mask = 0xFF ^ ((1 << bpc) - 1) # clear last bpc bits + + for c in range(3): + ch = channels[c] + if bit_idx >= bits.size: + break + + slots = ch.size + nbits_for_channel = slots * bpc + take = min(bits.size - bit_idx, nbits_for_channel) + + vals = ch & mask + + if bpc == 1: + vals[:take] |= bits[bit_idx:bit_idx + take].astype(np.uint8) + ch[:take] = vals[:take] + bit_idx += take + else: # bpc == 2 + # nombre de paires (2 bits) + pairs = take // 2 + if pairs > 0: + b = bits[bit_idx:bit_idx + pairs * 2].reshape(-1, 2) + twob = (b[:, 0].astype(np.uint8) << 1) | b[:, 1].astype(np.uint8) + vals[:pairs] |= twob + ch[:pairs] = vals[:pairs] + bit_idx += pairs * 2 + + channels[c] = ch + + out = np.stack(channels, axis=1).reshape(H, W, 3) + return out + +def _extract_bits_from_image(rgb: np.ndarray, nbits: int, bpc: int) -> np.ndarray: + H, W, _ = rgb.shape + capacity_bits = _capacity_bits(rgb, bpc) + if nbits > capacity_bits: + raise ValueError("Requested more bits than image capacity.") + + flat = rgb.reshape(-1, 3).astype(np.uint8) + channels = [flat[:, 0], flat[:, 1], flat[:, 2]] + + out_bits = np.zeros(nbits, dtype=np.uint8) + bit_idx = 0 + + for c in range(3): + ch = channels[c] + if bit_idx >= nbits: + break + + if bpc == 1: + take = min(nbits - bit_idx, ch.size) + out_bits[bit_idx:bit_idx + take] = (ch[:take] & 0x01).astype(np.uint8) + bit_idx += take + else: # 2 bits + remaining = nbits - bit_idx + pairs = min(ch.size, (remaining + 1) // 2) + if pairs <= 0: + continue + vals = ch[:pairs] & 0x03 + b1 = (vals >> 1) & 1 + b2 = vals & 1 + merged = np.column_stack([b1, b2]).reshape(-1) + take = min(remaining, merged.size) + out_bits[bit_idx:bit_idx + take] = merged[:take] + bit_idx += take + + return out_bits + +def _build_header_bytes(bpc: int, salt: bytes, nonce: bytes, payload_len: int) -> bytes: + return struct.pack(HEADER_FMT, MAGIC, bpc, salt, nonce, payload_len) + +def _read_header(rgb: np.ndarray, bpc_hint: int = 1) -> Tuple[int, bytes, bytes, int]: + """ + Lit l'entĂȘte (40 octets) et renvoie (bpc, salt, nonce, payload_len) + Essaie avec bpc=1 puis bpc=2 si nĂ©cessaire. + """ + def try_read(bpc): + bits_needed = HEADER_SIZE * 8 + bits = _extract_bits_from_image(rgb, bits_needed, bpc) + header_bytes = _bits_to_bytes(bits)[:HEADER_SIZE] + magic, bpc_enc, salt, nonce, payload_len = struct.unpack(HEADER_FMT, header_bytes) + if magic != MAGIC: + raise ValueError("Invalid magic header") + if bpc_enc not in (1, 2): + raise ValueError("Invalid bpc in header") + return bpc_enc, salt, nonce, payload_len + + try: + return try_read(1) + except Exception: + return try_read(2) + + +# ===================== ComfyUI Nodes ===================== + +CATEGORY = "EncryptMaster" + +class Stego_EmbedText: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "text": ("STRING", {"multiline": True, "default": ""}), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + "bits_per_channel": ([1, 2], {"default": 1}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + }, + } + + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, text, passphrase, bits_per_channel, associated_data=""): + if bits_per_channel not in (1, 2): + return (image,) + + # tensor -> PIL -> np RGB + pil = tensor2pil(image) + if pil.mode != "RGB": + pil = pil.convert("RGB") + rgb = np.array(pil, dtype=np.uint8) + + # chiffrer le texte + plaintext = text.encode("utf-8") + salt, nonce, cipher = encrypt_bytes(plaintext, passphrase, associated_data) + + header = _build_header_bytes(bits_per_channel, salt, nonce, len(cipher)) + payload = header + cipher + bits = _bytes_to_bits(payload) + + # capacitĂ© + cap_bits = _capacity_bits(rgb, bits_per_channel) + if bits.size > cap_bits: + need_bytes = (bits.size + 7) // 8 + have_bytes = cap_bits // 8 + return (tensor2pil(image),) if False else ( # dead code to keep type + pil2tensor(pil)[0], + ) # pragma: no cover + + rgb_stego = _embed_bits_into_image(rgb, bits, bits_per_channel) + out = Image.fromarray(rgb_stego, mode="RGB") + return pil2tensor(out) + + +class Stego_ExtractText: + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "image": ("IMAGE",), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + }, + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("text",) + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, image, passphrase, associated_data=""): + pil = tensor2pil(image) + if pil.mode != "RGB": + pil = pil.convert("RGB") + rgb = np.array(pil, dtype=np.uint8) + + try: + bpc, salt, nonce, payload_len = _read_header(rgb) + except Exception as e: + return (f"[Stego ERROR] {type(e).__name__}: {e}",) + + total_bits = (HEADER_SIZE + payload_len) * 8 + try: + bits = _extract_bits_from_image(rgb, total_bits, bpc) + except Exception as e: + return (f"[Stego ERROR] {type(e).__name__}: {e}",) + + blob = _bits_to_bytes(bits)[: HEADER_SIZE + payload_len] + try: + magic, bpc_enc, salt2, nonce2, payload_len2 = struct.unpack(HEADER_FMT, blob[:HEADER_SIZE]) + if magic != MAGIC or salt2 != salt or nonce2 != nonce or payload_len2 != payload_len: + return ("[Stego ERROR] Header mismatch",) + except Exception as e: + return (f"[Stego ERROR] {type(e).__name__}: {e}",) + + cipher = blob[HEADER_SIZE : HEADER_SIZE + payload_len] + try: + pt = decrypt_bytes(salt, nonce, cipher, passphrase, associated_data) + return (pt.decode("utf-8"),) + except Exception as e: + return (f"[Stego ERROR] {type(e).__name__}: {e}",) + + +# ===================== ComfyUI mappings ===================== + +NODE_CLASS_MAPPINGS = { + "Stego Embed Text": Stego_EmbedText, + "Stego Extract Text": Stego_ExtractText, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + "Stego Embed Text": "EncryptMaster — Stego Embed Text (AES-GCM)", + "Stego Extract Text": "EncryptMaster — Stego Extract Text (AES-GCM)", +} diff --git a/text_cipher.py b/text_cipher.py new file mode 100644 index 0000000..594b390 --- /dev/null +++ b/text_cipher.py @@ -0,0 +1,125 @@ +# text_cipher.py +# Text Cipher (AES-256-GCM + scrypt) +# +# EntrĂ©es: +# - text (STRING, multiline) : texte en clair OU chaĂźne chiffrĂ©e "armored" +# - passphrase (STRING, password) : mot de passe humain +# - mode ("encrypt" | "decrypt") +# - associated_data (STRING, optionnel) : mĂ©tadonnĂ©es authentifiĂ©es (AEAD) +# +# Sorties: +# - text (STRING) : si mode=encrypt => chaĂźne chiffrĂ©e "armored" +# si mode=decrypt => texte en clair +# +# DĂ©pendance : pip install cryptography + +import base64 +import os +from typing import Tuple + +from cryptography.hazmat.primitives.kdf.scrypt import Scrypt +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + +CATEGORY = "EncryptMaster" +DISPLAY_NAME = "Text Cipher (AES-GCM)" +NODE_NAME = "TextCipher" + +def _b64e(b: bytes) -> str: + return base64.urlsafe_b64encode(b).decode("utf-8").rstrip("=") + +def _b64d(s: str) -> bytes: + # Supporte base64 urlsafe sans padding + pad = "=" * (-len(s) % 4) + return base64.urlsafe_b64decode(s + pad) + +def _derive_key(passphrase: str, salt: bytes, length: int = 32) -> bytes: + # scrypt parameters: N=2^14, r=8, p=1 (bon Ă©quilibre sĂ©curitĂ©/perf CPU) + kdf = Scrypt(salt=salt, length=length, n=2**14, r=8, p=1) + return kdf.derive(passphrase.encode("utf-8")) + +def _encrypt(plaintext: str, passphrase: str, associated_data: str = "") -> str: + if not isinstance(plaintext, str): + raise TypeError("plaintext must be str") + if not passphrase: + raise ValueError("passphrase cannot be empty") + + salt = os.urandom(16) # pour KDF scrypt + key = _derive_key(passphrase, salt) # 32 bytes + aes = AESGCM(key) + nonce = os.urandom(12) # requis par AES-GCM + aad = associated_data.encode("utf-8") if associated_data else None + ct = aes.encrypt(nonce, plaintext.encode("utf-8"), aad) + # Format “armored” + return f"aesgcm-scrypt.v1${_b64e(salt)}${_b64e(nonce)}${_b64e(ct)}" + +def _parse_armored(armored: str) -> Tuple[bytes, bytes, bytes]: + # aesgcm-scrypt.v1$$$ + parts = armored.split("$") + if len(parts) != 4 or parts[0] != "aesgcm-scrypt.v1": + raise ValueError("Invalid armored format or version.") + salt = _b64d(parts[1]) + nonce = _b64d(parts[2]) + cipher = _b64d(parts[3]) + if len(salt) < 16 or len(nonce) != 12: + raise ValueError("Invalid salt/nonce length.") + return salt, nonce, cipher + +def _decrypt(armored: str, passphrase: str, associated_data: str = "") -> str: + if not isinstance(armored, str): + raise TypeError("ciphertext must be str") + if not passphrase: + raise ValueError("passphrase cannot be empty") + + salt, nonce, cipher = _parse_armored(armored) + key = _derive_key(passphrase, salt) + aes = AESGCM(key) + aad = associated_data.encode("utf-8") if associated_data else None + pt = aes.decrypt(nonce, cipher, aad) + return pt.decode("utf-8") + + +class TextCipher: + """ + Un node ComfyUI simple pour chiffrer/dĂ©chiffrer du texte avec AES-256-GCM + et dĂ©rivation de clĂ© via scrypt. Compatible “list mapping” (STRING). + """ + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "text": ("STRING", {"multiline": True, "default": ""}), + "passphrase": ("STRING", {"multiline": False, "default": "", "password": True}), + "mode": (["encrypt", "decrypt"], {"default": "encrypt"}), + }, + "optional": { + "associated_data": ("STRING", {"multiline": False, "default": ""}), + } + } + + RETURN_TYPES = ("STRING",) + RETURN_NAMES = ("text",) + FUNCTION = "run" + CATEGORY = CATEGORY + + def run(self, text: str, passphrase: str, mode: str, associated_data: str = ""): + try: + if mode == "encrypt": + out = _encrypt(text, passphrase, associated_data) + else: + out = _decrypt(text, passphrase, associated_data) + return (out,) + except Exception as e: + # Retourne l'erreur en clair (option pragmatique pour debug workflow) + return (f"[TextCipher ERROR] {type(e).__name__}: {e}",) + + +# Mappings pour ComfyUI +NODE_CLASS_MAPPINGS = { + NODE_NAME: TextCipher, +} + +NODE_DISPLAY_NAME_MAPPINGS = { + NODE_NAME: DISPLAY_NAME, +}