alee/stegasoo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added file support and increased file limits.

Browse Source
This commit is contained in:
Aaron D. Lee
2025-12-28 03:44:17 -05:00
parent 130835990e
commit 5bd49cb581
16 changed files with 1576 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
src/stegasoo/__init__.py
View File

@@ -1,10 +1,10 @@
"""
Stegasoo - Secure Steganography Library
A Python library for hiding encrypted messages in images using
A Python library for hiding encrypted messages and files in images using
hybrid photo + passphrase + PIN authentication.
Basic Usage:
Basic Usage - Text Message:
from stegasoo import encode, decode, generate_credentials
# Generate credentials
@@ -30,16 +30,36 @@ Basic Usage:
f.write(result.stego_image)
# Decode a message
message = decode(
decoded = decode(
stego_image=result.stego_image,
reference_photo=ref_photo,
day_phrase="apple forest thunder",
pin="123456"
)
print(message) # "Meet at midnight"
print(decoded.message) # "Meet at midnight"
File Embedding:
from stegasoo import encode_file, decode, FilePayload
# Encode a file
result = encode_file(
filepath="secret_document.pdf",
reference_photo=ref_photo,
carrier_image=carrier,
day_phrase="apple forest thunder",
pin="123456"
)
# Decode - automatically detects file vs text
decoded = decode(...)
if decoded.is_file:
with open(decoded.filename, 'wb') as f:
f.write(decoded.file_data)
else:
print(decoded.message)
"""
from .constants import __version__, DAY_NAMES
from .constants import __version__, DAY_NAMES, MAX_MESSAGE_SIZE, MAX_FILE_PAYLOAD_SIZE
from .models import (
Credentials,
EncodeInput,
@@ -49,6 +69,7 @@ from .models import (
EmbedStats,
KeyInfo,
ValidationResult,
FilePayload,
)
from .exceptions import (
StegasooError,
@@ -83,6 +104,8 @@ from .keygen import (
from .validation import (
validate_pin,
validate_message,
validate_payload,
validate_file_payload,
validate_image,
validate_rsa_key,
validate_security_factors,
@@ -90,6 +113,7 @@ from .validation import (
validate_date_string,
require_valid_pin,
require_valid_message,
require_valid_payload,
require_valid_image,
require_valid_rsa_key,
require_security_factors,
@@ -97,6 +121,7 @@ from .validation import (
from .crypto import (
encrypt_message,
decrypt_message,
decrypt_message_text,
derive_hybrid_key,
derive_pixel_key,
hash_photo,
@@ -109,6 +134,9 @@ from .steganography import (
extract_from_image,
calculate_capacity,
get_image_dimensions,
get_image_format,
is_lossless_format,
LOSSLESS_FORMATS,
)
from .utils import (
generate_filename,
@@ -122,11 +150,12 @@ from .utils import (
)
from datetime import date
from typing import Optional
from pathlib import Path
from typing import Optional, Union
def encode(
message: str,
message: Union[str, bytes, FilePayload],
reference_photo: bytes,
carrier_image: bytes,
day_phrase: str,
@@ -134,15 +163,16 @@ def encode(
rsa_key_data: Optional[bytes] = None,
rsa_password: Optional[str] = None,
date_str: Optional[str] = None,
output_format: Optional[str] = None,
) -> EncodeResult:
"""
Encode a secret message into an image.
Encode a secret message or file into an image.
High-level convenience function that handles validation,
encryption, and embedding in one call.
Args:
message: Secret message to hide
message: Secret message (str), raw bytes, or FilePayload to hide
reference_photo: Shared reference photo bytes
carrier_image: Image to hide message in
day_phrase: Today's passphrase
@@ -150,6 +180,8 @@ def encode(
rsa_key_data: RSA private key PEM bytes (optional if using PIN)
rsa_password: Password for RSA key if encrypted
date_str: Date string YYYY-MM-DD (defaults to today)
output_format: Force output format ('PNG', 'BMP'). If None, preserves
carrier format for lossless types, defaults to PNG for lossy.
Returns:
EncodeResult with stego image and metadata
@@ -159,9 +191,13 @@ def encode(
SecurityFactorError: If no PIN or RSA key provided
CapacityError: If carrier is too small
EncryptionError: If encryption fails
Note:
Output format is always lossless (PNG or BMP) to preserve hidden data.
If carrier is JPEG/GIF, output will be PNG to maintain data integrity.
"""
# Validate inputs
require_valid_message(message)
require_valid_payload(message)
require_valid_image(carrier_image, "Carrier image")
require_security_factors(pin, rsa_key_data)
@@ -174,7 +210,7 @@ def encode(
if date_str is None:
date_str = date.today().isoformat()
# Encrypt message
# Encrypt message/file
encrypted = encrypt_message(
message, reference_photo, day_phrase, date_str, pin, rsa_key_data
)
@@ -184,11 +220,13 @@ def encode(
reference_photo, day_phrase, date_str, pin, rsa_key_data
)
# Embed in image
stego_data, stats = embed_in_image(carrier_image, encrypted, pixel_key)
# Embed in image (returns extension too)
stego_data, stats, extension = embed_in_image(
carrier_image, encrypted, pixel_key, output_format=output_format
)
# Generate filename
filename = generate_filename(date_str)
# Generate filename with correct extension
filename = generate_filename(date_str, extension=extension)
return EncodeResult(
stego_image=stego_data,
@@ -200,6 +238,102 @@ def encode(
)
def encode_file(
filepath: Union[str, Path],
reference_photo: bytes,
carrier_image: bytes,
day_phrase: str,
pin: str = "",
rsa_key_data: Optional[bytes] = None,
rsa_password: Optional[str] = None,
date_str: Optional[str] = None,
output_format: Optional[str] = None,
filename_override: Optional[str] = None,
) -> EncodeResult:
"""
Encode a file into an image.
Convenience function for embedding files. Preserves original filename.
Args:
filepath: Path to file to embed
reference_photo: Shared reference photo bytes
carrier_image: Image to hide file in
day_phrase: Today's passphrase
pin: Static PIN (optional if using RSA key)
rsa_key_data: RSA private key PEM bytes (optional if using PIN)
rsa_password: Password for RSA key if encrypted
date_str: Date string YYYY-MM-DD (defaults to today)
output_format: Force output format ('PNG', 'BMP')
filename_override: Override the stored filename
Returns:
EncodeResult with stego image and metadata
"""
payload = FilePayload.from_file(str(filepath), filename_override)
return encode(
message=payload,
reference_photo=reference_photo,
carrier_image=carrier_image,
day_phrase=day_phrase,
pin=pin,
rsa_key_data=rsa_key_data,
rsa_password=rsa_password,
date_str=date_str,
output_format=output_format,
)
def encode_bytes(
data: bytes,
filename: str,
reference_photo: bytes,
carrier_image: bytes,
day_phrase: str,
pin: str = "",
rsa_key_data: Optional[bytes] = None,
rsa_password: Optional[str] = None,
date_str: Optional[str] = None,
output_format: Optional[str] = None,
mime_type: Optional[str] = None,
) -> EncodeResult:
"""
Encode raw bytes with a filename into an image.
Convenience function for embedding binary data with metadata.
Args:
data: Raw bytes to embed
filename: Filename to associate with the data
reference_photo: Shared reference photo bytes
carrier_image: Image to hide data in
day_phrase: Today's passphrase
pin: Static PIN (optional if using RSA key)
rsa_key_data: RSA private key PEM bytes (optional if using PIN)
rsa_password: Password for RSA key if encrypted
date_str: Date string YYYY-MM-DD (defaults to today)
output_format: Force output format ('PNG', 'BMP')
mime_type: MIME type of the data
Returns:
EncodeResult with stego image and metadata
"""
payload = FilePayload(data=data, filename=filename, mime_type=mime_type)
return encode(
message=payload,
reference_photo=reference_photo,
carrier_image=carrier_image,
day_phrase=day_phrase,
pin=pin,
rsa_key_data=rsa_key_data,
rsa_password=rsa_password,
date_str=date_str,
output_format=output_format,
)
def decode(
stego_image: bytes,
reference_photo: bytes,
@@ -207,15 +341,15 @@ def decode(
pin: str = "",
rsa_key_data: Optional[bytes] = None,
rsa_password: Optional[str] = None,
) -> str:
) -> DecodeResult:
"""
Decode a secret message from a stego image.
Decode a secret message or file from a stego image.
High-level convenience function that handles extraction
and decryption in one call.
Args:
stego_image: Image containing hidden message
stego_image: Image containing hidden message/file
reference_photo: Shared reference photo bytes
day_phrase: Passphrase for the day message was encoded
pin: Static PIN (if used during encoding)
@@ -223,7 +357,12 @@ def decode(
rsa_password: Password for RSA key if encrypted
Returns:
Decrypted message string
DecodeResult with:
- .payload_type: 'text' or 'file'
- .message: Decoded text (if text)
- .file_data: Decoded bytes (if file)
- .filename: Original filename (if file)
- .is_text / .is_file: Convenience properties
Raises:
ValidationError: If inputs are invalid
@@ -260,21 +399,72 @@ def decode(
if not encrypted:
raise ExtractionError("Could not extract data. Check your inputs.")
# Decrypt
# Decrypt and return full result
return decrypt_message(encrypted, reference_photo, day_phrase, pin, rsa_key_data)
def decode_text(
stego_image: bytes,
reference_photo: bytes,
day_phrase: str,
pin: str = "",
rsa_key_data: Optional[bytes] = None,
rsa_password: Optional[str] = None,
) -> str:
"""
Decode a text message from a stego image.
Convenience function that returns just the text string.
Raises an error if the content is a binary file.
Args:
stego_image: Image containing hidden message
reference_photo: Shared reference photo bytes
day_phrase: Passphrase for the day message was encoded
pin: Static PIN (if used during encoding)
rsa_key_data: RSA private key PEM bytes (if used during encoding)
rsa_password: Password for RSA key if encrypted
Returns:
Decrypted message string
Raises:
DecryptionError: If content is a binary file, not text
"""
result = decode(stego_image, reference_photo, day_phrase, pin, rsa_key_data, rsa_password)
if result.is_file:
# Try to decode file as text
if result.file_data:
try:
return result.file_data.decode('utf-8')
except UnicodeDecodeError:
raise DecryptionError(
f"Content is a binary file ({result.filename or 'unnamed'}), not text. "
"Use decode() instead and check result.is_file."
)
return ""
return result.message or ""
__all__ = [
# Version
'__version__',
# High-level API
'encode',
'encode_file',
'encode_bytes',
'decode',
'decode_text',
'generate_credentials',
# Constants
'DAY_NAMES',
'LOSSLESS_FORMATS',
'MAX_MESSAGE_SIZE',
'MAX_FILE_PAYLOAD_SIZE',
# Models
'Credentials',
@@ -285,6 +475,7 @@ __all__ = [
'EmbedStats',
'KeyInfo',
'ValidationResult',
'FilePayload',
# Exceptions
'StegasooError',
@@ -318,6 +509,8 @@ __all__ = [
# Validation
'validate_pin',
'validate_message',
'validate_payload',
'validate_file_payload',
'validate_image',
'validate_rsa_key',
'validate_security_factors',
@@ -325,6 +518,7 @@ __all__ = [
'validate_date_string',
'require_valid_pin',
'require_valid_message',
'require_valid_payload',
'require_valid_image',
'require_valid_rsa_key',
'require_security_factors',
@@ -332,6 +526,7 @@ __all__ = [
# Crypto
'encrypt_message',
'decrypt_message',
'decrypt_message_text',
'derive_hybrid_key',
'derive_pixel_key',
'hash_photo',
@@ -344,6 +539,8 @@ __all__ = [
'extract_from_image',
'calculate_capacity',
'get_image_dimensions',
'get_image_format',
'is_lossless_format',
# Utilities
'generate_filename',

22
src/stegasoo/constants.py
View File

@@ -11,7 +11,7 @@ from pathlib import Path
# VERSION
# ============================================================================
__version__ = "2.0.1"
__version__ = "2.1.1"
# ============================================================================
# FILE FORMAT
@@ -20,6 +20,10 @@ __version__ = "2.0.1"
MAGIC_HEADER = b'\x89ST3'
FORMAT_VERSION = 3
# Payload type markers
PAYLOAD_TEXT = 0x01
PAYLOAD_FILE = 0x02
# ============================================================================
# CRYPTO PARAMETERS
# ============================================================================
@@ -40,9 +44,11 @@ PBKDF2_ITERATIONS = 600000
# INPUT LIMITS
# ============================================================================
MAX_IMAGE_PIXELS = 4_000_000 # ~4 megapixels (2000x2000)
MAX_MESSAGE_SIZE = 50_000 # 50 KB
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5 MB
MAX_IMAGE_PIXELS = 16_000_000 # ~16 megapixels (4000x4000)
MAX_MESSAGE_SIZE = 250_000 # 250 KB (text messages)
MAX_FILE_PAYLOAD_SIZE = 250_000 # 250 KB (file payloads)
MAX_FILENAME_LENGTH = 255 # Max filename length to store
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB (upload limit)
MIN_PIN_LENGTH = 6
MAX_PIN_LENGTH = 9
@@ -78,11 +84,17 @@ DAY_NAMES = ('Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday',
def get_data_dir() -> Path:
"""Get the data directory path."""
# Check multiple locations
# From src/stegasoo/constants.py:
# .parent = src/stegasoo/
# .parent.parent = src/
# .parent.parent.parent = project root (where data/ lives)
candidates = [
Path(__file__).parent.parent.parent/ 'data', # Development
Path(__file__).parent.parent.parent / 'data', # Development: src/stegasoo -> project root
Path(__file__).parent / 'data', # Installed package
Path('/app/data'), # Docker
Path.cwd() / 'data', # Current directory
Path.cwd().parent / 'data', # One level up from cwd
Path.cwd().parent.parent / 'data', # Two levels up from cwd
]
for path in candidates:

180
src/stegasoo/crypto.py
View File

@@ -2,13 +2,15 @@
Stegasoo Cryptographic Functions
Key derivation, encryption, and decryption using AES-256-GCM.
Supports both text messages and binary file payloads.
"""
import io
import hashlib
import secrets
import struct
from typing import Optional
import json
from typing import Optional, Union
from PIL import Image
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
@@ -19,7 +21,10 @@ from .constants import (
SALT_SIZE, IV_SIZE, TAG_SIZE,
ARGON2_TIME_COST, ARGON2_MEMORY_COST, ARGON2_PARALLELISM,
PBKDF2_ITERATIONS,
PAYLOAD_TEXT, PAYLOAD_FILE,
MAX_FILENAME_LENGTH,
)
from .models import FilePayload, DecodeResult
from .exceptions import (
EncryptionError, DecryptionError, KeyDerivationError, InvalidHeaderError
)
@@ -171,8 +176,112 @@ def derive_pixel_key(
return hashlib.sha256(material + b"pixel_selection").digest()
def _pack_payload(
content: Union[str, bytes, FilePayload],
) -> tuple[bytes, int]:
"""
Pack payload with type marker and metadata.
Format for text:
[type:1][data]
Format for file:
[type:1][filename_len:2][filename][mime_len:2][mime][data]
Args:
content: Text string, raw bytes, or FilePayload
Returns:
Tuple of (packed bytes, payload type)
"""
if isinstance(content, str):
# Text message
data = content.encode('utf-8')
return bytes([PAYLOAD_TEXT]) + data, PAYLOAD_TEXT
elif isinstance(content, FilePayload):
# File with metadata
filename = content.filename[:MAX_FILENAME_LENGTH].encode('utf-8')
mime = (content.mime_type or '')[:100].encode('utf-8')
packed = (
bytes([PAYLOAD_FILE]) +
struct.pack('>H', len(filename)) +
filename +
struct.pack('>H', len(mime)) +
mime +
content.data
)
return packed, PAYLOAD_FILE
else:
# Raw bytes - treat as file with no name
packed = (
bytes([PAYLOAD_FILE]) +
struct.pack('>H', 0) + # No filename
struct.pack('>H', 0) + # No mime
content
)
return packed, PAYLOAD_FILE
def _unpack_payload(data: bytes) -> DecodeResult:
"""
Unpack payload and extract content with metadata.
Args:
data: Packed payload bytes
Returns:
DecodeResult with appropriate content
"""
if len(data) < 1:
raise DecryptionError("Empty payload")
payload_type = data[0]
if payload_type == PAYLOAD_TEXT:
# Text message
text = data[1:].decode('utf-8')
return DecodeResult(payload_type='text', message=text)
elif payload_type == PAYLOAD_FILE:
# File with metadata
offset = 1
# Read filename
filename_len = struct.unpack('>H', data[offset:offset+2])[0]
offset += 2
filename = data[offset:offset+filename_len].decode('utf-8') if filename_len else None
offset += filename_len
# Read mime type
mime_len = struct.unpack('>H', data[offset:offset+2])[0]
offset += 2
mime_type = data[offset:offset+mime_len].decode('utf-8') if mime_len else None
offset += mime_len
# Rest is file data
file_data = data[offset:]
return DecodeResult(
payload_type='file',
file_data=file_data,
filename=filename,
mime_type=mime_type
)
else:
# Unknown type - try to decode as text (backward compatibility)
try:
text = data.decode('utf-8')
return DecodeResult(payload_type='text', message=text)
except UnicodeDecodeError:
return DecodeResult(payload_type='file', file_data=data)
def encrypt_message(
message: str | bytes,
message: Union[str, bytes, FilePayload],
photo_data: bytes,
day_phrase: str,
date_str: str,
@@ -180,7 +289,7 @@ def encrypt_message(
rsa_key_data: Optional[bytes] = None
) -> bytes:
"""
Encrypt message using AES-256-GCM with hybrid key derivation.
Encrypt message or file using AES-256-GCM with hybrid key derivation.
Message format:
- Magic header (4 bytes)
@@ -193,7 +302,7 @@ def encrypt_message(
- Ciphertext (variable, padded)
Args:
message: Message to encrypt
message: Message string, raw bytes, or FilePayload to encrypt
photo_data: Reference photo bytes
day_phrase: The day's phrase
date_str: Date string (YYYY-MM-DD)
@@ -211,15 +320,15 @@ def encrypt_message(
key = derive_hybrid_key(photo_data, day_phrase, date_str, salt, pin, rsa_key_data)
iv = secrets.token_bytes(IV_SIZE)
if isinstance(message, str):
message = message.encode('utf-8')
# Pack payload with type marker
packed_payload, _ = _pack_payload(message)
# Random padding to hide message length
padding_len = secrets.randbelow(256) + 64
padded_len = ((len(message) + padding_len + 255) // 256) * 256
padding_needed = padded_len - len(message)
padding = secrets.token_bytes(padding_needed - 4) + struct.pack('>I', len(message))
padded_message = message + padding
padded_len = ((len(packed_payload) + padding_len + 255) // 256) * 256
padding_needed = padded_len - len(packed_payload)
padding = secrets.token_bytes(padding_needed - 4) + struct.pack('>I', len(packed_payload))
padded_message = packed_payload + padding
# Encrypt with AES-256-GCM
cipher = Cipher(algorithms.AES(key), modes.GCM(iv), backend=default_backend())
@@ -291,7 +400,7 @@ def decrypt_message(
day_phrase: str,
pin: str = "",
rsa_key_data: Optional[bytes] = None
) -> str:
) -> DecodeResult:
"""
Decrypt message using the embedded date from the header.
@@ -303,7 +412,7 @@ def decrypt_message(
rsa_key_data: Optional RSA key bytes
Returns:
Decrypted message string
DecodeResult with decrypted content
Raises:
InvalidHeaderError: If data doesn't have valid Stegasoo header
@@ -329,7 +438,11 @@ def decrypt_message(
padded_plaintext = decryptor.update(header['ciphertext']) + decryptor.finalize()
original_length = struct.unpack('>I', padded_plaintext[-4:])[0]
return padded_plaintext[:original_length].decode('utf-8')
payload_data = padded_plaintext[:original_length]
result = _unpack_payload(payload_data)
result.date_encoded = header['date']
return result
except Exception as e:
raise DecryptionError(
@@ -337,6 +450,47 @@ def decrypt_message(
) from e
def decrypt_message_text(
encrypted_data: bytes,
photo_data: bytes,
day_phrase: str,
pin: str = "",
rsa_key_data: Optional[bytes] = None
) -> str:
"""
Decrypt message and return as text string.
For backward compatibility - returns text content or raises error for files.
Args:
encrypted_data: Encrypted message bytes
photo_data: Reference photo bytes
day_phrase: The day's phrase
pin: Optional static PIN
rsa_key_data: Optional RSA key bytes
Returns:
Decrypted message string
Raises:
DecryptionError: If decryption fails or content is a file
"""
result = decrypt_message(encrypted_data, photo_data, day_phrase, pin, rsa_key_data)
if result.is_file:
if result.file_data:
# Try to decode as text
try:
return result.file_data.decode('utf-8')
except UnicodeDecodeError:
raise DecryptionError(
f"Content is a binary file ({result.filename or 'unnamed'}), not text"
)
return ""
return result.message or ""
def get_date_from_encrypted(encrypted_data: bytes) -> Optional[str]:
"""
Extract the date string from encrypted data without decrypting.

51
src/stegasoo/models.py
View File

@@ -6,7 +6,7 @@ Dataclasses for structured data exchange between modules and frontends.
from dataclasses import dataclass, field
from datetime import date
from typing import Optional
from typing import Optional, Union
@dataclass
@@ -43,10 +43,35 @@ class Credentials:
return self.phrase_entropy + self.pin_entropy + self.rsa_entropy
@dataclass
class FilePayload:
"""Represents a file to be embedded."""
data: bytes
filename: str
mime_type: Optional[str] = None
@property
def size(self) -> int:
return len(self.data)
@classmethod
def from_file(cls, filepath: str, filename: Optional[str] = None) -> 'FilePayload':
"""Create FilePayload from a file path."""
from pathlib import Path
import mimetypes
path = Path(filepath)
data = path.read_bytes()
name = filename or path.name
mime, _ = mimetypes.guess_type(name)
return cls(data=data, filename=name, mime_type=mime)
@dataclass
class EncodeInput:
"""Input parameters for encoding a message."""
message: str
message: Union[str, bytes, FilePayload] # Text, raw bytes, or file
reference_photo: bytes
carrier_image: bytes
day_phrase: str
@@ -90,8 +115,26 @@ class DecodeInput:
@dataclass
class DecodeResult:
"""Result of decoding operation."""
message: str
date_encoded: str
payload_type: str # 'text' or 'file'
message: Optional[str] = None # For text payloads
file_data: Optional[bytes] = None # For file payloads
filename: Optional[str] = None # Original filename for file payloads
mime_type: Optional[str] = None # MIME type hint
date_encoded: Optional[str] = None
@property
def is_file(self) -> bool:
return self.payload_type == 'file'
@property
def is_text(self) -> bool:
return self.payload_type == 'text'
def get_content(self) -> Union[str, bytes]:
"""Get the decoded content (text or bytes)."""
if self.is_text:
return self.message or ""
return self.file_data or b""
@dataclass

74
src/stegasoo/steganography.py
View File

@@ -16,6 +16,43 @@ from .models import EmbedStats
from .exceptions import CapacityError, ExtractionError, EmbeddingError
# Lossless formats that preserve LSB data
LOSSLESS_FORMATS = {'PNG', 'BMP', 'TIFF'}
# Format to extension mapping
FORMAT_TO_EXT = {
'PNG': 'png',
'BMP': 'bmp',
'TIFF': 'tiff',
}
# Extension to PIL format mapping
EXT_TO_FORMAT = {
'png': 'PNG',
'bmp': 'BMP',
'tiff': 'TIFF',
'tif': 'TIFF',
}
def get_output_format(input_format: Optional[str]) -> tuple[str, str]:
"""
Determine the output format based on input format.
Args:
input_format: PIL format string of input image (e.g., 'JPEG', 'PNG')
Returns:
Tuple of (PIL format string, file extension) for output
Falls back to PNG for lossy or unknown formats.
"""
if input_format and input_format.upper() in LOSSLESS_FORMATS:
fmt = input_format.upper()
return fmt, FORMAT_TO_EXT.get(fmt, 'png')
# Default to PNG for lossy formats (JPEG, GIF) or unknown
return 'PNG', 'png'
def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list[int]:
"""
Generate pseudo-random pixel indices for embedding.
@@ -83,8 +120,9 @@ def embed_in_image(
carrier_data: bytes,
encrypted_data: bytes,
pixel_key: bytes,
bits_per_channel: int = 1
) -> tuple[bytes, EmbedStats]:
bits_per_channel: int = 1,
output_format: Optional[str] = None
) -> tuple[bytes, EmbedStats, str]:
"""
Embed encrypted data in carrier image using LSB steganography.
@@ -96,9 +134,11 @@ def embed_in_image(
encrypted_data: Data to embed
pixel_key: Key for pixel selection
bits_per_channel: Bits to use per color channel (1-2)
output_format: Force specific output format (PNG, BMP).
If None, auto-detect from carrier (lossless) or default to PNG.
Returns:
Tuple of (PNG image bytes, EmbedStats)
Tuple of (image bytes, EmbedStats, file extension)
Raises:
CapacityError: If carrier is too small
@@ -106,6 +146,8 @@ def embed_in_image(
"""
try:
img = Image.open(io.BytesIO(carrier_data))
input_format = img.format
if img.mode != 'RGB':
img = img.convert('RGB')
@@ -160,8 +202,15 @@ def embed_in_image(
stego_img = Image.new('RGB', img.size)
stego_img.putdata(new_pixels)
# Determine output format
if output_format:
out_fmt = output_format.upper()
out_ext = FORMAT_TO_EXT.get(out_fmt, 'png')
else:
out_fmt, out_ext = get_output_format(input_format)
output = io.BytesIO()
stego_img.save(output, 'PNG')
stego_img.save(output, out_fmt)
output.seek(0)
stats = EmbedStats(
@@ -171,7 +220,7 @@ def embed_in_image(
bytes_embedded=len(data_with_len)
)
return output.getvalue(), stats
return output.getvalue(), stats, out_ext
except CapacityError:
raise
@@ -284,3 +333,18 @@ def get_image_dimensions(image_data: bytes) -> tuple[int, int]:
"""Get image dimensions without loading full image."""
img = Image.open(io.BytesIO(image_data))
return img.size
def get_image_format(image_data: bytes) -> Optional[str]:
"""Get image format (PIL format string like 'PNG', 'JPEG')."""
try:
img = Image.open(io.BytesIO(image_data))
return img.format
except Exception:
return None
def is_lossless_format(image_data: bytes) -> bool:
"""Check if image is in a lossless format suitable for steganography."""
fmt = get_image_format(image_data)
return fmt is not None and fmt.upper() in LOSSLESS_FORMATS

14
src/stegasoo/utils.py
View File

@@ -15,15 +15,20 @@ from typing import Optional
from .constants import DAY_NAMES
def generate_filename(date_str: Optional[str] = None, prefix: str = "") -> str:
def generate_filename(
date_str: Optional[str] = None,
prefix: str = "",
extension: str = "png"
) -> str:
"""
Generate a filename for stego images.
Format: {prefix}{random}_{YYYYMMDD}.png
Format: {prefix}{random}_{YYYYMMDD}.{extension}
Args:
date_str: Date string (YYYY-MM-DD), defaults to today
prefix: Optional prefix
extension: File extension without dot (default: 'png')
Returns:
Filename string
@@ -34,7 +39,10 @@ def generate_filename(date_str: Optional[str] = None, prefix: str = "") -> str:
date_compact = date_str.replace('-', '')
random_hex = secrets.token_hex(4)
return f"{prefix}{random_hex}_{date_compact}.png"
# Ensure extension doesn't have a leading dot
extension = extension.lstrip('.')
return f"{prefix}{random_hex}_{date_compact}.{extension}"
def parse_date_from_filename(filename: str) -> Optional[str]:

90
src/stegasoo/validation.py
View File

@@ -5,17 +5,17 @@ Validators for all user inputs with clear error messages.
"""
import io
from typing import Optional
from typing import Optional, Union
from PIL import Image
from .constants import (
MIN_PIN_LENGTH, MAX_PIN_LENGTH,
MAX_MESSAGE_SIZE, MAX_IMAGE_PIXELS, MAX_FILE_SIZE,
MAX_MESSAGE_SIZE, MAX_FILE_PAYLOAD_SIZE, MAX_IMAGE_PIXELS, MAX_FILE_SIZE,
MIN_RSA_BITS, MIN_KEY_PASSWORD_LENGTH,
ALLOWED_IMAGE_EXTENSIONS, ALLOWED_KEY_EXTENSIONS,
)
from .models import ValidationResult
from .models import ValidationResult, FilePayload
from .exceptions import (
ValidationError, PinValidationError, MessageValidationError,
ImageValidationError, KeyValidationError, SecurityFactorError,
@@ -61,7 +61,7 @@ def validate_pin(pin: str, required: bool = False) -> ValidationResult:
def validate_message(message: str) -> ValidationResult:
"""
Validate message content and size.
Validate text message content and size.
Args:
message: Message text
@@ -80,6 +80,81 @@ def validate_message(message: str) -> ValidationResult:
return ValidationResult.ok(length=len(message))
def validate_payload(payload: Union[str, bytes, FilePayload]) -> ValidationResult:
"""
Validate a payload (text message, bytes, or file).
Args:
payload: Text string, raw bytes, or FilePayload
Returns:
ValidationResult
"""
if isinstance(payload, str):
return validate_message(payload)
elif isinstance(payload, FilePayload):
if not payload.data:
return ValidationResult.error("File is empty")
if len(payload.data) > MAX_FILE_PAYLOAD_SIZE:
return ValidationResult.error(
f"File too large ({len(payload.data):,} bytes). "
f"Maximum: {MAX_FILE_PAYLOAD_SIZE:,} bytes ({MAX_FILE_PAYLOAD_SIZE // 1024} KB)"
)
return ValidationResult.ok(
size=len(payload.data),
filename=payload.filename,
mime_type=payload.mime_type
)
elif isinstance(payload, bytes):
if not payload:
return ValidationResult.error("Payload is empty")
if len(payload) > MAX_FILE_PAYLOAD_SIZE:
return ValidationResult.error(
f"Payload too large ({len(payload):,} bytes). "
f"Maximum: {MAX_FILE_PAYLOAD_SIZE:,} bytes ({MAX_FILE_PAYLOAD_SIZE // 1024} KB)"
)
return ValidationResult.ok(size=len(payload))
else:
return ValidationResult.error(f"Invalid payload type: {type(payload)}")
def validate_file_payload(
file_data: bytes,
filename: str = "",
max_size: int = MAX_FILE_PAYLOAD_SIZE
) -> ValidationResult:
"""
Validate a file for embedding.
Args:
file_data: Raw file bytes
filename: Original filename (for display in errors)
max_size: Maximum allowed size in bytes
Returns:
ValidationResult
"""
if not file_data:
return ValidationResult.error("File is empty")
if len(file_data) > max_size:
size_kb = len(file_data) / 1024
max_kb = max_size / 1024
return ValidationResult.error(
f"File '{filename or 'unnamed'}' too large ({size_kb:.1f} KB). "
f"Maximum: {max_kb:.0f} KB"
)
return ValidationResult.ok(size=len(file_data), filename=filename)
def validate_image(
image_data: bytes,
name: str = "Image",
@@ -319,6 +394,13 @@ def require_valid_message(message: str) -> None:
raise MessageValidationError(result.error_message)
def require_valid_payload(payload: Union[str, bytes, FilePayload]) -> None:
"""Validate payload (text, bytes, or file), raising exception on failure."""
result = validate_payload(payload)
if not result.is_valid:
raise MessageValidationError(result.error_message)
def require_valid_image(image_data: bytes, name: str = "Image") -> None:
"""Validate image, raising exception on failure."""
result = validate_image(image_data, name)