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

Apply black formatter to all Python files

Browse Source 
Reformatted 29 files for consistent code style and CI compliance.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Aaron D. Lee
2026-01-02 17:44:41 -05:00
parent 221678d934
commit afa88bc73b
29 changed files with 2067 additions and 1814 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main.py
View File

@@ -22,6 +22,7 @@ def main():
"""
try:
from stegasoo.cli import main as cli_main
cli_main()
except ImportError as e:
# Provide helpful error if dependencies are missing
@@ -43,6 +44,7 @@ def version():
"""Print version and exit."""
try:
from stegasoo import __version__
print(f"stegasoo {__version__}")
except ImportError:
print("stegasoo (version unknown)")

13
src/stegasoo/__init__.py
View File

@@ -60,6 +60,7 @@ try:
extract_key_from_qr,
generate_qr_code,
)
HAS_QR_UTILS = True
except ImportError:
HAS_QR_UTILS = False
@@ -151,13 +152,11 @@ DCT_BYTES_PER_PIXEL = 0.125
__all__ = [
# Version
"__version__",
# Core
"encode",
"decode",
"decode_file",
"decode_text",
# Generation
"generate_pin",
"generate_passphrase",
@@ -165,7 +164,6 @@ __all__ = [
"generate_credentials",
"export_rsa_key_pem",
"load_rsa_key",
# Channel key management (v4.0.0)
"generate_channel_key",
"get_channel_key",
@@ -177,28 +175,22 @@ __all__ = [
"format_channel_key",
"get_active_channel_key",
"get_channel_fingerprint",
# Image utilities
"get_image_info",
"compare_capacity",
# Utilities
"generate_filename",
# Crypto
"has_argon2",
# Steganography
"has_dct_support",
"compare_modes",
"will_fit_by_mode",
# QR utilities
"generate_qr_code",
"extract_key_from_qr",
"detect_and_crop_qr",
"HAS_QR_UTILS",
# Validation
"validate_reference_photo",
"validate_carrier",
@@ -212,7 +204,6 @@ __all__ = [
"validate_dct_output_format",
"validate_dct_color_mode",
"validate_channel_key",
# Models
"ImageInfo",
"CapacityComparison",
@@ -222,7 +213,6 @@ __all__ = [
"FilePayload",
"Credentials",
"ValidationResult",
# Exceptions
"StegasooError",
"ValidationError",
@@ -242,7 +232,6 @@ __all__ = [
"ExtractionError",
"EmbeddingError",
"InvalidHeaderError",
# Constants
"FORMAT_VERSION",
"MIN_PASSPHRASE_WORDS",

86
src/stegasoo/batch.py
View File

@@ -23,6 +23,7 @@ from .constants import ALLOWED_IMAGE_EXTENSIONS, LOSSLESS_FORMATS
class BatchStatus(Enum):
"""Status of individual batch items."""
PENDING = "pending"
PROCESSING = "processing"
SUCCESS = "success"
@@ -33,6 +34,7 @@ class BatchStatus(Enum):
@dataclass
class BatchItem:
"""Represents a single item in a batch operation."""
input_path: Path
output_path: Path | None = None
status: BatchStatus = BatchStatus.PENDING
@@ -84,6 +86,7 @@ class BatchCredentials:
)
result = processor.batch_encode(images, creds, message="secret")
"""
reference_photo: bytes
passphrase: str # v3.2.0: renamed from day_phrase
pin: str = ""
@@ -101,27 +104,28 @@ class BatchCredentials:
}
@classmethod
def from_dict(cls, data: dict) -> 'BatchCredentials':
def from_dict(cls, data: dict) -> "BatchCredentials":
"""
Create BatchCredentials from a dictionary.
Handles both v3.2.0 format (passphrase) and legacy formats (day_phrase, phrase).
"""
# Handle legacy 'day_phrase' and 'phrase' keys
passphrase = data.get('passphrase') or data.get('day_phrase') or data.get('phrase', '')
passphrase = data.get("passphrase") or data.get("day_phrase") or data.get("phrase", "")
return cls(
reference_photo=data['reference_photo'],
reference_photo=data["reference_photo"],
passphrase=passphrase,
pin=data.get('pin', ''),
rsa_key_data=data.get('rsa_key_data'),
rsa_password=data.get('rsa_password'),
pin=data.get("pin", ""),
rsa_key_data=data.get("rsa_key_data"),
rsa_password=data.get("rsa_password"),
)
@dataclass
class BatchResult:
"""Summary of a batch operation."""
operation: str
total: int = 0
succeeded: int = 0
@@ -232,18 +236,17 @@ class BatchProcessor:
yield path
elif path.is_dir():
pattern = '**/*' if recursive else '*'
pattern = "**/*" if recursive else "*"
for file_path in path.glob(pattern):
if file_path.is_file() and self._is_valid_image(file_path):
yield file_path
def _is_valid_image(self, path: Path) -> bool:
"""Check if path is a valid image file."""
return path.suffix.lower().lstrip('.') in ALLOWED_IMAGE_EXTENSIONS
return path.suffix.lower().lstrip(".") in ALLOWED_IMAGE_EXTENSIONS
def _normalize_credentials(
self,
credentials: dict | BatchCredentials | None
self, credentials: dict | BatchCredentials | None
) -> BatchCredentials:
"""
Normalize credentials to BatchCredentials object.
@@ -341,7 +344,11 @@ class BatchProcessor:
self._do_encode(item, message, file_payload, creds, compress)
item.status = BatchStatus.SUCCESS
item.output_size = item.output_path.stat().st_size if item.output_path and item.output_path.exists() else 0
item.output_size = (
item.output_path.stat().st_size
if item.output_path and item.output_path.exists()
else 0
)
item.message = f"Encoded to {item.output_path.name}"
except Exception as e:
@@ -412,7 +419,9 @@ class BatchProcessor:
output_dir=item.output_path,
credentials=creds.to_dict(),
)
item.message = decoded.get('message', '') if isinstance(decoded, dict) else str(decoded)
item.message = (
decoded.get("message", "") if isinstance(decoded, dict) else str(decoded)
)
else:
# Use stegasoo decode
item.message = self._do_decode(item, creds)
@@ -441,10 +450,7 @@ class BatchProcessor:
completed = 0
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(process_func, item): item
for item in result.items
}
futures = {executor.submit(process_func, item): item for item in result.items}
for future in as_completed(futures):
item = future.result()
@@ -469,7 +475,7 @@ class BatchProcessor:
message: str | None,
file_payload: Path | None,
creds: BatchCredentials,
compress: bool
compress: bool,
) -> None:
"""
Perform actual encoding using stegasoo.encode.
@@ -555,16 +561,13 @@ class BatchProcessor:
return self._mock_decode(item, creds)
def _mock_encode(
self,
item: BatchItem,
message: str,
creds: BatchCredentials,
compress: bool
self, item: BatchItem, message: str, creds: BatchCredentials, compress: bool
) -> None:
"""Mock encode for testing - replace with actual stego.encode()"""
# This is a placeholder - in real usage, you'd call your actual encode function
# For now, just copy the file to simulate encoding
import shutil
if item.output_path:
shutil.copy(item.input_path, item.output_path)
@@ -605,23 +608,27 @@ def batch_capacity_check(
capacity_bits = pixels * 3
capacity_bytes = (capacity_bits // 8) - 100 # Header overhead
results.append({
"path": str(img_path),
"dimensions": f"{width}x{height}",
"pixels": pixels,
"format": img.format,
"mode": img.mode,
"capacity_bytes": max(0, capacity_bytes),
"capacity_kb": max(0, capacity_bytes // 1024),
"valid": pixels <= MAX_IMAGE_PIXELS and img.format in LOSSLESS_FORMATS,
"warnings": _get_image_warnings(img, img_path),
})
results.append(
{
"path": str(img_path),
"dimensions": f"{width}x{height}",
"pixels": pixels,
"format": img.format,
"mode": img.mode,
"capacity_bytes": max(0, capacity_bytes),
"capacity_kb": max(0, capacity_bytes // 1024),
"valid": pixels <= MAX_IMAGE_PIXELS and img.format in LOSSLESS_FORMATS,
"warnings": _get_image_warnings(img, img_path),
}
)
except Exception as e:
results.append({
"path": str(img_path),
"error": str(e),
"valid": False,
})
results.append(
{
"path": str(img_path),
"error": str(e),
"valid": False,
}
)
return results
@@ -638,7 +645,7 @@ def _get_image_warnings(img, path: Path) -> list[str]:
if img.size[0] * img.size[1] > MAX_IMAGE_PIXELS:
warnings.append(f"Image exceeds {MAX_IMAGE_PIXELS:,} pixel limit")
if img.mode not in ('RGB', 'RGBA'):
if img.mode not in ("RGB", "RGBA"):
warnings.append(f"Non-RGB mode ({img.mode}) - will be converted")
return warnings
@@ -646,6 +653,7 @@ def _get_image_warnings(img, path: Path) -> list[str]:
# CLI-friendly functions
def print_batch_result(result: BatchResult, verbose: bool = False) -> None:
"""Print batch result summary to console."""
print(f"\n{'='*60}")

98
src/stegasoo/channel.py
View File

@@ -34,17 +34,17 @@ from .debug import debug
# Channel key format: 8 groups of 4 alphanumeric chars (32 chars total)
# Example: ABCD-1234-EFGH-5678-IJKL-9012-MNOP-3456
CHANNEL_KEY_PATTERN = re.compile(r'^[A-Z0-9]{4}(-[A-Z0-9]{4}){7}$')
CHANNEL_KEY_PATTERN = re.compile(r"^[A-Z0-9]{4}(-[A-Z0-9]{4}){7}$")
CHANNEL_KEY_LENGTH = 32 # Characters (excluding dashes)
CHANNEL_KEY_FORMATTED_LENGTH = 39 # With dashes
# Environment variable name
CHANNEL_KEY_ENV_VAR = 'STEGASOO_CHANNEL_KEY'
CHANNEL_KEY_ENV_VAR = "STEGASOO_CHANNEL_KEY"
# Config locations (in priority order)
CONFIG_LOCATIONS = [
Path('./config/channel.key'), # Project config
Path.home() / '.stegasoo' / 'channel.key', # User config
Path("./config/channel.key"), # Project config
Path.home() / ".stegasoo" / "channel.key", # User config
]
@@ -61,8 +61,8 @@ def generate_channel_key() -> str:
39
"""
# Generate 32 random alphanumeric characters
alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
raw_key = ''.join(secrets.choice(alphabet) for _ in range(CHANNEL_KEY_LENGTH))
alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
raw_key = "".join(secrets.choice(alphabet) for _ in range(CHANNEL_KEY_LENGTH))
formatted = format_channel_key(raw_key)
debug.print(f"Generated channel key: {get_channel_fingerprint(formatted)}")
@@ -87,19 +87,17 @@ def format_channel_key(raw_key: str) -> str:
"ABCD-1234-EFGH-5678-IJKL-9012-MNOP-3456"
"""
# Remove any existing dashes, spaces, and convert to uppercase
clean = raw_key.replace('-', '').replace(' ', '').upper()
clean = raw_key.replace("-", "").replace(" ", "").upper()
if len(clean) != CHANNEL_KEY_LENGTH:
raise ValueError(
f"Channel key must be {CHANNEL_KEY_LENGTH} characters (got {len(clean)})"
)
raise ValueError(f"Channel key must be {CHANNEL_KEY_LENGTH} characters (got {len(clean)})")
# Validate characters
if not all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789' for c in clean):
if not all(c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" for c in clean):
raise ValueError("Channel key must contain only letters A-Z and digits 0-9")
# Format with dashes every 4 characters
return '-'.join(clean[i:i+4] for i in range(0, CHANNEL_KEY_LENGTH, 4))
return "-".join(clean[i : i + 4] for i in range(0, CHANNEL_KEY_LENGTH, 4))
def validate_channel_key(key: str) -> bool:
@@ -148,7 +146,7 @@ def get_channel_key() -> str | None:
... print("Public mode")
"""
# 1. Check environment variable
env_key = os.environ.get(CHANNEL_KEY_ENV_VAR, '').strip()
env_key = os.environ.get(CHANNEL_KEY_ENV_VAR, "").strip()
if env_key:
if validate_channel_key(env_key):
debug.print(f"Channel key from environment: {get_channel_fingerprint(env_key)}")
@@ -173,7 +171,7 @@ def get_channel_key() -> str | None:
return None
def set_channel_key(key: str, location: str = 'project') -> Path:
def set_channel_key(key: str, location: str = "project") -> Path:
"""
Save a channel key to config file.
@@ -194,16 +192,16 @@ def set_channel_key(key: str, location: str = 'project') -> Path:
"""
formatted = format_channel_key(key)
if location == 'user':
config_path = Path.home() / '.stegasoo' / 'channel.key'
if location == "user":
config_path = Path.home() / ".stegasoo" / "channel.key"
else:
config_path = Path('./config/channel.key')
config_path = Path("./config/channel.key")
# Create directory if needed
config_path.parent.mkdir(parents=True, exist_ok=True)
# Write key with newline
config_path.write_text(formatted + '\n')
config_path.write_text(formatted + "\n")
# Set restrictive permissions (owner read/write only)
try:
@@ -215,7 +213,7 @@ def set_channel_key(key: str, location: str = 'project') -> Path:
return config_path
def clear_channel_key(location: str = 'all') -> list[Path]:
def clear_channel_key(location: str = "all") -> list[Path]:
"""
Remove channel key configuration.
@@ -232,10 +230,10 @@ def clear_channel_key(location: str = 'all') -> list[Path]:
deleted = []
paths_to_check = []
if location in ('project', 'all'):
paths_to_check.append(Path('./config/channel.key'))
if location in ('user', 'all'):
paths_to_check.append(Path.home() / '.stegasoo' / 'channel.key')
if location in ("project", "all"):
paths_to_check.append(Path("./config/channel.key"))
if location in ("user", "all"):
paths_to_check.append(Path.home() / ".stegasoo" / "channel.key")
for path in paths_to_check:
if path.exists():
@@ -275,7 +273,7 @@ def get_channel_key_hash(key: str | None = None) -> bytes | None:
# Hash the formatted key to get consistent 32 bytes
formatted = format_channel_key(key)
return hashlib.sha256(formatted.encode('utf-8')).digest()
return hashlib.sha256(formatted.encode("utf-8")).digest()
def get_channel_fingerprint(key: str | None = None) -> str | None:
@@ -300,11 +298,11 @@ def get_channel_fingerprint(key: str | None = None) -> str | None:
return None
formatted = format_channel_key(key)
parts = formatted.split('-')
parts = formatted.split("-")
# Show first and last group, mask the rest
masked = [parts[0]] + ['••••'] * 6 + [parts[-1]]
return '-'.join(masked)
masked = [parts[0]] + ["••••"] * 6 + [parts[-1]]
return "-".join(masked)
def get_channel_status() -> dict:
@@ -328,10 +326,10 @@ def get_channel_status() -> dict:
if key:
# Find which source provided the key
source = 'unknown'
env_key = os.environ.get(CHANNEL_KEY_ENV_VAR, '').strip()
source = "unknown"
env_key = os.environ.get(CHANNEL_KEY_ENV_VAR, "").strip()
if env_key and validate_channel_key(env_key):
source = 'environment'
source = "environment"
else:
for config_path in CONFIG_LOCATIONS:
if config_path.exists():
@@ -344,19 +342,19 @@ def get_channel_status() -> dict:
continue
return {
'mode': 'private',
'configured': True,
'fingerprint': get_channel_fingerprint(key),
'source': source,
'key': key,
"mode": "private",
"configured": True,
"fingerprint": get_channel_fingerprint(key),
"source": source,
"key": key,
}
else:
return {
'mode': 'public',
'configured': False,
'fingerprint': None,
'source': None,
'key': None,
"mode": "public",
"configured": False,
"fingerprint": None,
"source": None,
"key": None,
}
@@ -378,14 +376,14 @@ def has_channel_key() -> bool:
# CLI SUPPORT
# =============================================================================
if __name__ == '__main__':
if __name__ == "__main__":
import sys
def print_status():
"""Print current channel status."""
status = get_channel_status()
print(f"Mode: {status['mode'].upper()}")
if status['configured']:
if status["configured"]:
print(f"Fingerprint: {status['fingerprint']}")
print(f"Source: {status['source']}")
else:
@@ -406,17 +404,17 @@ if __name__ == '__main__':
cmd = sys.argv[1].lower()
if cmd == 'generate':
if cmd == "generate":
key = generate_channel_key()
print("Generated channel key:")
print(f" {key}")
print()
save = input("Save to config? [y/N]: ").strip().lower()
if save == 'y':
if save == "y":
path = set_channel_key(key)
print(f"Saved to: {path}")
elif cmd == 'set':
elif cmd == "set":
if len(sys.argv) < 3:
print("Usage: python -m stegasoo.channel set <KEY>")
sys.exit(1)
@@ -431,22 +429,22 @@ if __name__ == '__main__':
print(f"Error: {e}")
sys.exit(1)
elif cmd == 'show':
elif cmd == "show":
status = get_channel_status()
if status['configured']:
if status["configured"]:
print(f"Channel key: {status['key']}")
print(f"Source: {status['source']}")
else:
print("No channel key configured")
elif cmd == 'clear':
deleted = clear_channel_key('all')
elif cmd == "clear":
deleted = clear_channel_key("all")
if deleted:
print(f"Removed channel key from: {', '.join(str(p) for p in deleted)}")
else:
print("No channel key files found")
elif cmd == 'status':
elif cmd == "status":
print_status()
else:

257
src/stegasoo/cli.py
View File

@@ -33,12 +33,12 @@ from .constants import (
)
# Click context settings
CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
@click.group(context_settings=CONTEXT_SETTINGS)
@click.version_option(__version__, '-v', '--version')
@click.option('--json', 'json_output', is_flag=True, help='Output results as JSON')
@click.version_option(__version__, "-v", "--version")
@click.option("--json", "json_output", is_flag=True, help="Output results as JSON")
@click.pass_context
def cli(ctx, json_output):
"""
@@ -47,31 +47,47 @@ def cli(ctx, json_output):
Hide messages in images using PIN + passphrase security.
"""
ctx.ensure_object(dict)
ctx.obj['json'] = json_output
ctx.obj["json"] = json_output
# =============================================================================
# ENCODE COMMANDS
# =============================================================================
@cli.command()
@click.argument('image', type=click.Path(exists=True))
@click.option('-m', '--message', help='Message to encode')
@click.option('-f', '--file', 'file_payload', type=click.Path(exists=True),
help='File to embed instead of message')
@click.option('-o', '--output', type=click.Path(), help='Output image path')
@click.option('--passphrase', prompt=True, hide_input=True,
confirmation_prompt=True, help='Passphrase (recommend 4+ words)')
@click.option('--pin', prompt=True, hide_input=True,
confirmation_prompt=True, help='PIN code')
@click.option('--compress/--no-compress', default=True,
help='Enable/disable compression (default: enabled)')
@click.option('--algorithm', type=click.Choice(['zlib', 'lz4', 'none']),
default='zlib', help='Compression algorithm')
@click.option('--dry-run', is_flag=True, help='Show capacity usage without encoding')
@click.argument("image", type=click.Path(exists=True))
@click.option("-m", "--message", help="Message to encode")
@click.option(
"-f",
"--file",
"file_payload",
type=click.Path(exists=True),
help="File to embed instead of message",
)
@click.option("-o", "--output", type=click.Path(), help="Output image path")
@click.option(
"--passphrase",
prompt=True,
hide_input=True,
confirmation_prompt=True,
help="Passphrase (recommend 4+ words)",
)
@click.option("--pin", prompt=True, hide_input=True, confirmation_prompt=True, help="PIN code")
@click.option(
"--compress/--no-compress", default=True, help="Enable/disable compression (default: enabled)"
)
@click.option(
"--algorithm",
type=click.Choice(["zlib", "lz4", "none"]),
default="zlib",
help="Compression algorithm",
)
@click.option("--dry-run", is_flag=True, help="Show capacity usage without encoding")
@click.pass_context
def encode(ctx, image, message, file_payload, output, passphrase, pin,
compress, algorithm, dry_run):
def encode(
ctx, image, message, file_payload, output, passphrase, pin, compress, algorithm, dry_run
):
"""
Encode a message or file into an image.
@@ -88,13 +104,13 @@ def encode(ctx, image, message, file_payload, output, passphrase, pin,
# Parse compression algorithm
algo_map = {
'zlib': CompressionAlgorithm.ZLIB,
'lz4': CompressionAlgorithm.LZ4,
'none': CompressionAlgorithm.NONE,
"zlib": CompressionAlgorithm.ZLIB,
"lz4": CompressionAlgorithm.LZ4,
"none": CompressionAlgorithm.NONE,
}
compression_algo = algo_map[algorithm] if compress else CompressionAlgorithm.NONE
if algorithm == 'lz4' and not HAS_LZ4:
if algorithm == "lz4" and not HAS_LZ4:
click.echo("Warning: LZ4 not available, falling back to zlib", err=True)
compression_algo = CompressionAlgorithm.ZLIB
@@ -103,7 +119,7 @@ def encode(ctx, image, message, file_payload, output, passphrase, pin,
payload_size = Path(file_payload).stat().st_size
payload_type = "file"
else:
payload_size = len(message.encode('utf-8'))
payload_size = len(message.encode("utf-8"))
payload_type = "text"
# Get image capacity
@@ -123,7 +139,7 @@ def encode(ctx, image, message, file_payload, output, passphrase, pin,
"fits": payload_size < capacity_bytes,
}
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(json.dumps(result, indent=2))
else:
click.echo(f"Image: {image} ({width}x{height})")
@@ -138,25 +154,29 @@ def encode(ctx, image, message, file_payload, output, passphrase, pin,
# For now, show what would be done
output = output or f"{Path(image).stem}_encoded.png"
if ctx.obj.get('json'):
click.echo(json.dumps({
"status": "success",
"input": image,
"output": output,
"payload_type": payload_type,
"compression": algorithm_name(compression_algo),
}, indent=2))
if ctx.obj.get("json"):
click.echo(
json.dumps(
{
"status": "success",
"input": image,
"output": output,
"payload_type": payload_type,
"compression": algorithm_name(compression_algo),
},
indent=2,
)
)
else:
click.echo(f"✓ Encoded {payload_type} to {output}")
click.echo(f" Compression: {algorithm_name(compression_algo)}")
@cli.command()
@click.argument('image', type=click.Path(exists=True))
@click.option('--passphrase', prompt=True, hide_input=True, help='Passphrase')
@click.option('--pin', prompt=True, hide_input=True, help='PIN code')
@click.option('-o', '--output', type=click.Path(),
help='Output path for file payloads')
@click.argument("image", type=click.Path(exists=True))
@click.option("--passphrase", prompt=True, hide_input=True, help="Passphrase")
@click.option("--pin", prompt=True, hide_input=True, help="PIN code")
@click.option("-o", "--output", type=click.Path(), help="Output path for file payloads")
@click.pass_context
def decode(ctx, image, passphrase, pin, output):
"""
@@ -176,46 +196,68 @@ def decode(ctx, image, passphrase, pin, output):
"message": "[Decoded message would appear here]",
}
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(json.dumps(result, indent=2))
else:
click.echo(f"Decoded from {image}:")
click.echo(result['message'])
click.echo(result["message"])
# =============================================================================
# BATCH COMMANDS
# =============================================================================
@cli.group()
def batch():
"""Batch operations on multiple images."""
pass
@batch.command('encode')
@click.argument('images', nargs=-1, required=True, type=click.Path(exists=True))
@click.option('-m', '--message', help='Message to encode in all images')
@click.option('-f', '--file', 'file_payload', type=click.Path(exists=True),
help='File to embed in all images')
@click.option('-o', '--output-dir', type=click.Path(),
help='Output directory (default: same as input)')
@click.option('--suffix', default='_encoded', help='Output filename suffix')
@click.option('--passphrase', prompt=True, hide_input=True,
confirmation_prompt=True, help='Passphrase (recommend 4+ words)')
@click.option('--pin', prompt=True, hide_input=True,
confirmation_prompt=True, help='PIN code')
@click.option('--compress/--no-compress', default=True,
help='Enable/disable compression')
@click.option('--algorithm', type=click.Choice(['zlib', 'lz4', 'none']),
default='zlib', help='Compression algorithm')
@click.option('-r', '--recursive', is_flag=True,
help='Search directories recursively')
@click.option('-j', '--jobs', default=4, help='Parallel workers (default: 4)')
@click.option('-v', '--verbose', is_flag=True, help='Show detailed output')
@batch.command("encode")
@click.argument("images", nargs=-1, required=True, type=click.Path(exists=True))
@click.option("-m", "--message", help="Message to encode in all images")
@click.option(
"-f", "--file", "file_payload", type=click.Path(exists=True), help="File to embed in all images"
)
@click.option(
"-o", "--output-dir", type=click.Path(), help="Output directory (default: same as input)"
)
@click.option("--suffix", default="_encoded", help="Output filename suffix")
@click.option(
"--passphrase",
prompt=True,
hide_input=True,
confirmation_prompt=True,
help="Passphrase (recommend 4+ words)",
)
@click.option("--pin", prompt=True, hide_input=True, confirmation_prompt=True, help="PIN code")
@click.option("--compress/--no-compress", default=True, help="Enable/disable compression")
@click.option(
"--algorithm",
type=click.Choice(["zlib", "lz4", "none"]),
default="zlib",
help="Compression algorithm",
)
@click.option("-r", "--recursive", is_flag=True, help="Search directories recursively")
@click.option("-j", "--jobs", default=4, help="Parallel workers (default: 4)")
@click.option("-v", "--verbose", is_flag=True, help="Show detailed output")
@click.pass_context
def batch_encode(ctx, images, message, file_payload, output_dir, suffix,
passphrase, pin, compress, algorithm, recursive, jobs, verbose):
def batch_encode(
ctx,
images,
message,
file_payload,
output_dir,
suffix,
passphrase,
pin,
compress,
algorithm,
recursive,
jobs,
verbose,
):
"""
Encode message into multiple images.
@@ -232,7 +274,7 @@ def batch_encode(ctx, images, message, file_payload, output_dir, suffix,
# Progress callback
def progress(current, total, item):
if not ctx.obj.get('json'):
if not ctx.obj.get("json"):
status = "✓" if item.status.value == "success" else "✗"
click.echo(f"[{current}/{total}] {status} {item.input_path.name}")
@@ -248,25 +290,23 @@ def batch_encode(ctx, images, message, file_payload, output_dir, suffix,
credentials=credentials,
compress=compress,
recursive=recursive,
progress_callback=progress if not ctx.obj.get('json') else None,
progress_callback=progress if not ctx.obj.get("json") else None,
)
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(result.to_json())
else:
print_batch_result(result, verbose)
@batch.command('decode')
@click.argument('images', nargs=-1, required=True, type=click.Path(exists=True))
@click.option('-o', '--output-dir', type=click.Path(),
help='Output directory for file payloads')
@click.option('--passphrase', prompt=True, hide_input=True, help='Passphrase')
@click.option('--pin', prompt=True, hide_input=True, help='PIN code')
@click.option('-r', '--recursive', is_flag=True,
help='Search directories recursively')
@click.option('-j', '--jobs', default=4, help='Parallel workers (default: 4)')
@click.option('-v', '--verbose', is_flag=True, help='Show detailed output')
@batch.command("decode")
@click.argument("images", nargs=-1, required=True, type=click.Path(exists=True))
@click.option("-o", "--output-dir", type=click.Path(), help="Output directory for file payloads")
@click.option("--passphrase", prompt=True, hide_input=True, help="Passphrase")
@click.option("--pin", prompt=True, hide_input=True, help="PIN code")
@click.option("-r", "--recursive", is_flag=True, help="Search directories recursively")
@click.option("-j", "--jobs", default=4, help="Parallel workers (default: 4)")
@click.option("-v", "--verbose", is_flag=True, help="Show detailed output")
@click.pass_context
def batch_decode(ctx, images, output_dir, passphrase, pin, recursive, jobs, verbose):
"""
@@ -282,7 +322,7 @@ def batch_decode(ctx, images, output_dir, passphrase, pin, recursive, jobs, verb
# Progress callback
def progress(current, total, item):
if not ctx.obj.get('json'):
if not ctx.obj.get("json"):
status = "✓" if item.status.value == "success" else "✗"
click.echo(f"[{current}/{total}] {status} {item.input_path.name}")
@@ -294,19 +334,18 @@ def batch_decode(ctx, images, output_dir, passphrase, pin, recursive, jobs, verb
output_dir=Path(output_dir) if output_dir else None,
credentials=credentials,
recursive=recursive,
progress_callback=progress if not ctx.obj.get('json') else None,
progress_callback=progress if not ctx.obj.get("json") else None,
)
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(result.to_json())
else:
print_batch_result(result, verbose)
@batch.command('check')
@click.argument('images', nargs=-1, required=True, type=click.Path(exists=True))
@click.option('-r', '--recursive', is_flag=True,
help='Search directories recursively')
@batch.command("check")
@click.argument("images", nargs=-1, required=True, type=click.Path(exists=True))
@click.option("-r", "--recursive", is_flag=True, help="Search directories recursively")
@click.pass_context
def batch_check(ctx, images, recursive):
"""
@@ -320,22 +359,22 @@ def batch_check(ctx, images, recursive):
"""
results = batch_capacity_check(list(images), recursive)
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(json.dumps(results, indent=2))
else:
click.echo(f"{'Image':<40} {'Size':<12} {'Capacity':<12} {'Status'}")
click.echo("─" * 80)
for item in results:
if 'error' in item:
if "error" in item:
click.echo(f"{Path(item['path']).name:<40} {'ERROR':<12} {'':<12} {item['error']}")
else:
name = Path(item['path']).name
name = Path(item["path"]).name
if len(name) > 38:
name = name[:35] + "..."
status = "✓" if item['valid'] else "⚠"
warnings = ", ".join(item.get('warnings', []))
status = "✓" if item["valid"] else "⚠"
warnings = ", ".join(item.get("warnings", []))
click.echo(
f"{name:<40} "
@@ -349,11 +388,16 @@ def batch_check(ctx, images, recursive):
# UTILITY COMMANDS
# =============================================================================
@cli.command()
@click.option('--words', default=DEFAULT_PASSPHRASE_WORDS,
help=f'Number of words in passphrase (default: {DEFAULT_PASSPHRASE_WORDS})')
@click.option('--pin-length', default=DEFAULT_PIN_LENGTH,
help=f'PIN length (default: {DEFAULT_PIN_LENGTH})')
@click.option(
"--words",
default=DEFAULT_PASSPHRASE_WORDS,
help=f"Number of words in passphrase (default: {DEFAULT_PASSPHRASE_WORDS})",
)
@click.option(
"--pin-length", default=DEFAULT_PIN_LENGTH, help=f"PIN length (default: {DEFAULT_PIN_LENGTH})"
)
@click.pass_context
def generate(ctx, words, pin_length):
"""
@@ -368,24 +412,37 @@ def generate(ctx, words, pin_length):
import secrets
# Generate PIN
pin = ''.join(str(secrets.randbelow(10)) for _ in range(pin_length))
pin = "".join(str(secrets.randbelow(10)) for _ in range(pin_length))
# Ensure PIN doesn't start with 0
if pin[0] == '0':
if pin[0] == "0":
pin = str(secrets.randbelow(9) + 1) + pin[1:]
# Generate passphrase (would use BIP-39 wordlist)
# Placeholder - actual implementation uses constants.get_wordlist()
try:
from .constants import get_wordlist
wordlist = get_wordlist()
phrase_words = [secrets.choice(wordlist) for _ in range(words)]
except (ImportError, FileNotFoundError):
# Fallback for testing
sample_words = ['alpha', 'bravo', 'charlie', 'delta', 'echo', 'foxtrot',
'golf', 'hotel', 'india', 'juliet', 'kilo', 'lima']
sample_words = [
"alpha",
"bravo",
"charlie",
"delta",
"echo",
"foxtrot",
"golf",
"hotel",
"india",
"juliet",
"kilo",
"lima",
]
phrase_words = [secrets.choice(sample_words) for _ in range(words)]
passphrase = ' '.join(phrase_words)
passphrase = " ".join(phrase_words)
result = {
"passphrase": passphrase,
@@ -394,7 +451,7 @@ def generate(ctx, words, pin_length):
"pin_length": pin_length,
}
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(json.dumps(result, indent=2))
else:
click.echo(f"Passphrase: {passphrase}")
@@ -418,7 +475,7 @@ def info(ctx):
},
}
if ctx.obj.get('json'):
if ctx.obj.get("json"):
click.echo(json.dumps(info_data, indent=2))
else:
click.echo(f"Stegasoo v{__version__}")
@@ -437,5 +494,5 @@ def main():
cli(obj={})
if __name__ == '__main__':
if __name__ == "__main__":
main()

19
src/stegasoo/compression.py
View File

@@ -12,6 +12,7 @@ from enum import IntEnum
# Optional LZ4 support (faster, slightly worse ratio)
try:
import lz4.frame
HAS_LZ4 = True
except ImportError:
HAS_LZ4 = False
@@ -19,13 +20,14 @@ except ImportError:
class CompressionAlgorithm(IntEnum):
"""Supported compression algorithms."""
NONE = 0
ZLIB = 1
LZ4 = 2
# Magic bytes for compressed payloads
COMPRESSION_MAGIC = b'\x00CMP'
COMPRESSION_MAGIC = b"\x00CMP"
# Minimum size to bother compressing (small data often expands)
MIN_COMPRESS_SIZE = 64
@@ -36,6 +38,7 @@ ZLIB_LEVEL = 6
class CompressionError(Exception):
"""Raised when compression/decompression fails."""
pass
@@ -77,7 +80,7 @@ def compress(data: bytes, algorithm: CompressionAlgorithm = CompressionAlgorithm
return _wrap_uncompressed(data)
# Build header: MAGIC + algorithm + original_size + compressed_data
header = COMPRESSION_MAGIC + struct.pack('<BI', algorithm, len(data))
header = COMPRESSION_MAGIC + struct.pack("<BI", algorithm, len(data))
return header + compressed
@@ -101,7 +104,7 @@ def decompress(data: bytes) -> bytes:
# Parse header
algorithm = CompressionAlgorithm(data[4])
original_size = struct.unpack('<I', data[5:9])[0]
original_size = struct.unpack("<I", data[5:9])[0]
compressed_data = data[9:]
if algorithm == CompressionAlgorithm.NONE:
@@ -125,16 +128,14 @@ def decompress(data: bytes) -> bytes:
# Verify size
if len(result) != original_size:
raise CompressionError(
f"Size mismatch: expected {original_size}, got {len(result)}"
)
raise CompressionError(f"Size mismatch: expected {original_size}, got {len(result)}")
return result
def _wrap_uncompressed(data: bytes) -> bytes:
"""Wrap uncompressed data with header for consistency."""
header = COMPRESSION_MAGIC + struct.pack('<BI', CompressionAlgorithm.NONE, len(data))
header = COMPRESSION_MAGIC + struct.pack("<BI", CompressionAlgorithm.NONE, len(data))
return header + data
@@ -150,7 +151,9 @@ def get_compression_ratio(original: bytes, compressed: bytes) -> float:
return len(compressed) / len(original)
def estimate_compressed_size(data: bytes, algorithm: CompressionAlgorithm = CompressionAlgorithm.ZLIB) -> int:
def estimate_compressed_size(
data: bytes, algorithm: CompressionAlgorithm = CompressionAlgorithm.ZLIB
) -> int:
"""
Estimate compressed size without full compression.
Uses sampling for large data.

61
src/stegasoo/constants.py
View File

@@ -26,7 +26,7 @@ __version__ = "4.0.1"
# FILE FORMAT
# ============================================================================
MAGIC_HEADER = b'\x89ST3'
MAGIC_HEADER = b"\x89ST3"
# FORMAT VERSION HISTORY:
# Version 1-3: Date-dependent encryption (v3.0.x - v3.1.x)
@@ -58,21 +58,21 @@ PBKDF2_ITERATIONS = 600000
# INPUT LIMITS
# ============================================================================
MAX_IMAGE_PIXELS = 24_000_000 # ~24 megapixels
MIN_IMAGE_PIXELS = 256 * 256 # Minimum viable image size
MAX_IMAGE_PIXELS = 24_000_000 # ~24 megapixels
MIN_IMAGE_PIXELS = 256 * 256 # Minimum viable image size
MAX_MESSAGE_SIZE = 250_000 # 250 KB (text messages)
MAX_MESSAGE_CHARS = 250_000 # Alias for clarity in templates
MIN_MESSAGE_LENGTH = 1 # Minimum message length
MAX_MESSAGE_SIZE = 250_000 # 250 KB (text messages)
MAX_MESSAGE_CHARS = 250_000 # Alias for clarity in templates
MIN_MESSAGE_LENGTH = 1 # Minimum message length
MAX_MESSAGE_LENGTH = MAX_MESSAGE_SIZE # Alias for consistency
MAX_PAYLOAD_SIZE = MAX_MESSAGE_SIZE # Maximum payload size (alias)
MAX_FILENAME_LENGTH = 255 # Max filename length to store
MAX_FILENAME_LENGTH = 255 # Max filename length to store
# File size limits
MAX_FILE_SIZE = 30 * 1024 * 1024 # 30MB total file size
MAX_FILE_SIZE = 30 * 1024 * 1024 # 30MB total file size
MAX_FILE_PAYLOAD_SIZE = 2 * 1024 * 1024 # 2MB payload
MAX_UPLOAD_SIZE = 30 * 1024 * 1024 # 30MB max upload (Flask)
MAX_UPLOAD_SIZE = 30 * 1024 * 1024 # 30MB max upload (Flask)
# PIN configuration
MIN_PIN_LENGTH = 6
@@ -119,11 +119,11 @@ QR_CROP_MIN_PADDING_PX = 10 # Minimum padding in pixels
# FILE TYPES
# ============================================================================
ALLOWED_IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'bmp', 'gif'}
ALLOWED_KEY_EXTENSIONS = {'pem', 'key'}
ALLOWED_IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "bmp", "gif"}
ALLOWED_KEY_EXTENSIONS = {"pem", "key"}
# Lossless image formats (safe for steganography)
LOSSLESS_FORMATS = {'PNG', 'BMP', 'TIFF'}
LOSSLESS_FORMATS = {"PNG", "BMP", "TIFF"}
# Supported image formats for steganography
SUPPORTED_IMAGE_FORMATS = LOSSLESS_FORMATS
@@ -132,7 +132,7 @@ SUPPORTED_IMAGE_FORMATS = LOSSLESS_FORMATS
# DAYS (kept for organizational/UI purposes, not crypto)
# ============================================================================
DAY_NAMES = ('Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday')
DAY_NAMES = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
# ============================================================================
# COMPRESSION
@@ -145,7 +145,7 @@ MIN_COMPRESS_SIZE = 64
ZLIB_COMPRESSION_LEVEL = 6
# Compression header magic bytes
COMPRESSION_MAGIC = b'\x00CMP'
COMPRESSION_MAGIC = b"\x00CMP"
# ============================================================================
# BATCH PROCESSING
@@ -164,6 +164,7 @@ BATCH_OUTPUT_SUFFIX = "_encoded"
# DATA FILES
# ============================================================================
def get_data_dir() -> Path:
"""Get the data directory path."""
# Check multiple locations
@@ -172,12 +173,12 @@ def get_data_dir() -> Path:
# .parent.parent = src/
# .parent.parent.parent = project root (where data/ lives)
candidates = [
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
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:
@@ -190,7 +191,7 @@ def get_data_dir() -> Path:
def get_bip39_words() -> list[str]:
"""Load BIP-39 wordlist."""
wordlist_path = get_data_dir() / 'bip39-words.txt'
wordlist_path = get_data_dir() / "bip39-words.txt"
if not wordlist_path.exists():
raise FileNotFoundError(
@@ -219,14 +220,14 @@ def get_wordlist() -> list[str]:
# =============================================================================
# Embedding modes
EMBED_MODE_LSB = 'lsb' # Spatial LSB embedding (default, original mode)
EMBED_MODE_DCT = 'dct' # DCT domain embedding (new in v3.0)
EMBED_MODE_AUTO = 'auto' # Auto-detect on decode
EMBED_MODE_LSB = "lsb" # Spatial LSB embedding (default, original mode)
EMBED_MODE_DCT = "dct" # DCT domain embedding (new in v3.0)
EMBED_MODE_AUTO = "auto" # Auto-detect on decode
# DCT-specific constants
DCT_MAGIC_HEADER = b'\x89DCT' # Magic header for DCT mode
DCT_MAGIC_HEADER = b"\x89DCT" # Magic header for DCT mode
DCT_FORMAT_VERSION = 1
DCT_STEP_SIZE = 8 # QIM quantization step
DCT_STEP_SIZE = 8 # QIM quantization step
# Valid embedding modes
VALID_EMBED_MODES = {EMBED_MODE_LSB, EMBED_MODE_DCT}
@@ -247,13 +248,13 @@ def detect_stego_mode(encrypted_data: bytes) -> str:
'lsb' or 'dct' or 'unknown'
"""
if len(encrypted_data) < 4:
return 'unknown'
return "unknown"
header = encrypted_data[:4]
if header == b'\x89ST3':
if header == b"\x89ST3":
return EMBED_MODE_LSB
elif header == b'\x89DCT':
elif header == b"\x89DCT":
return EMBED_MODE_DCT
else:
return 'unknown'
return "unknown"

123
src/stegasoo/crypto.py
View File

@@ -44,6 +44,7 @@ from .models import DecodeResult, FilePayload
# Check for Argon2 availability
try:
from argon2.low_level import Type, hash_secret_raw
HAS_ARGON2 = True
except ImportError:
HAS_ARGON2 = False
@@ -79,15 +80,17 @@ def _resolve_channel_key(channel_key: str | bool | None) -> bytes | None:
# Auto-detect from environment/config
if channel_key is None or channel_key == CHANNEL_KEY_AUTO:
from .channel import get_channel_key_hash
return get_channel_key_hash()
# Explicit key provided - validate and hash it
if isinstance(channel_key, str):
from .channel import format_channel_key, validate_channel_key
if not validate_channel_key(channel_key):
raise ValueError(f"Invalid channel key format: {channel_key}")
formatted = format_channel_key(channel_key)
return hashlib.sha256(formatted.encode('utf-8')).digest()
return hashlib.sha256(formatted.encode("utf-8")).digest()
raise ValueError(f"Invalid channel_key type: {type(channel_key)}")
@@ -96,6 +99,7 @@ def _resolve_channel_key(channel_key: str | bool | None) -> bytes | None:
# CORE CRYPTO FUNCTIONS
# =============================================================================
def hash_photo(image_data: bytes) -> bytes:
"""
Compute deterministic hash of photo pixel content.
@@ -109,7 +113,7 @@ def hash_photo(image_data: bytes) -> bytes:
Returns:
32-byte SHA-256 hash
"""
img: Image.Image = Image.open(io.BytesIO(image_data)).convert('RGB')
img: Image.Image = Image.open(io.BytesIO(image_data)).convert("RGB")
pixels = img.tobytes()
# Double-hash with prefix for additional mixing
@@ -163,12 +167,7 @@ def derive_hybrid_key(
channel_hash = _resolve_channel_key(channel_key)
# Build key material
key_material = (
photo_hash +
passphrase.lower().encode() +
pin.encode() +
salt
)
key_material = photo_hash + passphrase.lower().encode() + pin.encode() + salt
# Add RSA key hash if provided
if rsa_key_data:
@@ -186,7 +185,7 @@ def derive_hybrid_key(
memory_cost=ARGON2_MEMORY_COST,
parallelism=ARGON2_PARALLELISM,
hash_len=32,
type=Type.ID
type=Type.ID,
)
else:
kdf = PBKDF2HMAC(
@@ -194,7 +193,7 @@ def derive_hybrid_key(
length=32,
salt=salt,
iterations=PBKDF2_ITERATIONS,
backend=default_backend()
backend=default_backend(),
)
key = kdf.derive(key_material)
@@ -232,11 +231,7 @@ def derive_pixel_key(
# Resolve channel key
channel_hash = _resolve_channel_key(channel_key)
material = (
photo_hash +
passphrase.lower().encode() +
pin.encode()
)
material = photo_hash + passphrase.lower().encode() + pin.encode()
if rsa_key_data:
material += hashlib.sha256(rsa_key_data).digest()
@@ -268,31 +263,31 @@ def _pack_payload(
"""
if isinstance(content, str):
# Text message
data = content.encode('utf-8')
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')
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
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
bytes([PAYLOAD_FILE])
+ struct.pack(">H", 0) # No filename
+ struct.pack(">H", 0) # No mime
+ content
)
return packed, PAYLOAD_FILE
@@ -314,42 +309,39 @@ def _unpack_payload(data: bytes) -> DecodeResult:
if payload_type == PAYLOAD_TEXT:
# Text message
text = data[1:].decode('utf-8')
return DecodeResult(payload_type='text', message=text)
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]
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
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]
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
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
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)
text = data.decode("utf-8")
return DecodeResult(payload_type="text", message=text)
except UnicodeDecodeError:
return DecodeResult(payload_type='file', file_data=data)
return DecodeResult(payload_type="file", file_data=data)
# =============================================================================
@@ -415,7 +407,7 @@ def encrypt_message(
padding_len = secrets.randbelow(256) + 64
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))
padding = secrets.token_bytes(padding_needed - 4) + struct.pack(">I", len(packed_payload))
padded_message = packed_payload + padding
# Build header for AAD
@@ -428,13 +420,7 @@ def encrypt_message(
ciphertext = encryptor.update(padded_message) + encryptor.finalize()
# v4.0.0: Header with flags byte
return (
header +
salt +
iv +
encryptor.tag +
ciphertext
)
return header + salt + iv + encryptor.tag + ciphertext
except Exception as e:
raise EncryptionError(f"Encryption failed: {e}") from e
@@ -464,22 +450,22 @@ def parse_header(encrypted_data: bytes) -> dict | None:
flags = encrypted_data[5]
offset = 6
salt = encrypted_data[offset:offset + SALT_SIZE]
salt = encrypted_data[offset : offset + SALT_SIZE]
offset += SALT_SIZE
iv = encrypted_data[offset:offset + IV_SIZE]
iv = encrypted_data[offset : offset + IV_SIZE]
offset += IV_SIZE
tag = encrypted_data[offset:offset + TAG_SIZE]
tag = encrypted_data[offset : offset + TAG_SIZE]
offset += TAG_SIZE
ciphertext = encrypted_data[offset:]
return {
'version': version,
'flags': flags,
'has_channel_key': bool(flags & FLAG_CHANNEL_KEY),
'salt': salt,
'iv': iv,
'tag': tag,
'ciphertext': ciphertext
"version": version,
"flags": flags,
"has_channel_key": bool(flags & FLAG_CHANNEL_KEY),
"salt": salt,
"iv": iv,
"tag": tag,
"ciphertext": ciphertext,
}
except Exception:
return None
@@ -518,26 +504,24 @@ def decrypt_message(
# Check for channel key mismatch and provide helpful error
channel_hash = _resolve_channel_key(channel_key)
has_configured_key = channel_hash is not None
message_has_key = header['has_channel_key']
message_has_key = header["has_channel_key"]
try:
key = derive_hybrid_key(
photo_data, passphrase, header['salt'], pin, rsa_key_data, channel_key
photo_data, passphrase, header["salt"], pin, rsa_key_data, channel_key
)
# Reconstruct header for AAD verification
aad_header = MAGIC_HEADER + bytes([FORMAT_VERSION, header['flags']])
aad_header = MAGIC_HEADER + bytes([FORMAT_VERSION, header["flags"]])
cipher = Cipher(
algorithms.AES(key),
modes.GCM(header['iv'], header['tag']),
backend=default_backend()
algorithms.AES(key), modes.GCM(header["iv"], header["tag"]), backend=default_backend()
)
decryptor = cipher.decryptor()
decryptor.authenticate_additional_data(aad_header)
padded_plaintext = decryptor.update(header['ciphertext']) + decryptor.finalize()
original_length = struct.unpack('>I', padded_plaintext[-4:])[0]
padded_plaintext = decryptor.update(header["ciphertext"]) + decryptor.finalize()
original_length = struct.unpack(">I", padded_plaintext[-4:])[0]
payload_data = padded_plaintext[:original_length]
result = _unpack_payload(payload_data)
@@ -596,7 +580,7 @@ def decrypt_message_text(
if result.file_data:
# Try to decode as text
try:
return result.file_data.decode('utf-8')
return result.file_data.decode("utf-8")
except UnicodeDecodeError:
raise DecryptionError(
f"Content is a binary file ({result.filename or 'unnamed'}), not text"
@@ -615,6 +599,7 @@ def has_argon2() -> bool:
# CHANNEL KEY UTILITIES (exposed for convenience)
# =============================================================================
def get_active_channel_key() -> str | None:
"""
Get the currently configured channel key (if any).
@@ -623,6 +608,7 @@ def get_active_channel_key() -> str | None:
Formatted channel key string, or None if not configured
"""
from .channel import get_channel_key
return get_channel_key()
@@ -637,4 +623,5 @@ def get_channel_fingerprint(key: str | None = None) -> str | None:
Masked key like "ABCD-••••-••••-••••-••••-••••-••••-3456" or None
"""
from .channel import get_channel_fingerprint as _get_fingerprint
return _get_fingerprint(key)

312
src/stegasoo/dct_steganography.py
View File

@@ -28,10 +28,12 @@ from PIL import Image
# Prefer scipy.fft (newer, more stable) over scipy.fftpack
try:
from scipy.fft import dct, idct
HAS_SCIPY = True
except ImportError:
try:
from scipy.fftpack import dct, idct
HAS_SCIPY = True
except ImportError:
HAS_SCIPY = False
@@ -41,6 +43,7 @@ except ImportError:
# Check for jpegio availability (for proper JPEG mode)
try:
import jpegio as jio
HAS_JPEGIO = True
except ImportError:
HAS_JPEGIO = False
@@ -53,19 +56,49 @@ except ImportError:
BLOCK_SIZE = 8
EMBED_POSITIONS = [
(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), (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),
(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),
(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),
]
DEFAULT_EMBED_POSITIONS = EMBED_POSITIONS[4:20]
QUANT_STEP = 25
DCT_MAGIC = b'DCTS'
DCT_MAGIC = b"DCTS"
HEADER_SIZE = 10
OUTPUT_FORMAT_PNG = 'png'
OUTPUT_FORMAT_JPEG = 'jpeg'
OUTPUT_FORMAT_PNG = "png"
OUTPUT_FORMAT_JPEG = "jpeg"
JPEG_OUTPUT_QUALITY = 95
JPEGIO_MAGIC = b'JPGS'
JPEGIO_MAGIC = b"JPGS"
JPEGIO_MIN_COEF_MAGNITUDE = 2
JPEGIO_EMBED_CHANNEL = 0
FLAG_COLOR_MODE = 0x01
@@ -83,9 +116,10 @@ JPEGIO_MAX_QUANT_VALUE_THRESHOLD = 1 # If all quant values <= this, normalize
# DATA CLASSES
# ============================================================================
class DCTOutputFormat(Enum):
PNG = 'png'
JPEG = 'jpeg'
PNG = "png"
JPEG = "jpeg"
@dataclass
@@ -99,7 +133,7 @@ class DCTEmbedStats:
image_height: int
output_format: str
jpeg_native: bool = False
color_mode: str = 'grayscale'
color_mode: str = "grayscale"
@dataclass
@@ -119,11 +153,10 @@ class DCTCapacityInfo:
# AVAILABILITY CHECKS
# ============================================================================
def _check_scipy():
if not HAS_SCIPY:
raise ImportError(
"DCT steganography requires scipy. Install with: pip install scipy"
)
raise ImportError("DCT steganography requires scipy. Install with: pip install scipy")
def has_dct_support() -> bool:
@@ -139,25 +172,26 @@ def has_jpegio_support() -> bool:
# These create fresh arrays to avoid scipy memory corruption issues
# ============================================================================
def _safe_dct2(block: np.ndarray) -> np.ndarray:
"""
Apply 2D DCT with memory isolation.
Creates a completely fresh array to avoid heap corruption.
"""
# Create a brand new array (not a view)
safe_block = np.array(block, dtype=np.float64, copy=True, order='C')
safe_block = np.array(block, dtype=np.float64, copy=True, order="C")
# First DCT on columns (transpose -> DCT rows -> transpose back)
temp = np.zeros_like(safe_block, dtype=np.float64, order='C')
temp = np.zeros_like(safe_block, dtype=np.float64, order="C")
for i in range(BLOCK_SIZE):
col = np.array(safe_block[:, i], dtype=np.float64, copy=True)
temp[:, i] = dct(col, norm='ortho')
temp[:, i] = dct(col, norm="ortho")
# Second DCT on rows
result = np.zeros_like(temp, dtype=np.float64, order='C')
result = np.zeros_like(temp, dtype=np.float64, order="C")
for i in range(BLOCK_SIZE):
row = np.array(temp[i, :], dtype=np.float64, copy=True)
result[i, :] = dct(row, norm='ortho')
result[i, :] = dct(row, norm="ortho")
return result
@@ -168,19 +202,19 @@ def _safe_idct2(block: np.ndarray) -> np.ndarray:
Creates a completely fresh array to avoid heap corruption.
"""
# Create a brand new array (not a view)
safe_block = np.array(block, dtype=np.float64, copy=True, order='C')
safe_block = np.array(block, dtype=np.float64, copy=True, order="C")
# First IDCT on rows
temp = np.zeros_like(safe_block, dtype=np.float64, order='C')
temp = np.zeros_like(safe_block, dtype=np.float64, order="C")
for i in range(BLOCK_SIZE):
row = np.array(safe_block[i, :], dtype=np.float64, copy=True)
temp[i, :] = idct(row, norm='ortho')
temp[i, :] = idct(row, norm="ortho")
# Second IDCT on columns
result = np.zeros_like(temp, dtype=np.float64, order='C')
result = np.zeros_like(temp, dtype=np.float64, order="C")
for i in range(BLOCK_SIZE):
col = np.array(temp[:, i], dtype=np.float64, copy=True)
result[:, i] = idct(col, norm='ortho')
result[:, i] = idct(col, norm="ortho")
return result
@@ -189,20 +223,21 @@ def _safe_idct2(block: np.ndarray) -> np.ndarray:
# IMAGE PROCESSING HELPERS
# ============================================================================
def _to_grayscale(image_data: bytes) -> np.ndarray:
img = Image.open(io.BytesIO(image_data))
gray = img.convert('L')
return np.array(gray, dtype=np.float64, copy=True, order='C')
gray = img.convert("L")
return np.array(gray, dtype=np.float64, copy=True, order="C")
def _extract_y_channel(image_data: bytes) -> np.ndarray:
img = Image.open(io.BytesIO(image_data))
if img.mode != 'RGB':
img = img.convert('RGB')
if img.mode != "RGB":
img = img.convert("RGB")
rgb = np.array(img, dtype=np.float64, copy=True, order='C')
rgb = np.array(img, dtype=np.float64, copy=True, order="C")
Y = 0.299 * rgb[:, :, 0] + 0.587 * rgb[:, :, 1] + 0.114 * rgb[:, :, 2]
return np.array(Y, dtype=np.float64, copy=True, order='C')
return np.array(Y, dtype=np.float64, copy=True, order="C")
def _pad_to_blocks(image: np.ndarray) -> tuple[np.ndarray, tuple[int, int]]:
@@ -211,27 +246,27 @@ def _pad_to_blocks(image: np.ndarray) -> tuple[np.ndarray, tuple[int, int]]:
new_w = ((w + BLOCK_SIZE - 1) // BLOCK_SIZE) * BLOCK_SIZE
if new_h == h and new_w == w:
return np.array(image, dtype=np.float64, copy=True, order='C'), (h, w)
return np.array(image, dtype=np.float64, copy=True, order="C"), (h, w)
padded = np.zeros((new_h, new_w), dtype=np.float64, order='C')
padded = np.zeros((new_h, new_w), dtype=np.float64, order="C")
padded[:h, :w] = image
# Simple edge replication for padding
if new_h > h:
for i in range(h, new_h):
padded[i, :w] = padded[h-1, :w]
padded[i, :w] = padded[h - 1, :w]
if new_w > w:
for j in range(w, new_w):
padded[:h, j] = padded[:h, w-1]
padded[:h, j] = padded[:h, w - 1]
if new_h > h and new_w > w:
padded[h:, w:] = padded[h-1, w-1]
padded[h:, w:] = padded[h - 1, w - 1]
return padded, (h, w)
def _unpad_image(image: np.ndarray, original_size: tuple[int, int]) -> np.ndarray:
h, w = original_size
return np.array(image[:h, :w], dtype=np.float64, copy=True, order='C')
return np.array(image[:h, :w], dtype=np.float64, copy=True, order="C")
def _embed_bit_in_coeff(coef: float, bit: int, quant_step: int = QUANT_STEP) -> float:
@@ -251,7 +286,7 @@ def _extract_bit_from_coeff(coef: float, quant_step: int = QUANT_STEP) -> int:
def _generate_block_order(num_blocks: int, seed: bytes) -> list:
hash_bytes = hashlib.sha256(seed).digest()
rng = np.random.RandomState(int.from_bytes(hash_bytes[:4], 'big'))
rng = np.random.RandomState(int.from_bytes(hash_bytes[:4], "big"))
order = list(range(num_blocks))
rng.shuffle(order)
return order
@@ -259,25 +294,23 @@ def _generate_block_order(num_blocks: int, seed: bytes) -> list:
def _save_stego_image(image: np.ndarray, output_format: str = OUTPUT_FORMAT_PNG) -> bytes:
clipped = np.clip(image, 0, 255).astype(np.uint8)
img = Image.fromarray(clipped, mode='L')
img = Image.fromarray(clipped, mode="L")
buffer = io.BytesIO()
if output_format == OUTPUT_FORMAT_JPEG:
img.save(buffer, format='JPEG', quality=JPEG_OUTPUT_QUALITY,
subsampling=0, optimize=True)
img.save(buffer, format="JPEG", quality=JPEG_OUTPUT_QUALITY, subsampling=0, optimize=True)
else:
img.save(buffer, format='PNG', optimize=True)
img.save(buffer, format="PNG", optimize=True)
return buffer.getvalue()
def _save_color_image(rgb_array: np.ndarray, output_format: str = OUTPUT_FORMAT_PNG) -> bytes:
clipped = np.clip(rgb_array, 0, 255).astype(np.uint8)
img = Image.fromarray(clipped, mode='RGB')
img = Image.fromarray(clipped, mode="RGB")
buffer = io.BytesIO()
if output_format == OUTPUT_FORMAT_JPEG:
img.save(buffer, format='JPEG', quality=JPEG_OUTPUT_QUALITY,
subsampling=0, optimize=True)
img.save(buffer, format="JPEG", quality=JPEG_OUTPUT_QUALITY, subsampling=0, optimize=True)
else:
img.save(buffer, format='PNG', optimize=True)
img.save(buffer, format="PNG", optimize=True)
return buffer.getvalue()
@@ -286,9 +319,13 @@ def _rgb_to_ycbcr(rgb: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
G = rgb[:, :, 1].astype(np.float64)
B = rgb[:, :, 2].astype(np.float64)
Y = np.array(0.299 * R + 0.587 * G + 0.114 * B, dtype=np.float64, copy=True, order='C')
Cb = np.array(128 - 0.168736 * R - 0.331264 * G + 0.5 * B, dtype=np.float64, copy=True, order='C')
Cr = np.array(128 + 0.5 * R - 0.418688 * G - 0.081312 * B, dtype=np.float64, copy=True, order='C')
Y = np.array(0.299 * R + 0.587 * G + 0.114 * B, dtype=np.float64, copy=True, order="C")
Cb = np.array(
128 - 0.168736 * R - 0.331264 * G + 0.5 * B, dtype=np.float64, copy=True, order="C"
)
Cr = np.array(
128 + 0.5 * R - 0.418688 * G - 0.081312 * B, dtype=np.float64, copy=True, order="C"
)
return Y, Cb, Cr
@@ -298,7 +335,7 @@ def _ycbcr_to_rgb(Y: np.ndarray, Cb: np.ndarray, Cr: np.ndarray) -> np.ndarray:
G = Y - 0.344136 * (Cb - 128) - 0.714136 * (Cr - 128)
B = Y + 1.772 * (Cb - 128)
rgb = np.zeros((Y.shape[0], Y.shape[1], 3), dtype=np.float64, order='C')
rgb = np.zeros((Y.shape[0], Y.shape[1], 3), dtype=np.float64, order="C")
rgb[:, :, 0] = R
rgb[:, :, 1] = G
rgb[:, :, 2] = B
@@ -306,19 +343,21 @@ def _ycbcr_to_rgb(Y: np.ndarray, Cb: np.ndarray, Cr: np.ndarray) -> np.ndarray:
def _create_header(data_length: int, flags: int = 0) -> bytes:
return struct.pack('>4sBBI', DCT_MAGIC, 1, flags, data_length)
return struct.pack(">4sBBI", DCT_MAGIC, 1, flags, data_length)
def _parse_header(header_bits: list) -> tuple[int, int, int]:
if len(header_bits) < HEADER_SIZE * 8:
raise ValueError("Insufficient header data")
header_bytes = bytes([
sum(header_bits[i*8:(i+1)*8][j] << (7-j) for j in range(8))
for i in range(HEADER_SIZE)
])
header_bytes = bytes(
[
sum(header_bits[i * 8 : (i + 1) * 8][j] << (7 - j) for j in range(8))
for i in range(HEADER_SIZE)
]
)
magic, version, flags, length = struct.unpack('>4sBBI', header_bytes)
magic, version, flags, length = struct.unpack(">4sBBI", header_bytes)
if magic != DCT_MAGIC:
raise ValueError("Invalid DCT stego magic bytes")
@@ -330,9 +369,11 @@ def _parse_header(header_bits: list) -> tuple[int, int, int]:
# JPEGIO HELPERS
# ============================================================================
def _jpegio_bytes_to_file(data: bytes, suffix: str = '.jpg') -> str:
def _jpegio_bytes_to_file(data: bytes, suffix: str = ".jpg") -> str:
import os
import tempfile
fd, path = tempfile.mkstemp(suffix=suffix)
try:
os.write(fd, data)
@@ -355,20 +396,20 @@ def _jpegio_get_usable_positions(coef_array: np.ndarray) -> list:
def _jpegio_generate_order(num_positions: int, seed: bytes) -> list:
hash_bytes = hashlib.sha256(seed + b"jpeg_coef_order").digest()
rng = np.random.RandomState(int.from_bytes(hash_bytes[:4], 'big'))
rng = np.random.RandomState(int.from_bytes(hash_bytes[:4], "big"))
order = list(range(num_positions))
rng.shuffle(order)
return order
def _jpegio_create_header(data_length: int, flags: int = 0) -> bytes:
return struct.pack('>4sBBI', JPEGIO_MAGIC, 1, flags, data_length)
return struct.pack(">4sBBI", JPEGIO_MAGIC, 1, flags, data_length)
def _jpegio_parse_header(header_bytes: bytes) -> tuple[int, int, int]:
if len(header_bytes) < HEADER_SIZE:
raise ValueError("Insufficient header data")
magic, version, flags, length = struct.unpack('>4sBBI', header_bytes[:HEADER_SIZE])
magic, version, flags, length = struct.unpack(">4sBBI", header_bytes[:HEADER_SIZE])
if magic != JPEGIO_MAGIC:
raise ValueError(f"Invalid JPEG stego magic: {magic}")
return version, flags, length
@@ -378,6 +419,7 @@ def _jpegio_parse_header(header_bytes: bytes) -> tuple[int, int, int]:
# PUBLIC API
# ============================================================================
def calculate_dct_capacity(image_data: bytes) -> DCTCapacityInfo:
"""Calculate DCT embedding capacity of an image."""
_check_scipy()
@@ -405,7 +447,7 @@ def calculate_dct_capacity(image_data: bytes) -> DCTCapacityInfo:
bits_per_block=bits_per_block,
total_capacity_bits=total_bits,
total_capacity_bytes=total_bytes,
usable_capacity_bytes=usable_bytes
usable_capacity_bytes=usable_bytes,
)
@@ -427,24 +469,24 @@ def estimate_capacity_comparison(image_data: bytes) -> dict:
dct_bytes = (blocks * 16) // 8 - HEADER_SIZE
return {
'width': width,
'height': height,
'lsb': {
'capacity_bytes': lsb_bytes,
'capacity_kb': lsb_bytes / 1024,
'output': 'PNG/BMP (color)',
"width": width,
"height": height,
"lsb": {
"capacity_bytes": lsb_bytes,
"capacity_kb": lsb_bytes / 1024,
"output": "PNG/BMP (color)",
},
'dct': {
'capacity_bytes': dct_bytes,
'capacity_kb': dct_bytes / 1024,
'output': 'PNG or JPEG (grayscale)',
'ratio_vs_lsb': (dct_bytes / lsb_bytes * 100) if lsb_bytes > 0 else 0,
'available': HAS_SCIPY,
"dct": {
"capacity_bytes": dct_bytes,
"capacity_kb": dct_bytes / 1024,
"output": "PNG or JPEG (grayscale)",
"ratio_vs_lsb": (dct_bytes / lsb_bytes * 100) if lsb_bytes > 0 else 0,
"available": HAS_SCIPY,
},
"jpeg_native": {
"available": HAS_JPEGIO,
"note": "Uses jpegio for proper JPEG coefficient embedding",
},
'jpeg_native': {
'available': HAS_JPEGIO,
'note': 'Uses jpegio for proper JPEG coefficient embedding',
}
}
@@ -453,14 +495,14 @@ def embed_in_dct(
carrier_image: bytes,
seed: bytes,
output_format: str = OUTPUT_FORMAT_PNG,
color_mode: str = 'color',
color_mode: str = "color",
) -> tuple[bytes, DCTEmbedStats]:
"""Embed data using DCT coefficient modification."""
if output_format not in (OUTPUT_FORMAT_PNG, OUTPUT_FORMAT_JPEG):
raise ValueError(f"Invalid output format: {output_format}")
if color_mode not in ('color', 'grayscale'):
color_mode = 'color'
if color_mode not in ("color", "grayscale"):
color_mode = "color"
if output_format == OUTPUT_FORMAT_JPEG and HAS_JPEGIO:
return _embed_jpegio(data, carrier_image, seed, color_mode)
@@ -474,7 +516,7 @@ def _embed_scipy_dct_safe(
carrier_image: bytes,
seed: bytes,
output_format: str,
color_mode: str = 'color',
color_mode: str = "color",
) -> tuple[bytes, DCTEmbedStats]:
"""
Embed using scipy DCT with safe memory handling.
@@ -494,7 +536,7 @@ def _embed_scipy_dct_safe(
img = Image.open(io.BytesIO(carrier_image))
width, height = img.size
flags = FLAG_COLOR_MODE if color_mode == 'color' else 0
flags = FLAG_COLOR_MODE if color_mode == "color" else 0
# Prepare payload bits
header = _create_header(len(data), flags)
@@ -509,12 +551,12 @@ def _embed_scipy_dct_safe(
block_order = _generate_block_order(num_blocks, seed)
blocks_x = width // BLOCK_SIZE
if color_mode == 'color' and img.mode in ('RGB', 'RGBA'):
if img.mode == 'RGBA':
img = img.convert('RGB')
if color_mode == "color" and img.mode in ("RGB", "RGBA"):
if img.mode == "RGBA":
img = img.convert("RGB")
# Process color image
rgb = np.array(img, dtype=np.float64, copy=True, order='C')
rgb = np.array(img, dtype=np.float64, copy=True, order="C")
img.close()
Y, Cb, Cr = _rgb_to_ycbcr(rgb)
@@ -592,7 +634,7 @@ def _embed_in_channel_safe(
h, w = channel.shape
# Create result with explicit new memory
result = np.array(channel, dtype=np.float64, copy=True, order='C')
result = np.array(channel, dtype=np.float64, copy=True, order="C")
bit_idx = 0
@@ -605,8 +647,10 @@ def _embed_in_channel_safe(
# Extract block - create brand new array
block = np.array(
result[by:by+BLOCK_SIZE, bx:bx+BLOCK_SIZE],
dtype=np.float64, copy=True, order='C'
result[by : by + BLOCK_SIZE, bx : bx + BLOCK_SIZE],
dtype=np.float64,
copy=True,
order="C",
)
# Apply safe DCT (row-by-row)
@@ -617,8 +661,7 @@ def _embed_in_channel_safe(
if bit_idx >= len(bits):
break
dct_block[pos[0], pos[1]] = _embed_bit_in_coeff(
float(dct_block[pos[0], pos[1]]),
bits[bit_idx]
float(dct_block[pos[0], pos[1]]), bits[bit_idx]
)
bit_idx += 1
@@ -626,7 +669,7 @@ def _embed_in_channel_safe(
modified_block = _safe_idct2(dct_block)
# Copy back
result[by:by+BLOCK_SIZE, bx:bx+BLOCK_SIZE] = modified_block
result[by : by + BLOCK_SIZE, bx : bx + BLOCK_SIZE] = modified_block
# Clean up this iteration
del block, dct_block, modified_block
@@ -654,13 +697,13 @@ def _normalize_jpeg_for_jpegio(image_data: bytes) -> bytes:
img = Image.open(io.BytesIO(image_data))
# Only process JPEGs
if img.format != 'JPEG':
if img.format != "JPEG":
img.close()
return image_data
# Check quantization tables
needs_normalization = False
if hasattr(img, 'quantization') and img.quantization:
if hasattr(img, "quantization") and img.quantization:
for table_id, table in img.quantization.items():
# If all values in any table are <= threshold, normalize
if max(table) <= JPEGIO_MAX_QUANT_VALUE_THRESHOLD:
@@ -672,11 +715,11 @@ def _normalize_jpeg_for_jpegio(image_data: bytes) -> bytes:
return image_data
# Re-save at safe quality level
if img.mode != 'RGB':
img = img.convert('RGB')
if img.mode != "RGB":
img = img.convert("RGB")
buffer = io.BytesIO()
img.save(buffer, format='JPEG', quality=JPEGIO_NORMALIZE_QUALITY, subsampling=0)
img.save(buffer, format="JPEG", quality=JPEGIO_NORMALIZE_QUALITY, subsampling=0)
img.close()
return buffer.getvalue()
@@ -686,7 +729,7 @@ def _embed_jpegio(
data: bytes,
carrier_image: bytes,
seed: bytes,
color_mode: str = 'color',
color_mode: str = "color",
) -> tuple[bytes, DCTEmbedStats]:
"""Embed using jpegio for proper JPEG coefficient modification."""
import os
@@ -698,18 +741,18 @@ def _embed_jpegio(
img = Image.open(io.BytesIO(carrier_image))
width, height = img.size
if img.format != 'JPEG':
if img.format != "JPEG":
buffer = io.BytesIO()
if img.mode != 'RGB':
img = img.convert('RGB')
img.save(buffer, format='JPEG', quality=95, subsampling=0)
if img.mode != "RGB":
img = img.convert("RGB")
img.save(buffer, format="JPEG", quality=95, subsampling=0)
carrier_image = buffer.getvalue()
img.close()
input_path = _jpegio_bytes_to_file(carrier_image, suffix='.jpg')
output_path = tempfile.mktemp(suffix='.jpg')
input_path = _jpegio_bytes_to_file(carrier_image, suffix=".jpg")
output_path = tempfile.mktemp(suffix=".jpg")
flags = FLAG_COLOR_MODE if color_mode == 'color' else 0
flags = FLAG_COLOR_MODE if color_mode == "color" else 0
try:
jpeg = jio.read(input_path)
@@ -750,7 +793,7 @@ def _embed_jpegio(
jio.write(jpeg, output_path)
with open(output_path, 'rb') as f:
with open(output_path, "rb") as f:
stego_bytes = f.read()
stats = DCTEmbedStats(
@@ -782,7 +825,7 @@ def extract_from_dct(stego_image: bytes, seed: bytes) -> bytes:
fmt = img.format
img.close()
if fmt == 'JPEG' and HAS_JPEGIO:
if fmt == "JPEG" and HAS_JPEGIO:
try:
return _extract_jpegio(stego_image, seed)
except ValueError:
@@ -798,7 +841,7 @@ def _extract_scipy_dct_safe(stego_image: bytes, seed: bytes) -> bytes:
width, height = img.size
mode = img.mode
if mode in ('RGB', 'RGBA'):
if mode in ("RGB", "RGBA"):
channel = _extract_y_channel(stego_image)
else:
channel = _to_grayscale(stego_image)
@@ -821,8 +864,10 @@ def _extract_scipy_dct_safe(stego_image: bytes, seed: bytes) -> bytes:
bx = (block_num % blocks_x) * BLOCK_SIZE
block = np.array(
padded[by:by+BLOCK_SIZE, bx:bx+BLOCK_SIZE],
dtype=np.float64, copy=True, order='C'
padded[by : by + BLOCK_SIZE, bx : bx + BLOCK_SIZE],
dtype=np.float64,
copy=True,
order="C",
)
dct_block = _safe_dct2(block)
@@ -834,7 +879,7 @@ def _extract_scipy_dct_safe(stego_image: bytes, seed: bytes) -> bytes:
if len(all_bits) >= HEADER_SIZE * 8:
try:
_, flags, data_length = _parse_header(all_bits[:HEADER_SIZE * 8])
_, flags, data_length = _parse_header(all_bits[: HEADER_SIZE * 8])
total_needed = (HEADER_SIZE + data_length) * 8
if len(all_bits) >= total_needed:
break
@@ -845,12 +890,14 @@ def _extract_scipy_dct_safe(stego_image: bytes, seed: bytes) -> bytes:
gc.collect()
_, flags, data_length = _parse_header(all_bits)
data_bits = all_bits[HEADER_SIZE * 8:(HEADER_SIZE + data_length) * 8]
data_bits = all_bits[HEADER_SIZE * 8 : (HEADER_SIZE + data_length) * 8]
data = bytes([
sum(data_bits[i*8:(i+1)*8][j] << (7-j) for j in range(8))
for i in range(data_length)
])
data = bytes(
[
sum(data_bits[i * 8 : (i + 1) * 8][j] << (7 - j) for j in range(8))
for i in range(data_length)
]
)
return data
@@ -863,7 +910,7 @@ def _extract_jpegio(stego_image: bytes, seed: bytes) -> bytes:
# (shouldn't happen with stego images, but be defensive)
stego_image = _normalize_jpeg_for_jpegio(stego_image)
temp_path = _jpegio_bytes_to_file(stego_image, suffix='.jpg')
temp_path = _jpegio_bytes_to_file(stego_image, suffix=".jpg")
try:
jpeg = jio.read(temp_path)
@@ -873,15 +920,17 @@ def _extract_jpegio(stego_image: bytes, seed: bytes) -> bytes:
order = _jpegio_generate_order(len(all_positions), seed)
header_bits = []
for pos_idx in order[:HEADER_SIZE * 8]:
for pos_idx in order[: HEADER_SIZE * 8]:
row, col = all_positions[pos_idx]
coef = coef_array[row, col]
header_bits.append(coef & 1)
header_bytes = bytes([
sum(header_bits[i*8:(i+1)*8][j] << (7-j) for j in range(8))
for i in range(HEADER_SIZE)
])
header_bytes = bytes(
[
sum(header_bits[i * 8 : (i + 1) * 8][j] << (7 - j) for j in range(8))
for i in range(HEADER_SIZE)
]
)
_, flags, data_length = _jpegio_parse_header(header_bytes)
@@ -895,12 +944,14 @@ def _extract_jpegio(stego_image: bytes, seed: bytes) -> bytes:
coef = coef_array[row, col]
all_bits.append(coef & 1)
data_bits = all_bits[HEADER_SIZE * 8:]
data_bits = all_bits[HEADER_SIZE * 8 :]
data = bytes([
sum(data_bits[i*8:(i+1)*8][j] << (7-j) for j in range(8))
for i in range(data_length)
])
data = bytes(
[
sum(data_bits[i * 8 : (i + 1) * 8][j] << (7 - j) for j in range(8))
for i in range(data_length)
]
)
return data
@@ -915,13 +966,14 @@ def _extract_jpegio(stego_image: bytes, seed: bytes) -> bytes:
# CONVENIENCE FUNCTIONS
# ============================================================================
def get_output_extension(output_format: str) -> str:
if output_format == OUTPUT_FORMAT_JPEG:
return '.jpg'
return '.png'
return ".jpg"
return ".png"
def get_output_mimetype(output_format: str) -> str:
if output_format == OUTPUT_FORMAT_JPEG:
return 'image/jpeg'
return 'image/png'
return "image/jpeg"
return "image/png"

18
src/stegasoo/debug.py
View File

@@ -68,6 +68,7 @@ def debug_exception(e: Exception, context: str = "") -> None:
def time_function(func: Callable) -> Callable:
"""Decorator to time function execution for performance debugging."""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
if not (DEBUG_ENABLED and LOG_PERFORMANCE):
@@ -96,16 +97,17 @@ def memory_usage() -> dict[str, float | str]:
import os
import psutil
process = psutil.Process(os.getpid())
mem_info = process.memory_info()
return {
'rss_mb': mem_info.rss / 1024 / 1024,
'vms_mb': mem_info.vms / 1024 / 1024,
'percent': process.memory_percent(),
"rss_mb": mem_info.rss / 1024 / 1024,
"vms_mb": mem_info.vms / 1024 / 1024,
"percent": process.memory_percent(),
}
except ImportError:
return {'error': 'psutil not installed'}
return {"error": "psutil not installed"}
def hexdump(data: bytes, offset: int = 0, length: int = 64) -> str:
@@ -117,16 +119,16 @@ def hexdump(data: bytes, offset: int = 0, length: int = 64) -> str:
data_to_dump = data[:length]
for i in range(0, len(data_to_dump), 16):
chunk = data_to_dump[i:i+16]
hex_str = ' '.join(f'{b:02x}' for b in chunk)
chunk = data_to_dump[i : i + 16]
hex_str = " ".join(f"{b:02x}" for b in chunk)
hex_str = hex_str.ljust(47)
ascii_str = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
ascii_str = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
result.append(f"{offset + i:08x}: {hex_str} {ascii_str}")
if len(data) > length:
result.append(f"... ({len(data) - length} more bytes)")
return '\n'.join(result)
return "\n".join(result)
class Debug:

19
src/stegasoo/decode.py
View File

@@ -75,9 +75,11 @@ def decode(
... channel_key="ABCD-1234-EFGH-5678-IJKL-9012-MNOP-3456"
... )
"""
debug.print(f"decode: passphrase length={len(passphrase.split())} words, "
f"mode={embed_mode}, "
f"channel_key={'explicit' if isinstance(channel_key, str) and channel_key else 'auto' if channel_key is None else 'none'}")
debug.print(
f"decode: passphrase length={len(passphrase.split())} words, "
f"mode={embed_mode}, "
f"channel_key={'explicit' if isinstance(channel_key, str) and channel_key else 'auto' if channel_key is None else 'none'}"
)
# Validate inputs
require_valid_image(stego_image, "Stego image")
@@ -91,9 +93,8 @@ def decode(
# Derive pixel/coefficient selection key (with channel key)
from .crypto import derive_pixel_key
pixel_key = derive_pixel_key(
reference_photo, passphrase, pin, rsa_key_data, channel_key
)
pixel_key = derive_pixel_key(reference_photo, passphrase, pin, rsa_key_data, channel_key)
# Extract encrypted data
encrypted = extract_from_image(
@@ -109,9 +110,7 @@ def decode(
debug.print(f"Extracted {len(encrypted)} bytes from image")
# Decrypt (with channel key)
result = decrypt_message(
encrypted, reference_photo, passphrase, pin, rsa_key_data, channel_key
)
result = decrypt_message(encrypted, reference_photo, passphrase, pin, rsa_key_data, channel_key)
debug.print(f"Decryption successful: {result.payload_type}")
return result
@@ -222,7 +221,7 @@ def decode_text(
# Try to decode as text
if result.file_data:
try:
return result.file_data.decode('utf-8')
return result.file_data.decode("utf-8")
except UnicodeDecodeError:
raise DecryptionError(
f"Payload is a binary file ({result.filename or 'unnamed'}), not text"

14
src/stegasoo/encode.py
View File

@@ -82,9 +82,11 @@ def encode(
... channel_key="ABCD-1234-EFGH-5678-IJKL-9012-MNOP-3456"
... )
"""
debug.print(f"encode: passphrase length={len(passphrase.split())} words, "
f"pin={'set' if pin else 'none'}, mode={embed_mode}, "
f"channel_key={'explicit' if isinstance(channel_key, str) and channel_key else 'auto' if channel_key is None else 'none'}")
debug.print(
f"encode: passphrase length={len(passphrase.split())} words, "
f"pin={'set' if pin else 'none'}, mode={embed_mode}, "
f"channel_key={'explicit' if isinstance(channel_key, str) and channel_key else 'auto' if channel_key is None else 'none'}"
)
# Validate inputs
require_valid_payload(message)
@@ -105,9 +107,7 @@ def encode(
debug.print(f"Encrypted payload: {len(encrypted)} bytes")
# Derive pixel/coefficient selection key (with channel key)
pixel_key = derive_pixel_key(
reference_photo, passphrase, pin, rsa_key_data, channel_key
)
pixel_key = derive_pixel_key(reference_photo, passphrase, pin, rsa_key_data, channel_key)
# Embed in image
stego_data, stats, extension = embed_in_image(
@@ -124,7 +124,7 @@ def encode(
filename = generate_filename(extension=extension)
# Create result
if hasattr(stats, 'pixels_modified'):
if hasattr(stats, "pixels_modified"):
# LSB mode stats
return EncodeResult(
stego_image=stego_data,

23
src/stegasoo/exceptions.py
View File

@@ -7,6 +7,7 @@ Custom exception classes for clear error handling across all frontends.
class StegasooError(Exception):
"""Base exception for all Stegasoo errors."""
pass
@@ -14,33 +15,40 @@ class StegasooError(Exception):
# VALIDATION ERRORS
# ============================================================================
class ValidationError(StegasooError):
"""Base class for validation errors."""
pass
class PinValidationError(ValidationError):
"""PIN validation failed."""
pass
class MessageValidationError(ValidationError):
"""Message validation failed."""
pass
class ImageValidationError(ValidationError):
"""Image validation failed."""
pass
class KeyValidationError(ValidationError):
"""RSA key validation failed."""
pass
class SecurityFactorError(ValidationError):
"""Security factor requirements not met."""
pass
@@ -48,33 +56,40 @@ class SecurityFactorError(ValidationError):
# CRYPTO ERRORS
# ============================================================================
class CryptoError(StegasooError):
"""Base class for cryptographic errors."""
pass
class EncryptionError(CryptoError):
"""Encryption failed."""
pass
class DecryptionError(CryptoError):
"""Decryption failed (wrong key, corrupted data, etc.)."""
pass
class KeyDerivationError(CryptoError):
"""Key derivation failed."""
pass
class KeyGenerationError(CryptoError):
"""Key generation failed."""
pass
class KeyPasswordError(CryptoError):
"""RSA key password is incorrect or missing."""
pass
@@ -82,8 +97,10 @@ class KeyPasswordError(CryptoError):
# STEGANOGRAPHY ERRORS
# ============================================================================
class SteganographyError(StegasooError):
"""Base class for steganography errors."""
pass
@@ -100,16 +117,19 @@ class CapacityError(SteganographyError):
class ExtractionError(SteganographyError):
"""Failed to extract hidden data from image."""
pass
class EmbeddingError(SteganographyError):
"""Failed to embed data in image."""
pass
class InvalidHeaderError(SteganographyError):
"""Invalid or missing Stegasoo header in extracted data."""
pass
@@ -117,13 +137,16 @@ class InvalidHeaderError(SteganographyError):
# FILE ERRORS
# ============================================================================
class FileError(StegasooError):
"""Base class for file-related errors."""
pass
class FileNotFoundError(FileError):
"""Required file not found."""
pass

28
src/stegasoo/generate.py
View File

@@ -4,7 +4,6 @@ Stegasoo Generate Module (v3.2.0)
Public API for generating credentials (PINs, passphrases, RSA keys).
"""
from .constants import (
DEFAULT_PASSPHRASE_WORDS,
DEFAULT_PIN_LENGTH,
@@ -26,12 +25,12 @@ from .models import Credentials
# Re-export from keygen for convenience
__all__ = [
'generate_pin',
'generate_passphrase',
'generate_rsa_key',
'generate_credentials',
'export_rsa_key_pem',
'load_rsa_key',
"generate_pin",
"generate_passphrase",
"generate_rsa_key",
"generate_credentials",
"export_rsa_key_pem",
"load_rsa_key",
]
@@ -78,10 +77,7 @@ def generate_passphrase(words: int = DEFAULT_PASSPHRASE_WORDS) -> str:
return generate_phrase(words)
def generate_rsa_key(
bits: int = DEFAULT_RSA_BITS,
password: str | None = None
) -> str:
def generate_rsa_key(bits: int = DEFAULT_RSA_BITS, password: str | None = None) -> str:
"""
Generate an RSA private key in PEM format.
@@ -99,7 +95,7 @@ def generate_rsa_key(
"""
key_obj = _generate_rsa_key(bits)
pem_bytes = export_rsa_key_pem(key_obj, password)
return pem_bytes.decode('utf-8')
return pem_bytes.decode("utf-8")
def generate_credentials(
@@ -140,8 +136,10 @@ def generate_credentials(
if not use_pin and not use_rsa:
raise ValueError("Must select at least one security factor (PIN or RSA key)")
debug.print(f"Generating credentials: PIN={use_pin}, RSA={use_rsa}, "
f"passphrase_words={passphrase_words}")
debug.print(
f"Generating credentials: PIN={use_pin}, RSA={use_rsa}, "
f"passphrase_words={passphrase_words}"
)
# Generate passphrase (single, not daily)
passphrase = generate_phrase(passphrase_words)
@@ -154,7 +152,7 @@ def generate_credentials(
if use_rsa:
rsa_key_obj = _generate_rsa_key(rsa_bits)
rsa_key_bytes = export_rsa_key_pem(rsa_key_obj, rsa_password)
rsa_key_pem = rsa_key_bytes.decode('utf-8')
rsa_key_pem = rsa_key_bytes.decode("utf-8")
# Create Credentials object (v3.2.0 format)
creds = Credentials(

24
src/stegasoo/image_utils.py
View File

@@ -43,6 +43,7 @@ def get_image_info(image_data: bytes) -> ImageInfo:
if has_dct_support():
try:
from .dct_steganography import calculate_dct_capacity
dct_info = calculate_dct_capacity(image_data)
dct_capacity = dct_info.usable_capacity_bytes
except Exception as e:
@@ -61,8 +62,10 @@ def get_image_info(image_data: bytes) -> ImageInfo:
dct_capacity_kb=dct_capacity / 1024 if dct_capacity else None,
)
debug.print(f"Image info: {width}x{height}, LSB={lsb_capacity} bytes, "
f"DCT={dct_capacity or 'N/A'} bytes")
debug.print(
f"Image info: {width}x{height}, LSB={lsb_capacity} bytes, "
f"DCT={dct_capacity or 'N/A'} bytes"
)
return info
@@ -101,6 +104,7 @@ def compare_capacity(
if dct_available:
try:
from .dct_steganography import calculate_dct_capacity
dct_info = calculate_dct_capacity(carrier_image)
dct_bytes = dct_info.usable_capacity_bytes
dct_kb = dct_bytes / 1024
@@ -146,7 +150,7 @@ def validate_carrier_capacity(
from .steganography import calculate_capacity_by_mode
capacity_info = calculate_capacity_by_mode(carrier_image, embed_mode)
capacity = capacity_info['capacity_bytes']
capacity = capacity_info["capacity_bytes"]
# Add encryption overhead estimate
estimated_size = payload_size + 200 # Approximate overhead
@@ -156,11 +160,11 @@ def validate_carrier_capacity(
headroom = capacity - estimated_size
return {
'fits': fits,
'capacity': capacity,
'payload_size': payload_size,
'estimated_size': estimated_size,
'usage_percent': min(usage_percent, 100.0),
'headroom': headroom,
'mode': embed_mode,
"fits": fits,
"capacity": capacity,
"payload_size": payload_size,
"estimated_size": estimated_size,
"usage_percent": min(usage_percent, 100.0),
"headroom": headroom,
"mode": embed_mode,
}

78
src/stegasoo/keygen.py
View File

@@ -50,8 +50,10 @@ def generate_pin(length: int = DEFAULT_PIN_LENGTH) -> str:
>>> generate_pin(6)
"812345"
"""
debug.validate(MIN_PIN_LENGTH <= length <= MAX_PIN_LENGTH,
f"PIN length must be between {MIN_PIN_LENGTH} and {MAX_PIN_LENGTH}")
debug.validate(
MIN_PIN_LENGTH <= length <= MAX_PIN_LENGTH,
f"PIN length must be between {MIN_PIN_LENGTH} and {MAX_PIN_LENGTH}",
)
length = max(MIN_PIN_LENGTH, min(MAX_PIN_LENGTH, length))
@@ -59,7 +61,7 @@ def generate_pin(length: int = DEFAULT_PIN_LENGTH) -> str:
first_digit = str(secrets.randbelow(9) + 1)
# Remaining digits: 0-9
rest = ''.join(str(secrets.randbelow(10)) for _ in range(length - 1))
rest = "".join(str(secrets.randbelow(10)) for _ in range(length - 1))
pin = first_digit + rest
debug.print(f"Generated PIN: {pin}")
@@ -80,14 +82,16 @@ def generate_phrase(words_per_phrase: int = DEFAULT_PASSPHRASE_WORDS) -> str:
>>> generate_phrase(4)
"apple forest thunder mountain"
"""
debug.validate(MIN_PASSPHRASE_WORDS <= words_per_phrase <= MAX_PASSPHRASE_WORDS,
f"Words per phrase must be between {MIN_PASSPHRASE_WORDS} and {MAX_PASSPHRASE_WORDS}")
debug.validate(
MIN_PASSPHRASE_WORDS <= words_per_phrase <= MAX_PASSPHRASE_WORDS,
f"Words per phrase must be between {MIN_PASSPHRASE_WORDS} and {MAX_PASSPHRASE_WORDS}",
)
words_per_phrase = max(MIN_PASSPHRASE_WORDS, min(MAX_PASSPHRASE_WORDS, words_per_phrase))
wordlist = get_wordlist()
words = [secrets.choice(wordlist) for _ in range(words_per_phrase)]
phrase = ' '.join(words)
phrase = " ".join(words)
debug.print(f"Generated phrase: {phrase}")
return phrase
@@ -114,11 +118,12 @@ def generate_day_phrases(words_per_phrase: int = DEFAULT_PASSPHRASE_WORDS) -> di
{'Monday': 'apple forest thunder', 'Tuesday': 'banana river lightning', ...}
"""
import warnings
warnings.warn(
"generate_day_phrases() is deprecated in v3.2.0. "
"Use generate_phrase() for single passphrase.",
DeprecationWarning,
stacklevel=2
stacklevel=2,
)
phrases = {day: generate_phrase(words_per_phrase) for day in DAY_NAMES}
@@ -144,8 +149,7 @@ def generate_rsa_key(bits: int = DEFAULT_RSA_BITS) -> rsa.RSAPrivateKey:
>>> key.key_size
2048
"""
debug.validate(bits in VALID_RSA_SIZES,
f"RSA key size must be one of {VALID_RSA_SIZES}")
debug.validate(bits in VALID_RSA_SIZES, f"RSA key size must be one of {VALID_RSA_SIZES}")
if bits not in VALID_RSA_SIZES:
bits = DEFAULT_RSA_BITS
@@ -153,9 +157,7 @@ def generate_rsa_key(bits: int = DEFAULT_RSA_BITS) -> rsa.RSAPrivateKey:
debug.print(f"Generating {bits}-bit RSA key...")
try:
key = rsa.generate_private_key(
public_exponent=65537,
key_size=bits,
backend=default_backend()
public_exponent=65537, key_size=bits, backend=default_backend()
)
debug.print(f"RSA key generated: {bits} bits")
return key
@@ -164,10 +166,7 @@ def generate_rsa_key(bits: int = DEFAULT_RSA_BITS) -> rsa.RSAPrivateKey:
raise KeyGenerationError(f"Failed to generate RSA key: {e}") from e
def export_rsa_key_pem(
private_key: rsa.RSAPrivateKey,
password: str | None = None
) -> bytes:
def export_rsa_key_pem(private_key: rsa.RSAPrivateKey, password: str | None = None) -> bytes:
"""
Export RSA key to PEM format.
@@ -198,14 +197,11 @@ def export_rsa_key_pem(
return private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=encryption_algorithm
encryption_algorithm=encryption_algorithm,
)
def load_rsa_key(
key_data: bytes,
password: str | None = None
) -> rsa.RSAPrivateKey:
def load_rsa_key(key_data: bytes, password: str | None = None) -> rsa.RSAPrivateKey:
"""
Load RSA private key from PEM data.
@@ -223,8 +219,7 @@ def load_rsa_key(
Example:
>>> key = load_rsa_key(pem_data, "my_password")
"""
debug.validate(key_data is not None and len(key_data) > 0,
"Key data cannot be empty")
debug.validate(key_data is not None and len(key_data) > 0, "Key data cannot be empty")
try:
pwd_bytes = password.encode() if password else None
@@ -274,15 +269,11 @@ def get_key_info(key_data: bytes, password: str | None = None) -> KeyInfo:
"""
debug.print("Getting RSA key info")
# Check if encrypted
is_encrypted = b'ENCRYPTED' in key_data
is_encrypted = b"ENCRYPTED" in key_data
private_key = load_rsa_key(key_data, password)
info = KeyInfo(
key_size=private_key.key_size,
is_encrypted=is_encrypted,
pem_data=key_data
)
info = KeyInfo(key_size=private_key.key_size, is_encrypted=is_encrypted, pem_data=key_data)
debug.print(f"Key info: {info.key_size} bits, encrypted: {info.is_encrypted}")
return info
@@ -323,14 +314,15 @@ def generate_credentials(
>>> creds.pin
"812345"
"""
debug.validate(use_pin or use_rsa,
"Must select at least one security factor (PIN or RSA key)")
debug.validate(use_pin or use_rsa, "Must select at least one security factor (PIN or RSA key)")
if not use_pin and not use_rsa:
raise ValueError("Must select at least one security factor (PIN or RSA key)")
debug.print(f"Generating credentials: PIN={use_pin}, RSA={use_rsa}, "
f"passphrase_words={passphrase_words}")
debug.print(
f"Generating credentials: PIN={use_pin}, RSA={use_rsa}, "
f"passphrase_words={passphrase_words}"
)
# Generate single passphrase (v3.2.0 - no daily rotation)
passphrase = generate_phrase(passphrase_words)
@@ -342,7 +334,7 @@ def generate_credentials(
rsa_key_pem = None
if use_rsa:
rsa_key_obj = generate_rsa_key(rsa_bits)
rsa_key_pem = export_rsa_key_pem(rsa_key_obj, rsa_password).decode('utf-8')
rsa_key_pem = export_rsa_key_pem(rsa_key_obj, rsa_password).decode("utf-8")
# Create Credentials object (v3.2.0 format with single passphrase)
creds = Credentials(
@@ -361,12 +353,13 @@ def generate_credentials(
# LEGACY COMPATIBILITY
# =============================================================================
def generate_credentials_legacy(
use_pin: bool = True,
use_rsa: bool = False,
pin_length: int = DEFAULT_PIN_LENGTH,
rsa_bits: int = DEFAULT_RSA_BITS,
words_per_phrase: int = DEFAULT_PASSPHRASE_WORDS
words_per_phrase: int = DEFAULT_PASSPHRASE_WORDS,
) -> dict:
"""
Generate credentials in legacy format (v3.1.0 style with daily phrases).
@@ -387,11 +380,12 @@ def generate_credentials_legacy(
Dict with 'phrases' (dict), 'pin', 'rsa_key_pem', etc.
"""
import warnings
warnings.warn(
"generate_credentials_legacy() returns v3.1.0 format. "
"Use generate_credentials() for v3.2.0 format.",
DeprecationWarning,
stacklevel=2
stacklevel=2,
)
if not use_pin and not use_rsa:
@@ -405,12 +399,12 @@ def generate_credentials_legacy(
rsa_key_pem = None
if use_rsa:
rsa_key_obj = generate_rsa_key(rsa_bits)
rsa_key_pem = export_rsa_key_pem(rsa_key_obj).decode('utf-8')
rsa_key_pem = export_rsa_key_pem(rsa_key_obj).decode("utf-8")
return {
'phrases': phrases,
'pin': pin,
'rsa_key_pem': rsa_key_pem,
'rsa_bits': rsa_bits if use_rsa else None,
'words_per_phrase': words_per_phrase,
"phrases": phrases,
"pin": pin,
"rsa_key_pem": rsa_key_pem,
"rsa_bits": rsa_bits if use_rsa else None,
"words_per_phrase": words_per_phrase,
}

23
src/stegasoo/models.py
View File

@@ -21,6 +21,7 @@ class Credentials:
v3.2.0: Simplified to use single passphrase instead of daily rotation.
"""
passphrase: str # Single passphrase (no daily rotation)
pin: str | None = None
rsa_key_pem: str | None = None
@@ -64,6 +65,7 @@ class Credentials:
@dataclass
class FilePayload:
"""Represents a file to be embedded."""
data: bytes
filename: str
mime_type: str | None = None
@@ -73,7 +75,7 @@ class FilePayload:
return len(self.data)
@classmethod
def from_file(cls, filepath: str, filename: str | None = None) -> 'FilePayload':
def from_file(cls, filepath: str, filename: str | None = None) -> "FilePayload":
"""Create FilePayload from a file path."""
import mimetypes
from pathlib import Path
@@ -93,6 +95,7 @@ class EncodeInput:
v3.2.0: Removed date_str (date no longer used in crypto).
"""
message: str | bytes | FilePayload # Text, raw bytes, or file
reference_photo: bytes
carrier_image: bytes
@@ -109,6 +112,7 @@ class EncodeResult:
v3.2.0: date_used is now optional/cosmetic (not used in crypto).
"""
stego_image: bytes
filename: str
pixels_modified: int
@@ -129,6 +133,7 @@ class DecodeInput:
v3.2.0: Renamed day_phrase → passphrase, no date needed.
"""
stego_image: bytes
reference_photo: bytes
passphrase: str # Renamed from day_phrase
@@ -144,6 +149,7 @@ class DecodeResult:
v3.2.0: date_encoded is always None (date removed from crypto).
"""
payload_type: str # 'text' or 'file'
message: str | None = None # For text payloads
file_data: bytes | None = None # For file payloads
@@ -153,11 +159,11 @@ class DecodeResult:
@property
def is_file(self) -> bool:
return self.payload_type == 'file'
return self.payload_type == "file"
@property
def is_text(self) -> bool:
return self.payload_type == 'text'
return self.payload_type == "text"
def get_content(self) -> str | bytes:
"""Get the decoded content (text or bytes)."""
@@ -169,6 +175,7 @@ class DecodeResult:
@dataclass
class EmbedStats:
"""Statistics from image embedding."""
pixels_modified: int
total_pixels: int
capacity_used: float
@@ -183,6 +190,7 @@ class EmbedStats:
@dataclass
class KeyInfo:
"""Information about an RSA key."""
key_size: int
is_encrypted: bool
pem_data: bytes
@@ -191,13 +199,14 @@ class KeyInfo:
@dataclass
class ValidationResult:
"""Result of input validation."""
is_valid: bool
error_message: str = ""
details: dict = field(default_factory=dict)
warning: str | None = None # v3.2.0: Added for passphrase length warnings
@classmethod
def ok(cls, warning: str | None = None, **details) -> 'ValidationResult':
def ok(cls, warning: str | None = None, **details) -> "ValidationResult":
"""Create a successful validation result."""
result = cls(is_valid=True, details=details)
if warning:
@@ -205,7 +214,7 @@ class ValidationResult:
return result
@classmethod
def error(cls, message: str, **details) -> 'ValidationResult':
def error(cls, message: str, **details) -> "ValidationResult":
"""Create a failed validation result."""
return cls(is_valid=False, error_message=message, details=details)
@@ -214,9 +223,11 @@ class ValidationResult:
# NEW MODELS FOR V3.2.0 PUBLIC API
# =============================================================================
@dataclass
class ImageInfo:
"""Information about an image for steganography."""
width: int
height: int
pixels: int
@@ -232,6 +243,7 @@ class ImageInfo:
@dataclass
class CapacityComparison:
"""Comparison of embedding capacity between modes."""
image_width: int
image_height: int
lsb_available: bool
@@ -248,6 +260,7 @@ class CapacityComparison:
@dataclass
class GenerateResult:
"""Result of credential generation."""
passphrase: str
pin: str | None = None
rsa_key_pem: str | None = None

89
src/stegasoo/qr_utils.py
View File

@@ -20,6 +20,7 @@ from PIL import Image
try:
import qrcode
from qrcode.constants import ERROR_CORRECT_L, ERROR_CORRECT_M
HAS_QRCODE_WRITE = True
except ImportError:
HAS_QRCODE_WRITE = False
@@ -28,6 +29,7 @@ except ImportError:
try:
from pyzbar.pyzbar import ZBarSymbol
from pyzbar.pyzbar import decode as pyzbar_decode
HAS_QRCODE_READ = True
except ImportError:
HAS_QRCODE_READ = False
@@ -53,8 +55,8 @@ def compress_data(data: str) -> str:
Returns:
Compressed string with STEGASOO-Z: prefix
"""
compressed = zlib.compress(data.encode('utf-8'), level=9)
encoded = base64.b64encode(compressed).decode('ascii')
compressed = zlib.compress(data.encode("utf-8"), level=9)
encoded = base64.b64encode(compressed).decode("ascii")
return COMPRESSION_PREFIX + encoded
@@ -74,9 +76,9 @@ def decompress_data(data: str) -> str:
if not data.startswith(COMPRESSION_PREFIX):
raise ValueError("Data is not in compressed format")
encoded = data[len(COMPRESSION_PREFIX):]
encoded = data[len(COMPRESSION_PREFIX) :]
compressed = base64.b64decode(encoded)
return zlib.decompress(compressed).decode('utf-8')
return zlib.decompress(compressed).decode("utf-8")
def normalize_pem(pem_data: str) -> str:
@@ -101,25 +103,25 @@ def normalize_pem(pem_data: str) -> str:
import re
# Step 1: Normalize ALL line endings to \n
pem_data = pem_data.replace('\r\n', '\n').replace('\r', '\n')
pem_data = pem_data.replace("\r\n", "\n").replace("\r", "\n")
# Step 2: Remove leading/trailing whitespace
pem_data = pem_data.strip()
# Step 3: Remove any non-ASCII characters (QR artifacts)
pem_data = ''.join(char for char in pem_data if ord(char) < 128)
pem_data = "".join(char for char in pem_data if ord(char) < 128)
# Step 4: Extract header, content, and footer with flexible regex
# This handles variations like:
# - "PRIVATE KEY" vs "RSA PRIVATE KEY"
# - Extra spaces in headers
# - Missing spaces
pattern = r'(-----BEGIN[^-]*-----)(.*?)(-----END[^-]*-----)'
pattern = r"(-----BEGIN[^-]*-----)(.*?)(-----END[^-]*-----)"
match = re.search(pattern, pem_data, re.DOTALL | re.IGNORECASE)
if not match:
# Fallback: try even more permissive pattern
pattern = r'(-+BEGIN[^-]+-+)(.*?)(-+END[^-]+-+)'
pattern = r"(-+BEGIN[^-]+-+)(.*?)(-+END[^-]+-+)"
match = re.search(pattern, pem_data, re.DOTALL | re.IGNORECASE)
if not match:
@@ -132,38 +134,35 @@ def normalize_pem(pem_data: str) -> str:
# Step 5: Normalize header and footer
# Standardize spacing and ensure proper format
header = re.sub(r'\s+', ' ', header_raw)
footer = re.sub(r'\s+', ' ', footer_raw)
header = re.sub(r"\s+", " ", header_raw)
footer = re.sub(r"\s+", " ", footer_raw)
# Ensure exactly 5 dashes on each side
header = re.sub(r'^-+', '-----', header)
header = re.sub(r'-+$', '-----', header)
footer = re.sub(r'^-+', '-----', footer)
footer = re.sub(r'-+$', '-----', footer)
header = re.sub(r"^-+", "-----", header)
header = re.sub(r"-+$", "-----", header)
footer = re.sub(r"^-+", "-----", footer)
footer = re.sub(r"-+$", "-----", footer)
# Step 6: Clean the base64 content THOROUGHLY
# Remove ALL whitespace: spaces, tabs, newlines
# Keep only valid base64 characters: A-Z, a-z, 0-9, +, /, =
content_clean = ''.join(
char for char in content_raw
if char.isalnum() or char in '+/='
)
content_clean = "".join(char for char in content_raw if char.isalnum() or char in "+/=")
# Double-check: remove any remaining invalid characters
content_clean = re.sub(r'[^A-Za-z0-9+/=]', '', content_clean)
content_clean = re.sub(r"[^A-Za-z0-9+/=]", "", content_clean)
# Step 7: Fix base64 padding
# Base64 strings must be divisible by 4
remainder = len(content_clean) % 4
if remainder:
content_clean += '=' * (4 - remainder)
content_clean += "=" * (4 - remainder)
# Step 8: Split into 64-character lines (PEM standard)
lines = [content_clean[i:i+64] for i in range(0, len(content_clean), 64)]
lines = [content_clean[i : i + 64] for i in range(0, len(content_clean), 64)]
# Step 9: Reconstruct with EXACT PEM formatting
# Format: header\ncontent_line1\ncontent_line2\n...\nfooter\n
return header + '\n' + '\n'.join(lines) + '\n' + footer + '\n'
return header + "\n" + "\n".join(lines) + "\n" + footer + "\n"
def is_compressed(data: str) -> bool:
@@ -205,7 +204,7 @@ def can_fit_in_qr(data: str, compress: bool = False) -> bool:
if compress:
size = get_compressed_size(data)
else:
size = len(data.encode('utf-8'))
size = len(data.encode("utf-8"))
return size <= QR_MAX_BINARY
@@ -214,11 +213,7 @@ def needs_compression(data: str) -> bool:
return not can_fit_in_qr(data, compress=False) and can_fit_in_qr(data, compress=True)
def generate_qr_code(
data: str,
compress: bool = False,
error_correction=None
) -> bytes:
def generate_qr_code(data: str, compress: bool = False, error_correction=None) -> bytes:
"""
Generate a QR code PNG from string data.
@@ -244,10 +239,9 @@ def generate_qr_code(
qr_data = compress_data(data)
# Check size
if len(qr_data.encode('utf-8')) > QR_MAX_BINARY:
if len(qr_data.encode("utf-8")) > QR_MAX_BINARY:
raise ValueError(
f"Data too large for QR code ({len(qr_data)} bytes). "
f"Maximum: {QR_MAX_BINARY} bytes"
f"Data too large for QR code ({len(qr_data)} bytes). " f"Maximum: {QR_MAX_BINARY} bytes"
)
# Use lower error correction for larger data
@@ -266,7 +260,7 @@ def generate_qr_code(
img = qr.make_image(fill_color="black", back_color="white")
buf = io.BytesIO()
img.save(buf, format='PNG')
img.save(buf, format="PNG")
buf.seek(0)
return buf.getvalue()
@@ -294,8 +288,8 @@ def read_qr_code(image_data: bytes) -> str | None:
img: Image.Image = Image.open(io.BytesIO(image_data))
# Convert to RGB if necessary (pyzbar works best with RGB/grayscale)
if img.mode not in ('RGB', 'L'):
img = img.convert('RGB')
if img.mode not in ("RGB", "L"):
img = img.convert("RGB")
# Decode QR codes
decoded = pyzbar_decode(img, symbols=[ZBarSymbol.QRCODE])
@@ -304,7 +298,7 @@ def read_qr_code(image_data: bytes) -> str | None:
return None
# Return first QR code found
result: str = decoded[0].data.decode('utf-8')
result: str = decoded[0].data.decode("utf-8")
return result
except Exception:
@@ -321,7 +315,7 @@ def read_qr_code_from_file(filepath: str) -> str | None:
Returns:
Decoded string, or None if no QR code found
"""
with open(filepath, 'rb') as f:
with open(filepath, "rb") as f:
return read_qr_code(f.read())
@@ -355,7 +349,7 @@ def extract_key_from_qr(image_data: bytes) -> str | None:
key_pem = qr_data
# Step 3: Validate it looks like a PEM key
if '-----BEGIN' not in key_pem or '-----END' not in key_pem:
if "-----BEGIN" not in key_pem or "-----END" not in key_pem:
return None
# Step 4: Aggressively normalize PEM format
@@ -367,7 +361,7 @@ def extract_key_from_qr(image_data: bytes) -> str | None:
return None
# Step 5: Final validation - ensure it still looks like PEM
if '-----BEGIN' in key_pem and '-----END' in key_pem:
if "-----BEGIN" in key_pem and "-----END" in key_pem:
return key_pem
return None
@@ -383,14 +377,14 @@ def extract_key_from_qr_file(filepath: str) -> str | None:
Returns:
PEM-encoded RSA key string, or None if not found/invalid
"""
with open(filepath, 'rb') as f:
with open(filepath, "rb") as f:
return extract_key_from_qr(f.read())
def detect_and_crop_qr(
image_data: bytes,
padding_percent: float = QR_CROP_PADDING_PERCENT,
min_padding_px: int = QR_CROP_MIN_PADDING_PX
min_padding_px: int = QR_CROP_MIN_PADDING_PX,
) -> bytes | None:
"""
Detect QR code in image and crop to it, handling rotation.
@@ -420,8 +414,8 @@ def detect_and_crop_qr(
original_mode = img.mode
# Convert for pyzbar detection
if img.mode not in ('RGB', 'L'):
detect_img = img.convert('RGB')
if img.mode not in ("RGB", "L"):
detect_img = img.convert("RGB")
else:
detect_img = img
@@ -468,16 +462,17 @@ def detect_and_crop_qr(
# Convert to PNG bytes
buf = io.BytesIO()
# Preserve transparency if present
if original_mode in ('RGBA', 'LA', 'P'):
cropped.save(buf, format='PNG')
if original_mode in ("RGBA", "LA", "P"):
cropped.save(buf, format="PNG")
else:
cropped.save(buf, format='PNG')
cropped.save(buf, format="PNG")
buf.seek(0)
return buf.getvalue()
except Exception as e:
# Log for debugging but return None for clean API
import sys
print(f"QR crop error: {e}", file=sys.stderr)
return None
@@ -485,7 +480,7 @@ def detect_and_crop_qr(
def detect_and_crop_qr_file(
filepath: str,
padding_percent: float = QR_CROP_PADDING_PERCENT,
min_padding_px: int = QR_CROP_MIN_PADDING_PX
min_padding_px: int = QR_CROP_MIN_PADDING_PX,
) -> bytes | None:
"""
Detect QR code in image file and crop to it.
@@ -498,7 +493,7 @@ def detect_and_crop_qr_file(
Returns:
Cropped PNG image bytes, or None if no QR code found
"""
with open(filepath, 'rb') as f:
with open(filepath, "rb") as f:
return detect_and_crop_qr(f.read(), padding_percent, min_padding_px)

270
src/stegasoo/steganography.py
View File

@@ -40,21 +40,21 @@ from .exceptions import CapacityError, EmbeddingError
from .models import EmbedStats, FilePayload
# Lossless formats that preserve LSB data
LOSSLESS_FORMATS = {'PNG', 'BMP', 'TIFF'}
LOSSLESS_FORMATS = {"PNG", "BMP", "TIFF"}
# Format to extension mapping
FORMAT_TO_EXT = {
'PNG': 'png',
'BMP': 'bmp',
'TIFF': 'tiff',
"PNG": "png",
"BMP": "bmp",
"TIFF": "tiff",
}
# Extension to PIL format mapping
EXT_TO_FORMAT = {
'png': 'PNG',
'bmp': 'BMP',
'tiff': 'TIFF',
'tif': 'TIFF',
"png": "PNG",
"bmp": "BMP",
"tiff": "TIFF",
"tif": "TIFF",
}
# =============================================================================
@@ -73,17 +73,17 @@ EXT_TO_FORMAT = {
# v3.2.0 had 65 bytes (no flags byte)
# v3.1.0 had date field (10 bytes + 1 byte length) = 76 bytes header
HEADER_OVERHEAD = 66 # v4.0.0: Magic + version + flags + salt + iv + tag
LENGTH_PREFIX = 4 # 4 bytes for payload length in LSB embedding
HEADER_OVERHEAD = 66 # v4.0.0: Magic + version + flags + salt + iv + tag
LENGTH_PREFIX = 4 # 4 bytes for payload length in LSB embedding
ENCRYPTION_OVERHEAD = HEADER_OVERHEAD + LENGTH_PREFIX # 70 bytes total
# DCT output format options (v3.0.1)
DCT_OUTPUT_PNG = 'png'
DCT_OUTPUT_JPEG = 'jpeg'
DCT_OUTPUT_PNG = "png"
DCT_OUTPUT_JPEG = "jpeg"
# DCT color mode options (v3.0.1)
DCT_COLOR_GRAYSCALE = 'grayscale'
DCT_COLOR_COLOR = 'color'
DCT_COLOR_GRAYSCALE = "grayscale"
DCT_COLOR_COLOR = "color"
# =============================================================================
@@ -98,6 +98,7 @@ def _get_dct_module():
global _dct_module
if _dct_module is None:
from . import dct_steganography
_dct_module = dct_steganography
return _dct_module
@@ -124,6 +125,7 @@ def has_dct_support() -> bool:
# FORMAT UTILITIES
# =============================================================================
def get_output_format(input_format: str | None) -> tuple[str, str]:
"""
Determine the output format based on input format.
@@ -135,23 +137,25 @@ def get_output_format(input_format: str | None) -> tuple[str, str]:
Tuple of (PIL format string, file extension) for output
Falls back to PNG for lossy or unknown formats.
"""
debug.validate(input_format is None or isinstance(input_format, str),
"Input format must be string or None")
debug.validate(
input_format is None or isinstance(input_format, str), "Input format must be string or None"
)
if input_format and input_format.upper() in LOSSLESS_FORMATS:
fmt = input_format.upper()
ext = FORMAT_TO_EXT.get(fmt, 'png')
ext = FORMAT_TO_EXT.get(fmt, "png")
debug.print(f"Using lossless format: {fmt} -> .{ext}")
return fmt, ext
debug.print(f"Input format {input_format} is lossy or unknown, defaulting to PNG")
return 'PNG', 'png'
return "PNG", "png"
# =============================================================================
# CAPACITY FUNCTIONS
# =============================================================================
def will_fit(
payload: str | bytes | FilePayload | int,
carrier_image: bytes,
@@ -175,12 +179,12 @@ def will_fit(
payload_size = payload
payload_data = None
elif isinstance(payload, str):
payload_data = payload.encode('utf-8')
payload_data = payload.encode("utf-8")
payload_size = len(payload_data)
elif isinstance(payload, FilePayload):
payload_data = payload.data
filename_overhead = len(payload.filename.encode('utf-8')) if payload.filename else 0
mime_overhead = len(payload.mime_type.encode('utf-8')) if payload.mime_type else 0
filename_overhead = len(payload.filename.encode("utf-8")) if payload.filename else 0
mime_overhead = len(payload.mime_type.encode("utf-8")) if payload.mime_type else 0
payload_size = len(payload.data) + filename_overhead + mime_overhead + 5
else:
payload_data = payload
@@ -198,6 +202,7 @@ def will_fit(
if include_compression_estimate and payload_data is not None and len(payload_data) >= 64:
try:
import zlib
compressed = zlib.compress(payload_data, level=6)
compressed_size = len(compressed) + 9 # Compression header
if compressed_size < payload_size:
@@ -211,14 +216,14 @@ def will_fit(
usage_percent = (estimated_encrypted_size / capacity * 100) if capacity > 0 else 100.0
return {
'fits': fits,
'payload_size': payload_size,
'estimated_encrypted_size': estimated_encrypted_size,
'capacity': capacity,
'usage_percent': min(usage_percent, 100.0),
'headroom': headroom,
'compressed_estimate': compressed_estimate,
'mode': EMBED_MODE_LSB,
"fits": fits,
"payload_size": payload_size,
"estimated_encrypted_size": estimated_encrypted_size,
"capacity": capacity,
"usage_percent": min(usage_percent, 100.0),
"headroom": headroom,
"compressed_estimate": compressed_estimate,
"mode": EMBED_MODE_LSB,
}
@@ -233,8 +238,9 @@ def calculate_capacity(image_data: bytes, bits_per_channel: int = 1) -> int:
Returns:
Maximum bytes that can be embedded (minus overhead)
"""
debug.validate(bits_per_channel in (1, 2),
f"bits_per_channel must be 1 or 2, got {bits_per_channel}")
debug.validate(
bits_per_channel in (1, 2), f"bits_per_channel must be 1 or 2, got {bits_per_channel}"
)
img_file = Image.open(io.BytesIO(image_data))
try:
@@ -273,12 +279,12 @@ def calculate_capacity_by_mode(
dct_info = dct_mod.calculate_dct_capacity(image_data)
return {
'mode': EMBED_MODE_DCT,
'capacity_bytes': dct_info.usable_capacity_bytes,
'capacity_bits': dct_info.total_capacity_bits,
'width': dct_info.width,
'height': dct_info.height,
'total_blocks': dct_info.total_blocks,
"mode": EMBED_MODE_DCT,
"capacity_bytes": dct_info.usable_capacity_bytes,
"capacity_bits": dct_info.total_capacity_bits,
"width": dct_info.width,
"height": dct_info.height,
"total_blocks": dct_info.total_blocks,
}
else:
capacity = calculate_capacity(image_data, bits_per_channel)
@@ -289,12 +295,12 @@ def calculate_capacity_by_mode(
img.close()
return {
'mode': EMBED_MODE_LSB,
'capacity_bytes': capacity,
'capacity_bits': capacity * 8,
'width': width,
'height': height,
'bits_per_channel': bits_per_channel,
"mode": EMBED_MODE_LSB,
"capacity_bytes": capacity,
"capacity_bits": capacity * 8,
"width": width,
"height": height,
"bits_per_channel": bits_per_channel,
}
@@ -318,13 +324,13 @@ def will_fit_by_mode(
"""
if embed_mode == EMBED_MODE_DCT:
if not has_dct_support():
return {'fits': False, 'error': 'scipy not available', 'mode': EMBED_MODE_DCT}
return {"fits": False, "error": "scipy not available", "mode": EMBED_MODE_DCT}
if isinstance(payload, int):
payload_size = payload
elif isinstance(payload, str):
payload_size = len(payload.encode('utf-8'))
elif hasattr(payload, 'data'):
payload_size = len(payload.encode("utf-8"))
elif hasattr(payload, "data"):
payload_size = len(payload.data)
else:
payload_size = len(payload)
@@ -339,12 +345,12 @@ def will_fit_by_mode(
usage_percent = (estimated_size / capacity * 100) if capacity > 0 else 100.0
return {
'fits': fits,
'payload_size': payload_size,
'capacity': capacity,
'usage_percent': min(usage_percent, 100.0),
'headroom': capacity - estimated_size,
'mode': EMBED_MODE_DCT,
"fits": fits,
"payload_size": payload_size,
"capacity": capacity,
"usage_percent": min(usage_percent, 100.0),
"headroom": capacity - estimated_size,
"mode": EMBED_MODE_DCT,
}
else:
return will_fit(payload, carrier_image, bits_per_channel)
@@ -359,17 +365,17 @@ def get_available_modes() -> dict:
"""
return {
EMBED_MODE_LSB: {
'available': True,
'name': 'Spatial LSB',
'description': 'Embed in pixel LSBs, outputs PNG/BMP',
'output_format': 'PNG (color)',
"available": True,
"name": "Spatial LSB",
"description": "Embed in pixel LSBs, outputs PNG/BMP",
"output_format": "PNG (color)",
},
EMBED_MODE_DCT: {
'available': has_dct_support(),
'name': 'DCT Domain',
'description': 'Embed in DCT coefficients, outputs grayscale PNG or JPEG',
'output_formats': ['PNG (grayscale)', 'JPEG (grayscale)'],
'requires': 'scipy',
"available": has_dct_support(),
"name": "DCT Domain",
"description": "Embed in DCT coefficients, outputs grayscale PNG or JPEG",
"output_formats": ["PNG (grayscale)", "JPEG (grayscale)"],
"requires": "scipy",
},
}
@@ -403,20 +409,20 @@ def compare_modes(image_data: bytes) -> dict:
dct_available = False
return {
'width': width,
'height': height,
'lsb': {
'capacity_bytes': lsb_bytes,
'capacity_kb': lsb_bytes / 1024,
'available': True,
'output': 'PNG (color)',
"width": width,
"height": height,
"lsb": {
"capacity_bytes": lsb_bytes,
"capacity_kb": lsb_bytes / 1024,
"available": True,
"output": "PNG (color)",
},
'dct': {
'capacity_bytes': dct_bytes,
'capacity_kb': dct_bytes / 1024,
'available': dct_available,
'output': 'PNG or JPEG (grayscale)',
'ratio_vs_lsb': (dct_bytes / lsb_bytes * 100) if lsb_bytes > 0 else 0,
"dct": {
"capacity_bytes": dct_bytes,
"capacity_kb": dct_bytes / 1024,
"available": dct_available,
"output": "PNG or JPEG (grayscale)",
"ratio_vs_lsb": (dct_bytes / lsb_bytes * 100) if lsb_bytes > 0 else 0,
},
}
@@ -425,6 +431,7 @@ def compare_modes(image_data: bytes) -> dict:
# PIXEL INDEX GENERATION
# =============================================================================
@debug.time
def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list[int]:
"""
@@ -436,23 +443,24 @@ def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list
debug.validate(len(key) == 32, f"Pixel key must be 32 bytes, got {len(key)}")
debug.validate(num_pixels > 0, f"Number of pixels must be positive, got {num_pixels}")
debug.validate(num_needed > 0, f"Number needed must be positive, got {num_needed}")
debug.validate(num_needed <= num_pixels,
f"Cannot select {num_needed} pixels from {num_pixels} available")
debug.validate(
num_needed <= num_pixels, f"Cannot select {num_needed} pixels from {num_pixels} available"
)
debug.print(f"Generating {num_needed} pixel indices from {num_pixels} total pixels")
if num_needed >= num_pixels // 2:
debug.print(f"Using full shuffle (needed {num_needed}/{num_pixels} pixels)")
nonce = b'\x00' * 16
nonce = b"\x00" * 16
cipher = Cipher(algorithms.ChaCha20(key, nonce), mode=None, backend=default_backend())
encryptor = cipher.encryptor()
indices = list(range(num_pixels))
random_bytes = encryptor.update(b'\x00' * (num_pixels * 4))
random_bytes = encryptor.update(b"\x00" * (num_pixels * 4))
for i in range(num_pixels - 1, 0, -1):
j_bytes = random_bytes[(num_pixels - 1 - i) * 4:(num_pixels - i) * 4]
j = int.from_bytes(j_bytes, 'big') % (i + 1)
j_bytes = random_bytes[(num_pixels - 1 - i) * 4 : (num_pixels - i) * 4]
j = int.from_bytes(j_bytes, "big") % (i + 1)
indices[i], indices[j] = indices[j], indices[i]
selected = indices[:num_needed]
@@ -463,17 +471,17 @@ def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list
selected = []
used = set()
nonce = b'\x00' * 16
nonce = b"\x00" * 16
cipher = Cipher(algorithms.ChaCha20(key, nonce), mode=None, backend=default_backend())
encryptor = cipher.encryptor()
bytes_needed = (num_needed * 2) * 4
random_bytes = encryptor.update(b'\x00' * bytes_needed)
random_bytes = encryptor.update(b"\x00" * bytes_needed)
byte_offset = 0
collisions = 0
while len(selected) < num_needed and byte_offset < len(random_bytes) - 4:
idx = int.from_bytes(random_bytes[byte_offset:byte_offset + 4], 'big') % num_pixels
idx = int.from_bytes(random_bytes[byte_offset : byte_offset + 4], "big") % num_pixels
byte_offset += 4
if idx not in used:
@@ -486,8 +494,8 @@ def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list
debug.print(f"Need {num_needed - len(selected)} more indices, generating...")
extra_needed = num_needed - len(selected)
for _ in range(extra_needed * 2):
extra_bytes = encryptor.update(b'\x00' * 4)
idx = int.from_bytes(extra_bytes, 'big') % num_pixels
extra_bytes = encryptor.update(b"\x00" * 4)
idx = int.from_bytes(extra_bytes, "big") % num_pixels
if idx not in used:
used.add(idx)
selected.append(idx)
@@ -495,8 +503,10 @@ def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list
break
debug.print(f"Generated {len(selected)} indices with {collisions} collisions")
debug.validate(len(selected) == num_needed,
f"Failed to generate enough indices: {len(selected)}/{num_needed}")
debug.validate(
len(selected) == num_needed,
f"Failed to generate enough indices: {len(selected)}/{num_needed}",
)
return selected
@@ -504,6 +514,7 @@ def generate_pixel_indices(key: bytes, num_pixels: int, num_needed: int) -> list
# EMBEDDING FUNCTIONS
# =============================================================================
@debug.time
def embed_in_image(
data: bytes,
@@ -513,8 +524,8 @@ def embed_in_image(
output_format: str | None = None,
embed_mode: str = EMBED_MODE_LSB,
dct_output_format: str = DCT_OUTPUT_PNG,
dct_color_mode: str = 'grayscale',
) -> tuple[bytes, Union[EmbedStats, 'DCTEmbedStats'], str]:
dct_color_mode: str = "grayscale",
) -> tuple[bytes, Union[EmbedStats, "DCTEmbedStats"], str]:
"""
Embed data into an image using specified mode.
@@ -537,15 +548,15 @@ def embed_in_image(
ImportError: If DCT mode requested but scipy unavailable
"""
debug.print(f"embed_in_image: mode={embed_mode}, data={len(data)} bytes")
debug.validate(embed_mode in VALID_EMBED_MODES,
f"Invalid embed_mode: {embed_mode}. Use 'lsb' or 'dct'")
debug.validate(
embed_mode in VALID_EMBED_MODES, f"Invalid embed_mode: {embed_mode}. Use 'lsb' or 'dct'"
)
# DCT MODE
if embed_mode == EMBED_MODE_DCT:
if not has_dct_support():
raise ImportError(
"scipy is required for DCT embedding mode. "
"Install with: pip install scipy"
"scipy is required for DCT embedding mode. " "Install with: pip install scipy"
)
# Validate DCT output format
@@ -554,9 +565,9 @@ def embed_in_image(
dct_output_format = DCT_OUTPUT_PNG
# Validate DCT color mode (v3.0.1)
if dct_color_mode not in ('grayscale', 'color'):
if dct_color_mode not in ("grayscale", "color"):
debug.print(f"Invalid dct_color_mode '{dct_color_mode}', defaulting to grayscale")
dct_color_mode = 'grayscale'
dct_color_mode = "grayscale"
dct_mod = _get_dct_module()
@@ -571,12 +582,14 @@ def embed_in_image(
# Determine extension based on output format
if dct_output_format == DCT_OUTPUT_JPEG:
ext = 'jpg'
ext = "jpg"
else:
ext = 'png'
ext = "png"
debug.print(f"DCT embedding complete: {dct_output_format.upper()} output, "
f"color_mode={dct_color_mode}, ext={ext}")
debug.print(
f"DCT embedding complete: {dct_output_format.upper()} output, "
f"color_mode={dct_color_mode}, ext={ext}"
)
return stego_bytes, dct_stats, ext
# LSB MODE
@@ -595,10 +608,10 @@ def _embed_lsb(
"""
debug.print(f"LSB embedding {len(data)} bytes into image")
debug.data(pixel_key, "Pixel key for embedding")
debug.validate(bits_per_channel in (1, 2),
f"bits_per_channel must be 1 or 2, got {bits_per_channel}")
debug.validate(len(pixel_key) == 32,
f"Pixel key must be 32 bytes, got {len(pixel_key)}")
debug.validate(
bits_per_channel in (1, 2), f"bits_per_channel must be 1 or 2, got {bits_per_channel}"
)
debug.validate(len(pixel_key) == 32, f"Pixel key must be 32 bytes, got {len(pixel_key)}")
img_file = None
img = None
@@ -610,8 +623,8 @@ def _embed_lsb(
debug.print(f"Carrier image: {img_file.size[0]}x{img_file.size[1]}, format: {input_format}")
img = img_file.convert('RGB') if img_file.mode != 'RGB' else img_file.copy()
if img_file.mode != 'RGB':
img = img_file.convert("RGB") if img_file.mode != "RGB" else img_file.copy()
if img_file.mode != "RGB":
debug.print(f"Converting image from {img_file.mode} to RGB")
pixels = list(img.getdata())
@@ -622,16 +635,18 @@ def _embed_lsb(
debug.print(f"Image capacity: {max_bytes} bytes at {bits_per_channel} bit(s)/channel")
data_with_len = struct.pack('>I', len(data)) + data
data_with_len = struct.pack(">I", len(data)) + data
if len(data_with_len) > max_bytes:
debug.print(f"Capacity error: need {len(data_with_len)}, have {max_bytes}")
raise CapacityError(len(data_with_len), max_bytes)
debug.print(f"Total data to embed: {len(data_with_len)} bytes "
f"({len(data_with_len)/max_bytes*100:.1f}% of capacity)")
debug.print(
f"Total data to embed: {len(data_with_len)} bytes "
f"({len(data_with_len)/max_bytes*100:.1f}% of capacity)"
)
binary_data = ''.join(format(b, '08b') for b in data_with_len)
binary_data = "".join(format(b, "08b") for b in data_with_len)
pixels_needed = (len(binary_data) + bits_per_pixel - 1) // bits_per_pixel
debug.print(f"Need {pixels_needed} pixels to embed {len(binary_data)} bits")
@@ -654,7 +669,9 @@ def _embed_lsb(
for channel_idx, channel_val in enumerate([r, g, b]):
if bit_idx >= len(binary_data):
break
bits = binary_data[bit_idx:bit_idx + bits_per_channel].ljust(bits_per_channel, '0')
bits = binary_data[bit_idx : bit_idx + bits_per_channel].ljust(
bits_per_channel, "0"
)
new_val = (channel_val & clear_mask) | int(bits, 2)
if channel_val != new_val:
@@ -674,12 +691,12 @@ def _embed_lsb(
debug.print(f"Modified {modified_pixels} pixels (out of {len(selected_indices)} selected)")
stego_img = Image.new('RGB', img.size)
stego_img = Image.new("RGB", img.size)
stego_img.putdata(new_pixels)
if output_format:
out_fmt = output_format.upper()
out_ext = FORMAT_TO_EXT.get(out_fmt, 'png')
out_ext = FORMAT_TO_EXT.get(out_fmt, "png")
debug.print(f"Using forced output format: {out_fmt}")
else:
out_fmt, out_ext = get_output_format(input_format)
@@ -693,7 +710,7 @@ def _embed_lsb(
pixels_modified=modified_pixels,
total_pixels=num_pixels,
capacity_used=len(data_with_len) / max_bytes,
bytes_embedded=len(data_with_len)
bytes_embedded=len(data_with_len),
)
debug.print(f"LSB embedding complete: {out_fmt} image, {len(output.getvalue())} bytes")
@@ -718,6 +735,7 @@ def _embed_lsb(
# EXTRACTION FUNCTIONS
# =============================================================================
@debug.time
def extract_from_image(
image_data: bytes,
@@ -777,18 +795,15 @@ def _extract_dct(image_data: bytes, pixel_key: bytes) -> bytes | None:
return None
def _extract_lsb(
image_data: bytes,
pixel_key: bytes,
bits_per_channel: int = 1
) -> bytes | None:
def _extract_lsb(image_data: bytes, pixel_key: bytes, bits_per_channel: int = 1) -> bytes | None:
"""
Extract using LSB mode (internal implementation).
"""
debug.print(f"LSB extracting from {len(image_data)} byte image")
debug.data(pixel_key, "Pixel key for extraction")
debug.validate(bits_per_channel in (1, 2),
f"bits_per_channel must be 1 or 2, got {bits_per_channel}")
debug.validate(
bits_per_channel in (1, 2), f"bits_per_channel must be 1 or 2, got {bits_per_channel}"
)
img_file = None
img = None
@@ -797,8 +812,8 @@ def _extract_lsb(
img_file = Image.open(io.BytesIO(image_data))
debug.print(f"Image: {img_file.size[0]}x{img_file.size[1]}, format: {img_file.format}")
img = img_file.convert('RGB') if img_file.mode != 'RGB' else img_file.copy()
if img_file.mode != 'RGB':
img = img_file.convert("RGB") if img_file.mode != "RGB" else img_file.copy()
if img_file.mode != "RGB":
debug.print(f"Converting image from {img_file.mode} to RGB")
pixels = list(img.getdata())
@@ -812,7 +827,7 @@ def _extract_lsb(
initial_indices = generate_pixel_indices(pixel_key, num_pixels, initial_pixels)
binary_data = ''
binary_data = ""
for pixel_idx in initial_indices:
r, g, b = pixels[pixel_idx]
for channel in [r, g, b]:
@@ -825,7 +840,7 @@ def _extract_lsb(
debug.print(f"Not enough bits for length: {len(length_bits)}/32")
return None
data_length = struct.unpack('>I', int(length_bits, 2).to_bytes(4, 'big'))[0]
data_length = struct.unpack(">I", int(length_bits, 2).to_bytes(4, "big"))[0]
debug.print(f"Extracted length: {data_length} bytes")
except Exception as e:
debug.print(f"Failed to parse length: {e}")
@@ -843,14 +858,14 @@ def _extract_lsb(
selected_indices = generate_pixel_indices(pixel_key, num_pixels, pixels_needed)
binary_data = ''
binary_data = ""
for pixel_idx in selected_indices:
r, g, b = pixels[pixel_idx]
for channel in [r, g, b]:
for bit_pos in range(bits_per_channel - 1, -1, -1):
binary_data += str((channel >> bit_pos) & 1)
data_bits = binary_data[32:32 + (data_length * 8)]
data_bits = binary_data[32 : 32 + (data_length * 8)]
if len(data_bits) < data_length * 8:
debug.print(f"Insufficient bits: {len(data_bits)} < {data_length * 8}")
@@ -858,7 +873,7 @@ def _extract_lsb(
data_bytes = bytearray()
for i in range(0, len(data_bits), 8):
byte_bits = data_bits[i:i + 8]
byte_bits = data_bits[i : i + 8]
if len(byte_bits) == 8:
data_bytes.append(int(byte_bits, 2))
@@ -880,6 +895,7 @@ def _extract_lsb(
# UTILITY FUNCTIONS
# =============================================================================
def get_image_dimensions(image_data: bytes) -> tuple[int, int]:
"""Get image dimensions without loading full image."""
debug.validate(len(image_data) > 0, "Image data cannot be empty")

46
src/stegasoo/utils.py
View File

@@ -18,7 +18,7 @@ from .constants import DAY_NAMES
from .debug import debug
def strip_image_metadata(image_data: bytes, output_format: str = 'PNG') -> bytes:
def strip_image_metadata(image_data: bytes, output_format: str = "PNG") -> bytes:
"""
Remove all metadata (EXIF, ICC profiles, etc.) from an image.
@@ -41,8 +41,8 @@ def strip_image_metadata(image_data: bytes, output_format: str = 'PNG') -> bytes
img = Image.open(io.BytesIO(image_data))
# Convert to RGB if needed (handles RGBA, P, L, etc.)
if img.mode not in ('RGB', 'RGBA'):
img = img.convert('RGB')
if img.mode not in ("RGB", "RGBA"):
img = img.convert("RGB")
# Create fresh image - this discards all metadata
clean = Image.new(img.mode, img.size)
@@ -56,11 +56,7 @@ def strip_image_metadata(image_data: bytes, output_format: str = 'PNG') -> bytes
return output.getvalue()
def generate_filename(
date_str: str | None = None,
prefix: str = "",
extension: str = "png"
) -> str:
def generate_filename(date_str: str | None = None, prefix: str = "", extension: str = "png") -> str:
"""
Generate a filename for stego images.
@@ -78,17 +74,19 @@ def generate_filename(
>>> generate_filename("2023-12-25", "secret_", "png")
"secret_a1b2c3d4_20231225.png"
"""
debug.validate(bool(extension) and '.' not in extension,
f"Extension must not contain dot, got '{extension}'")
debug.validate(
bool(extension) and "." not in extension,
f"Extension must not contain dot, got '{extension}'",
)
if date_str is None:
date_str = date.today().isoformat()
date_compact = date_str.replace('-', '')
date_compact = date_str.replace("-", "")
random_hex = secrets.token_hex(4)
# Ensure extension doesn't have a leading dot
extension = extension.lstrip('.')
extension = extension.lstrip(".")
filename = f"{prefix}{random_hex}_{date_compact}.{extension}"
debug.print(f"Generated filename: {filename}")
@@ -114,7 +112,7 @@ def parse_date_from_filename(filename: str) -> str | None:
import re
# Try YYYYMMDD format
match = re.search(r'_(\d{4})(\d{2})(\d{2})(?:\.|$)', filename)
match = re.search(r"_(\d{4})(\d{2})(\d{2})(?:\.|$)", filename)
if match:
year, month, day = match.groups()
date_str = f"{year}-{month}-{day}"
@@ -122,7 +120,7 @@ def parse_date_from_filename(filename: str) -> str | None:
return date_str
# Try YYYY-MM-DD format
match = re.search(r'_(\d{4})-(\d{2})-(\d{2})(?:\.|$)', filename)
match = re.search(r"_(\d{4})-(\d{2})-(\d{2})(?:\.|$)", filename)
if match:
year, month, day = match.groups()
date_str = f"{year}-{month}-{day}"
@@ -147,11 +145,13 @@ def get_day_from_date(date_str: str) -> str:
>>> get_day_from_date("2023-12-25")
"Monday"
"""
debug.validate(len(date_str) == 10 and date_str[4] == '-' and date_str[7] == '-',
f"Invalid date format: {date_str}, expected YYYY-MM-DD")
debug.validate(
len(date_str) == 10 and date_str[4] == "-" and date_str[7] == "-",
f"Invalid date format: {date_str}, expected YYYY-MM-DD",
)
try:
year, month, day = map(int, date_str.split('-'))
year, month, day = map(int, date_str.split("-"))
d = date(year, month, day)
day_name = DAY_NAMES[d.weekday()]
debug.print(f"Date {date_str} is {day_name}")
@@ -231,11 +231,11 @@ class SecureDeleter:
debug.print("File is empty, nothing to overwrite")
return
patterns = [b'\x00', b'\xFF', bytes([random.randint(0, 255)])]
patterns = [b"\x00", b"\xff", bytes([random.randint(0, 255)])]
for pass_num in range(self.passes):
debug.print(f"Overwrite pass {pass_num + 1}/{self.passes}")
with open(file_path, 'r+b') as f:
with open(file_path, "r+b") as f:
for pattern_idx, pattern in enumerate(patterns):
f.seek(0)
# Write pattern in chunks for large files
@@ -243,7 +243,7 @@ class SecureDeleter:
for offset in range(0, length, chunk_size):
chunk = min(chunk_size, length - offset)
f.write(pattern * (chunk // len(pattern)))
f.write(pattern[:chunk % len(pattern)])
f.write(pattern[: chunk % len(pattern)])
# Final pass with random data
f.seek(0)
@@ -271,7 +271,7 @@ class SecureDeleter:
# First, securely overwrite all files
file_count = 0
for file_path in self.path.rglob('*'):
for file_path in self.path.rglob("*"):
if file_path.is_file():
self._overwrite_file(file_path)
file_count += 1
@@ -325,9 +325,9 @@ def format_file_size(size_bytes: int) -> str:
debug.validate(size_bytes >= 0, f"File size cannot be negative: {size_bytes}")
size: float = float(size_bytes)
for unit in ['B', 'KB', 'MB', 'GB']:
for unit in ["B", "KB", "MB", "GB"]:
if size < 1024:
if unit == 'B':
if unit == "B":
return f"{int(size)} {unit}"
return f"{size:.1f} {unit}"
size /= 1024

59
src/stegasoo/validation.py
View File

@@ -66,11 +66,9 @@ def validate_pin(pin: str, required: bool = False) -> ValidationResult:
return ValidationResult.error("PIN must contain only digits")
if len(pin) < MIN_PIN_LENGTH or len(pin) > MAX_PIN_LENGTH:
return ValidationResult.error(
f"PIN must be {MIN_PIN_LENGTH}-{MAX_PIN_LENGTH} digits"
)
return ValidationResult.error(f"PIN must be {MIN_PIN_LENGTH}-{MAX_PIN_LENGTH} digits")
if pin[0] == '0':
if pin[0] == "0":
return ValidationResult.error("PIN cannot start with zero")
return ValidationResult.ok(length=len(pin))
@@ -121,9 +119,7 @@ def validate_payload(payload: str | bytes | FilePayload) -> ValidationResult:
)
return ValidationResult.ok(
size=len(payload.data),
filename=payload.filename,
mime_type=payload.mime_type
size=len(payload.data), filename=payload.filename, mime_type=payload.mime_type
)
elif isinstance(payload, bytes):
@@ -143,9 +139,7 @@ def validate_payload(payload: str | bytes | FilePayload) -> ValidationResult:
def validate_file_payload(
file_data: bytes,
filename: str = "",
max_size: int = MAX_FILE_PAYLOAD_SIZE
file_data: bytes, filename: str = "", max_size: int = MAX_FILE_PAYLOAD_SIZE
) -> ValidationResult:
"""
Validate a file for embedding.
@@ -173,9 +167,7 @@ def validate_file_payload(
def validate_image(
image_data: bytes,
name: str = "Image",
check_size: bool = True
image_data: bytes, name: str = "Image", check_size: bool = True
) -> ValidationResult:
"""
Validate image data and dimensions.
@@ -202,18 +194,14 @@ def validate_image(
num_pixels = width * height
if check_size and num_pixels > MAX_IMAGE_PIXELS:
max_dim = int(MAX_IMAGE_PIXELS ** 0.5)
max_dim = int(MAX_IMAGE_PIXELS**0.5)
return ValidationResult.error(
f"{name} too large ({width}×{height} = {num_pixels:,} pixels). "
f"Maximum: ~{MAX_IMAGE_PIXELS:,} pixels ({max_dim}×{max_dim})"
)
return ValidationResult.ok(
width=width,
height=height,
pixels=num_pixels,
mode=img.mode,
format=img.format
width=width, height=height, pixels=num_pixels, mode=img.mode, format=img.format
)
except Exception as e:
@@ -221,9 +209,7 @@ def validate_image(
def validate_rsa_key(
key_data: bytes,
password: str | None = None,
required: bool = False
key_data: bytes, password: str | None = None, required: bool = False
) -> ValidationResult:
"""
Validate RSA private key.
@@ -256,10 +242,7 @@ def validate_rsa_key(
return ValidationResult.error(str(e))
def validate_security_factors(
pin: str,
rsa_key_data: bytes | None
) -> ValidationResult:
def validate_security_factors(pin: str, rsa_key_data: bytes | None) -> ValidationResult:
"""
Validate that at least one security factor is provided.
@@ -274,17 +257,13 @@ def validate_security_factors(
has_key = bool(rsa_key_data and len(rsa_key_data) > 0)
if not has_pin and not has_key:
return ValidationResult.error(
"You must provide at least a PIN or RSA Key"
)
return ValidationResult.error("You must provide at least a PIN or RSA Key")
return ValidationResult.ok(has_pin=has_pin, has_key=has_key)
def validate_file_extension(
filename: str,
allowed: set[str],
file_type: str = "File"
filename: str, allowed: set[str], file_type: str = "File"
) -> ValidationResult:
"""
Validate file extension.
@@ -297,10 +276,10 @@ def validate_file_extension(
Returns:
ValidationResult with extension
"""
if not filename or '.' not in filename:
if not filename or "." not in filename:
return ValidationResult.error(f"{file_type} must have a file extension")
ext = filename.rsplit('.', 1)[1].lower()
ext = filename.rsplit(".", 1)[1].lower()
if ext not in allowed:
return ValidationResult.error(
@@ -368,7 +347,7 @@ def validate_passphrase(passphrase: str) -> ValidationResult:
if len(words) < RECOMMENDED_PASSPHRASE_WORDS:
return ValidationResult.ok(
word_count=len(words),
warning=f"Recommend {RECOMMENDED_PASSPHRASE_WORDS}+ words for better security"
warning=f"Recommend {RECOMMENDED_PASSPHRASE_WORDS}+ words for better security",
)
return ValidationResult.ok(word_count=len(words))
@@ -378,6 +357,7 @@ def validate_passphrase(passphrase: str) -> ValidationResult:
# NEW VALIDATORS FOR V3.2.0
# =============================================================================
def validate_reference_photo(photo_data: bytes) -> ValidationResult:
"""Validate reference photo. Alias for validate_image."""
return validate_image(photo_data, "Reference photo")
@@ -418,7 +398,7 @@ def validate_dct_output_format(format_str: str) -> ValidationResult:
Returns:
ValidationResult
"""
valid_formats = {'png', 'jpeg'}
valid_formats = {"png", "jpeg"}
if format_str.lower() not in valid_formats:
return ValidationResult.error(
@@ -438,7 +418,7 @@ def validate_dct_color_mode(mode: str) -> ValidationResult:
Returns:
ValidationResult
"""
valid_modes = {'grayscale', 'color'}
valid_modes = {"grayscale", "color"}
if mode.lower() not in valid_modes:
return ValidationResult.error(
@@ -452,6 +432,7 @@ def validate_dct_color_mode(mode: str) -> ValidationResult:
# EXCEPTION-RAISING VALIDATORS (for CLI/API use)
# ============================================================================
def require_valid_pin(pin: str, required: bool = False) -> None:
"""Validate PIN, raising exception on failure."""
result = validate_pin(pin, required)
@@ -481,9 +462,7 @@ def require_valid_image(image_data: bytes, name: str = "Image") -> None:
def require_valid_rsa_key(
key_data: bytes,
password: str | None = None,
required: bool = False
key_data: bytes, password: str | None = None, required: bool = False
) -> None:
"""Validate RSA key, raising exception on failure."""
result = validate_rsa_key(key_data, password, required)