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

Add per-channel hybrid audio spread spectrum and env feature toggles

Browse Source 
Spread spectrum v2: independent per-channel embedding with round-robin
bit distribution, preserving spatial stereo/surround mix. Adaptive chip
tiers (256/512/1024) trade capacity for lossy codec robustness. LFE
channel skipped for 5.1+ layouts. v2 header (20B) with backward-
compatible v0 decode fallback.

Environment toggles (STEGASOO_AUDIO, STEGASOO_VIDEO) gate audio/video
features for minimal builds (e.g. Raspberry Pi image-only). Values:
auto (default, detect deps), 1/true (force on), 0/false (force off).

Web UI fixes: accordion defaults to step 1 on load, chevron arrow
styling, required attribute toggling for audio carrier type switch,
"Images & Mode" renamed to "Reference, Carrier, Mode".

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
adlee-was-taken
2026-02-28 11:58:40 -05:00
parent 0248bec813
commit ef5a9ce9cb
41 changed files with 4281 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files

474
tests/test_audio.py
View File

@@ -3,25 +3,37 @@ Tests for Stegasoo audio steganography.
Tests cover:
- Audio LSB roundtrip (encode + decode)
- Audio MDCT roundtrip (encode + decode)
- Audio spread spectrum roundtrip (v0 legacy + v2 per-channel)
- Wrong credentials fail to decode
- Capacity calculations
- Capacity calculations (per-tier)
- Format detection
- Audio validation
- Per-channel stereo/multichannel embedding (v4.4.0)
- Chip tier roundtrips (v4.4.0)
- LFE channel skipping (v4.4.0)
- Backward compat: v0 decode from v2 code
- Header v2 build/parse roundtrip
- Round-robin bit distribution
"""
import io
from pathlib import Path
import numpy as np
import pytest
import soundfile as sf
from stegasoo.constants import (
EMBED_MODE_AUDIO_LSB,
EMBED_MODE_AUDIO_SPREAD,
)
from stegasoo.constants import AUDIO_ENABLED, EMBED_MODE_AUDIO_LSB, EMBED_MODE_AUDIO_SPREAD
from stegasoo.models import AudioCapacityInfo, AudioEmbedStats, AudioInfo
pytestmark = pytest.mark.skipif(not AUDIO_ENABLED, reason="Audio support disabled (STEGASOO_AUDIO)")
# Path to real test data files
_TEST_DATA = Path(__file__).parent.parent / "test_data"
_REFERENCE_PNG = _TEST_DATA / "reference.png"
_SPEECH_WAV = _TEST_DATA / "stupid_elitist_speech.wav"
# =============================================================================
# FIXTURES
# =============================================================================
@@ -33,7 +45,6 @@ def carrier_wav() -> bytes:
sample_rate = 44100
duration = 1.0
num_samples = int(sample_rate * duration)
# Generate a simple sine wave
t = np.linspace(0, duration, num_samples, endpoint=False)
samples = (np.sin(2 * np.pi * 440 * t) * 16000).astype(np.int16)
@@ -45,9 +56,9 @@ def carrier_wav() -> bytes:
@pytest.fixture
def carrier_wav_stereo() -> bytes:
"""Generate a stereo test WAV file."""
"""Generate a stereo test WAV file (5 seconds for spread spectrum capacity)."""
sample_rate = 44100
duration = 1.0
duration = 5.0
num_samples = int(sample_rate * duration)
t = np.linspace(0, duration, num_samples, endpoint=False)
left = (np.sin(2 * np.pi * 440 * t) * 16000).astype(np.int16)
@@ -67,7 +78,6 @@ def carrier_wav_long() -> bytes:
duration = 15.0
num_samples = int(sample_rate * duration)
t = np.linspace(0, duration, num_samples, endpoint=False)
# Mix of frequencies for better MDCT embedding
samples = (
(np.sin(2 * np.pi * 440 * t) + np.sin(2 * np.pi * 880 * t) + np.sin(2 * np.pi * 1320 * t))
* 5000
@@ -80,12 +90,47 @@ def carrier_wav_long() -> bytes:
@pytest.fixture
def carrier_wav_spread_integration() -> bytes:
"""Generate a very long WAV (150 seconds) for spread spectrum integration tests.
def carrier_wav_stereo_long() -> bytes:
"""Generate a stereo WAV (15 seconds) for per-channel spread tests."""
sample_rate = 48000
duration = 15.0
num_samples = int(sample_rate * duration)
t = np.linspace(0, duration, num_samples, endpoint=False)
left = (np.sin(2 * np.pi * 440 * t) * 10000).astype(np.float64) / 32768.0
right = (np.sin(2 * np.pi * 660 * t) * 10000).astype(np.float64) / 32768.0
samples = np.column_stack([left, right])
Spread spectrum needs 1024 samples per bit. With encryption + RS overhead (~690 bytes),
we need at least 690*8*1024 = 5.7M samples ~ 130 seconds at 44.1kHz.
"""
buf = io.BytesIO()
sf.write(buf, samples, sample_rate, format="WAV", subtype="PCM_16")
buf.seek(0)
return buf.read()
@pytest.fixture
def carrier_wav_5_1() -> bytes:
"""Generate a 6-channel (5.1) WAV for LFE skip tests."""
sample_rate = 48000
duration = 15.0
num_samples = int(sample_rate * duration)
t = np.linspace(0, duration, num_samples, endpoint=False)
# 6 channels with different frequencies
freqs = [440, 554, 660, 80, 880, 1100] # ch3 = LFE (low freq)
channels = []
for freq in freqs:
ch = (np.sin(2 * np.pi * freq * t) * 8000).astype(np.float64) / 32768.0
channels.append(ch)
samples = np.column_stack(channels)
buf = io.BytesIO()
sf.write(buf, samples, sample_rate, format="WAV", subtype="PCM_16")
buf.seek(0)
return buf.read()
@pytest.fixture
def carrier_wav_spread_integration() -> bytes:
"""Generate a very long WAV (150 seconds) for spread spectrum integration tests."""
sample_rate = 44100
duration = 150.0
num_samples = int(sample_rate * duration)
@@ -103,7 +148,9 @@ def carrier_wav_spread_integration() -> bytes:
@pytest.fixture
def reference_photo() -> bytes:
"""Generate a small reference photo (PNG)."""
"""Load real reference photo from test_data, or generate a small one."""
if _REFERENCE_PNG.exists():
return _REFERENCE_PNG.read_bytes()
from PIL import Image
img = Image.new("RGB", (100, 100), color=(128, 64, 32))
@@ -113,6 +160,14 @@ def reference_photo() -> bytes:
return buf.read()
@pytest.fixture
def speech_wav() -> bytes:
"""Load real speech WAV from test_data (48kHz mono, ~68s)."""
if not _SPEECH_WAV.exists():
pytest.skip("test_data/stupid_elitist_speech.wav not found")
return _SPEECH_WAV.read_bytes()
# =============================================================================
# AUDIO LSB TESTS
# =============================================================================
@@ -134,7 +189,6 @@ class TestAudioLSB:
from stegasoo.audio_steganography import embed_in_audio_lsb, extract_from_audio_lsb
payload = b"Hello, audio steganography!"
# Prepend with magic header to simulate real usage pattern
key = b"\x42" * 32
stego_audio, stats = embed_in_audio_lsb(payload, carrier_wav, key)
@@ -145,7 +199,6 @@ class TestAudioLSB:
assert stats.samples_modified > 0
assert 0 < stats.capacity_used <= 1.0
# Extract
extracted = extract_from_audio_lsb(stego_audio, key)
assert extracted is not None
assert extracted == payload
@@ -174,7 +227,6 @@ class TestAudioLSB:
stego_audio, _ = embed_in_audio_lsb(payload, carrier_wav, correct_key)
extracted = extract_from_audio_lsb(stego_audio, wrong_key)
# Should return None or garbage (not the original message)
assert extracted is None or extracted != payload
def test_two_bits_per_sample(self, carrier_wav):
@@ -197,46 +249,97 @@ class TestAudioLSB:
indices1 = generate_sample_indices(key, 10000, 100)
indices2 = generate_sample_indices(key, 10000, 100)
# Same key should produce same indices
assert indices1 == indices2
# All indices should be valid
assert all(0 <= i < 10000 for i in indices1)
# No duplicates
assert len(set(indices1)) == len(indices1)
# =============================================================================
# AUDIO SPREAD SPECTRUM TESTS
# AUDIO SPREAD SPECTRUM TESTS (v2 per-channel)
# =============================================================================
class TestAudioSpread:
"""Tests for audio spread spectrum steganography."""
"""Tests for audio spread spectrum steganography (v2 per-channel)."""
def test_calculate_capacity(self, carrier_wav_long):
def test_calculate_capacity_default_tier(self, carrier_wav_long):
from stegasoo.spread_steganography import calculate_audio_spread_capacity
capacity = calculate_audio_spread_capacity(carrier_wav_long)
assert isinstance(capacity, AudioCapacityInfo)
assert capacity.usable_capacity_bytes > 0
assert capacity.embed_mode == EMBED_MODE_AUDIO_SPREAD
assert capacity.chip_tier == 2 # default
assert capacity.chip_length == 1024
def test_spread_roundtrip(self, carrier_wav_long):
"""Test spread spectrum embed/extract roundtrip."""
def test_calculate_capacity_per_tier(self, carrier_wav_long):
"""Capacity should increase as chip length decreases."""
from stegasoo.spread_steganography import calculate_audio_spread_capacity
cap_lossless = calculate_audio_spread_capacity(carrier_wav_long, chip_tier=0)
cap_high = calculate_audio_spread_capacity(carrier_wav_long, chip_tier=1)
cap_low = calculate_audio_spread_capacity(carrier_wav_long, chip_tier=2)
assert cap_lossless.chip_length == 256
assert cap_high.chip_length == 512
assert cap_low.chip_length == 1024
# Smaller chip = more capacity
assert cap_lossless.usable_capacity_bytes > cap_high.usable_capacity_bytes
assert cap_high.usable_capacity_bytes > cap_low.usable_capacity_bytes
def test_spread_roundtrip_default_tier(self, carrier_wav_long):
"""Test spread spectrum embed/extract roundtrip (default tier 2)."""
from stegasoo.spread_steganography import (
embed_in_audio_spread,
extract_from_audio_spread,
)
payload = b"Spread test"
payload = b"Spread test v2"
seed = b"\x42" * 32
stego_audio, stats = embed_in_audio_spread(payload, carrier_wav_long, seed)
assert isinstance(stats, AudioEmbedStats)
assert stats.embed_mode == EMBED_MODE_AUDIO_SPREAD
assert stats.chip_tier == 2
assert stats.chip_length == 1024
extracted = extract_from_audio_spread(stego_audio, seed)
assert extracted is not None
assert extracted == payload
def test_spread_roundtrip_tier_0(self, carrier_wav_long):
"""Test spread spectrum at tier 0 (chip=256, lossless)."""
from stegasoo.spread_steganography import (
embed_in_audio_spread,
extract_from_audio_spread,
)
payload = b"Lossless tier test with more data to embed for coverage"
seed = b"\x42" * 32
stego_audio, stats = embed_in_audio_spread(payload, carrier_wav_long, seed, chip_tier=0)
assert stats.chip_tier == 0
assert stats.chip_length == 256
extracted = extract_from_audio_spread(stego_audio, seed)
assert extracted is not None
assert extracted == payload
def test_spread_roundtrip_tier_1(self, carrier_wav_long):
"""Test spread spectrum at tier 1 (chip=512, high lossy)."""
from stegasoo.spread_steganography import (
embed_in_audio_spread,
extract_from_audio_spread,
)
payload = b"High lossy tier test"
seed = b"\x42" * 32
stego_audio, stats = embed_in_audio_spread(payload, carrier_wav_long, seed, chip_tier=1)
assert stats.chip_tier == 1
assert stats.chip_length == 512
extracted = extract_from_audio_spread(stego_audio, seed)
assert extracted is not None
@@ -258,6 +361,258 @@ class TestAudioSpread:
extracted = extract_from_audio_spread(stego_audio, wrong_seed)
assert extracted is None or extracted != payload
def test_per_channel_stereo_roundtrip(self, carrier_wav_stereo_long):
"""Test that stereo per-channel embedding/extraction works."""
from stegasoo.spread_steganography import (
embed_in_audio_spread,
extract_from_audio_spread,
)
payload = b"Stereo per-channel test"
seed = b"\xAB" * 32
stego_audio, stats = embed_in_audio_spread(
payload, carrier_wav_stereo_long, seed, chip_tier=0
)
assert stats.channels == 2
assert stats.embeddable_channels == 2
extracted = extract_from_audio_spread(stego_audio, seed)
assert extracted is not None
assert extracted == payload
def test_per_channel_preserves_spatial_mix(self, carrier_wav_stereo_long):
"""Verify that per-channel embedding doesn't destroy the spatial mix.
The difference between left and right channels should be preserved
(not zeroed out as the old mono-broadcast approach would do).
"""
from stegasoo.spread_steganography import embed_in_audio_spread
payload = b"Spatial preservation test"
seed = b"\xCD" * 32
# Read original
orig_samples, _ = sf.read(io.BytesIO(carrier_wav_stereo_long), dtype="float64", always_2d=True)
orig_diff = orig_samples[:, 0] - orig_samples[:, 1]
# Embed
stego_bytes, _ = embed_in_audio_spread(
payload, carrier_wav_stereo_long, seed, chip_tier=0
)
# Read stego
stego_samples, _ = sf.read(io.BytesIO(stego_bytes), dtype="float64", always_2d=True)
stego_diff = stego_samples[:, 0] - stego_samples[:, 1]
# The channel difference should not be identical (embedding adds different
# noise per channel), but should be very close (embedding is subtle)
# With the old mono-broadcast approach, stego_diff would equal orig_diff
# exactly in unmodified regions but differ where data was embedded.
# With per-channel, both channels get independent modifications.
correlation = np.corrcoef(orig_diff, stego_diff)[0, 1]
assert correlation > 0.95, f"Spatial mix correlation too low: {correlation}"
def test_capacity_scales_with_channels(self, carrier_wav_long, carrier_wav_stereo_long):
"""Stereo should have roughly double the capacity of mono."""
from stegasoo.spread_steganography import calculate_audio_spread_capacity
mono_cap = calculate_audio_spread_capacity(carrier_wav_long, chip_tier=0)
stereo_cap = calculate_audio_spread_capacity(carrier_wav_stereo_long, chip_tier=0)
# Stereo should be ~1.5-2.2x mono (not exact because header is ch0 only
# and the files have slightly different durations/sample rates)
ratio = stereo_cap.usable_capacity_bytes / mono_cap.usable_capacity_bytes
assert ratio > 1.3, f"Stereo/mono capacity ratio too low: {ratio}"
def test_lfe_skip_5_1(self, carrier_wav_5_1):
"""LFE channel (index 3) should be unmodified in 6-channel audio."""
from stegasoo.spread_steganography import embed_in_audio_spread
payload = b"LFE skip test"
seed = b"\xEE" * 32
# Read original LFE channel
orig_samples, _ = sf.read(io.BytesIO(carrier_wav_5_1), dtype="float64", always_2d=True)
orig_lfe = orig_samples[:, 3].copy()
stego_bytes, stats = embed_in_audio_spread(
payload, carrier_wav_5_1, seed, chip_tier=0
)
assert stats.embeddable_channels == 5 # 6 channels - 1 LFE = 5
stego_samples, _ = sf.read(io.BytesIO(stego_bytes), dtype="float64", always_2d=True)
stego_lfe = stego_samples[:, 3]
# LFE channel should be completely unmodified
np.testing.assert_array_equal(orig_lfe, stego_lfe)
def test_lfe_skip_roundtrip(self, carrier_wav_5_1):
"""5.1 audio embed/extract roundtrip with LFE skipping."""
from stegasoo.spread_steganography import (
embed_in_audio_spread,
extract_from_audio_spread,
)
payload = b"5.1 surround test"
seed = b"\xEE" * 32
stego_bytes, stats = embed_in_audio_spread(
payload, carrier_wav_5_1, seed, chip_tier=0
)
assert stats.channels == 6
assert stats.embeddable_channels == 5
extracted = extract_from_audio_spread(stego_bytes, seed)
assert extracted is not None
assert extracted == payload
# =============================================================================
# HEADER V2 TESTS
# =============================================================================
class TestHeaderV2:
"""Tests for v2 header construction and parsing."""
def test_header_v2_build_parse_roundtrip(self):
from stegasoo.spread_steganography import _build_header_v2, _parse_header
data_length = 12345
chip_tier = 1
num_ch = 2
lfe_skipped = False
header = _build_header_v2(data_length, chip_tier, num_ch, lfe_skipped)
assert len(header) == 20
magic_valid, version, length, tier, nch, lfe = _parse_header(header)
assert magic_valid
assert version == 2
assert length == data_length
assert tier == chip_tier
assert nch == num_ch
assert lfe is False
def test_header_v2_with_lfe_flag(self):
from stegasoo.spread_steganography import _build_header_v2, _parse_header
header = _build_header_v2(999, 0, 5, lfe_skipped=True)
magic_valid, version, length, tier, nch, lfe = _parse_header(header)
assert magic_valid
assert version == 2
assert length == 999
assert tier == 0
assert nch == 5
assert lfe is True
def test_header_v0_build_parse(self):
from stegasoo.spread_steganography import _build_header_v0, _parse_header
header = _build_header_v0(4567)
assert len(header) == 16
magic_valid, version, length, tier, nch, lfe = _parse_header(header)
assert magic_valid
assert version == 0
assert length == 4567
assert tier is None
assert nch is None
def test_header_bad_magic(self):
from stegasoo.spread_steganography import _parse_header
bad_header = b"XXXX" + b"\x00" * 16
magic_valid, version, length, tier, nch, lfe = _parse_header(bad_header)
assert not magic_valid
# =============================================================================
# ROUND-ROBIN BIT DISTRIBUTION TESTS
# =============================================================================
class TestRoundRobin:
"""Tests for round-robin bit distribution."""
def test_distribute_and_collect_identity(self):
from stegasoo.spread_steganography import (
_collect_bits_round_robin,
_distribute_bits_round_robin,
)
bits = [1, 0, 1, 1, 0, 0, 1, 0, 1, 1]
for num_ch in [1, 2, 3, 4, 5]:
per_ch = _distribute_bits_round_robin(bits, num_ch)
assert len(per_ch) == num_ch
reassembled = _collect_bits_round_robin(per_ch)
assert reassembled == bits, f"Failed for {num_ch} channels"
def test_distribute_round_robin_ordering(self):
from stegasoo.spread_steganography import _distribute_bits_round_robin
bits = [0, 1, 2, 3, 4, 5] # using ints for clarity
per_ch = _distribute_bits_round_robin(bits, 3)
# ch0: bits 0, 3 ch1: bits 1, 4 ch2: bits 2, 5
assert per_ch[0] == [0, 3]
assert per_ch[1] == [1, 4]
assert per_ch[2] == [2, 5]
def test_distribute_uneven(self):
from stegasoo.spread_steganography import (
_collect_bits_round_robin,
_distribute_bits_round_robin,
)
bits = [0, 1, 2, 3, 4] # 5 bits across 3 channels
per_ch = _distribute_bits_round_robin(bits, 3)
assert per_ch[0] == [0, 3]
assert per_ch[1] == [1, 4]
assert per_ch[2] == [2]
reassembled = _collect_bits_round_robin(per_ch)
assert reassembled == bits
# =============================================================================
# CHANNEL MANAGEMENT TESTS
# =============================================================================
class TestChannelManagement:
"""Tests for embeddable channel selection."""
def test_mono(self):
from stegasoo.spread_steganography import _embeddable_channels
assert _embeddable_channels(1) == [0]
def test_stereo(self):
from stegasoo.spread_steganography import _embeddable_channels
assert _embeddable_channels(2) == [0, 1]
def test_5_1_skips_lfe(self):
from stegasoo.spread_steganography import _embeddable_channels
channels = _embeddable_channels(6)
assert channels == [0, 1, 2, 4, 5]
assert 3 not in channels # LFE skipped
def test_7_1_skips_lfe(self):
from stegasoo.spread_steganography import _embeddable_channels
channels = _embeddable_channels(8)
assert 3 not in channels
assert len(channels) == 7
def test_quad_no_skip(self):
from stegasoo.spread_steganography import _embeddable_channels
# 4 channels < 6, so no LFE skip
assert _embeddable_channels(4) == [0, 1, 2, 3]
# =============================================================================
# FORMAT DETECTION TESTS
@@ -423,6 +778,36 @@ class TestIntegration:
assert result.message == "Spread integration test"
def test_spread_encode_decode_with_chip_tier(
self, carrier_wav_spread_integration, reference_photo
):
"""Test spread spectrum with explicit chip tier."""
from stegasoo.decode import decode_audio
from stegasoo.encode import encode_audio
stego_audio, stats = encode_audio(
message="Tier 0 integration",
reference_photo=reference_photo,
carrier_audio=carrier_wav_spread_integration,
passphrase="test words here now",
pin="123456",
embed_mode="audio_spread",
chip_tier=0,
)
assert stats.chip_tier == 0
assert stats.chip_length == 256
result = decode_audio(
stego_audio=stego_audio,
reference_photo=reference_photo,
passphrase="test words here now",
pin="123456",
embed_mode="audio_spread",
)
assert result.message == "Tier 0 integration"
def test_auto_detect_lsb(self, carrier_wav, reference_photo):
"""Test auto-detection finds LSB encoded audio."""
from stegasoo.decode import decode_audio
@@ -446,3 +831,32 @@ class TestIntegration:
)
assert result.message == "Auto-detect test"
def test_spread_with_real_speech(self, speech_wav, reference_photo):
"""Test spread spectrum with real speech audio from test_data."""
from stegasoo.decode import decode_audio
from stegasoo.encode import encode_audio
message = "Hidden in a speech about elitism"
stego_audio, stats = encode_audio(
message=message,
reference_photo=reference_photo,
carrier_audio=speech_wav,
passphrase="test words here now",
pin="123456",
embed_mode="audio_spread",
chip_tier=0, # lossless tier for max capacity
)
assert stats.chip_tier == 0
result = decode_audio(
stego_audio=stego_audio,
reference_photo=reference_photo,
passphrase="test words here now",
pin="123456",
embed_mode="audio_spread",
)
assert result.message == message