+ Where length_byte is the length of the filename (1-100 bytes).
+ """
+ if len(data) < 3:
+ return (None, data)
+
+ fname_len = data[0]
+
+ # Filename length must be reasonable (3-100 chars for "a.b" to reasonable max)
+ if fname_len < 3 or fname_len > 100:
+ return (None, data)
+
+ # Must have enough data for filename + at least 1 byte of content
+ if len(data) < fname_len + 2:
+ return (None, data)
+
+ try:
+ filename = data[1:1+fname_len].decode('utf-8')
+ except UnicodeDecodeError:
+ return (None, data)
+
+ # Validate filename structure
+ if '.' not in filename:
+ return (None, data)
+
+ # Check for invalid characters (only allow alphanumeric, ., -, _, space)
+ if not re.match(r'^[\w\-. ]+$', filename):
+ return (None, data)
+
+ # Filename must not start with . or space
+ if filename[0] in '. ':
+ return (None, data)
+
+ # Extract and validate extension
+ ext = filename.rsplit('.', 1)[-1].lower()
+ if ext not in VALID_FILE_EXTENSIONS:
+ return (None, data)
+
+ # All checks passed - this looks like a real file
+ file_data = data[1+fname_len:]
+ return (filename, file_data)
+
+
+def matryoshka_decode(image: Image.Image, max_depth: int = 3, password: str = None,
+ current_depth: int = 0) -> list:
+ """
+ 🪆 Recursively decode nested steganographic images.
+
+ Args:
+ image: The image to decode
+ max_depth: Maximum recursion depth (1-11)
+ password: Optional decryption password
+ current_depth: Current recursion level (internal)
+
+ Returns:
+ List of extraction results at each layer
+ """
+ results = []
+ layer_info = {
+ "depth": current_depth,
+ "type": "unknown",
+ "filename": None,
+ "data_size": 0,
+ "preview": "",
+ "has_nested": False,
+ "nested_results": [],
+ }
+
+ if current_depth >= max_depth:
+ layer_info["type"] = "max_depth_reached"
+ layer_info["preview"] = f"⚠️ Max depth ({max_depth}) reached"
+ results.append(layer_info)
+ return results
+
+ try:
+ # Try auto-decode first (looks for STEG header)
+ try:
+ data = decode(image, None)
+ layer_info["type"] = "steg_header"
+ except:
+ # If no STEG header, try smart scan
+ scan_results = smart_scan_image(image, password)
+ best_result = None
+ for r in scan_results:
+ if r.get("status") in ["STEG_DETECTED", "STEG_HEADER", "TEXT_FOUND"]:
+ best_result = r
+ break
+
+ if best_result and best_result.get("raw_data"):
+ data = best_result["raw_data"]
+ layer_info["type"] = f"smart_scan_{best_result['name']}"
+ else:
+ layer_info["type"] = "no_data_found"
+ layer_info["preview"] = "No hidden data detected"
+ results.append(layer_info)
+ return results
+
+ # Decrypt if password provided
+ if password:
+ try:
+ data = crypto.decrypt(data, password)
+ except:
+ pass # Decryption failed, use raw data
+
+ layer_info["data_size"] = len(data)
+
+ # Check if it's a file
+ filename, file_data = extract_file_from_data(data)
+
+ if filename:
+ layer_info["filename"] = filename
+ layer_info["data_size"] = len(file_data)
+
+ # Check if the extracted file is an image
+ if is_image_data(file_data):
+ layer_info["type"] = "nested_image"
+ layer_info["has_nested"] = True
+
+ # Recursively decode the nested image
+ try:
+ nested_img = Image.open(io.BytesIO(file_data))
+ nested_results = matryoshka_decode(
+ nested_img,
+ max_depth=max_depth,
+ password=password,
+ current_depth=current_depth + 1
+ )
+ layer_info["nested_results"] = nested_results
+ layer_info["preview"] = f"🪆 Found nested image: {filename}"
+ except Exception as e:
+ layer_info["preview"] = f"📁 Image file: {filename} (failed to recurse: {e})"
+ else:
+ layer_info["type"] = "file"
+ # Try to show preview of text files
+ ext = filename.split('.')[-1].lower() if '.' in filename else ''
+ if ext in ['txt', 'md', 'json', 'xml', 'html', 'css', 'js', 'py', 'csv']:
+ try:
+ layer_info["preview"] = file_data[:200].decode('utf-8')
+ except:
+ layer_info["preview"] = f"📁 Binary file: {filename}"
+ else:
+ layer_info["preview"] = f"📁 File: {filename} ({format_size(len(file_data))})"
+ else:
+ # Raw data - check if it's an image
+ if is_image_data(data):
+ layer_info["type"] = "nested_image_raw"
+ layer_info["has_nested"] = True
+
+ try:
+ nested_img = Image.open(io.BytesIO(data))
+ nested_results = matryoshka_decode(
+ nested_img,
+ max_depth=max_depth,
+ password=password,
+ current_depth=current_depth + 1
+ )
+ layer_info["nested_results"] = nested_results
+ layer_info["preview"] = "🪆 Found raw nested image data"
+ except Exception as e:
+ layer_info["preview"] = f"Image data (failed to recurse: {e})"
+ else:
+ # Try as text
+ try:
+ text = data.decode('utf-8')
+ layer_info["type"] = "text"
+ layer_info["preview"] = text[:300]
+ except:
+ layer_info["type"] = "binary"
+ layer_info["preview"] = f"Binary data: {data[:50].hex()}..."
+
+ # Store raw data for download
+ layer_info["raw_data"] = file_data if filename else data
+
+ except Exception as e:
+ layer_info["type"] = "error"
+ layer_info["preview"] = f"Error: {str(e)}"
+
+ results.append(layer_info)
+ return results
+
+
+def matryoshka_encode(payload: bytes, carriers: list, config = None,
+ password: str = None) -> tuple:
+ """
+ 🪆 Recursively encode nested steganographic images.
+
+ Creates a "Russian nesting doll" of hidden data, encoding the payload
+ into the innermost carrier, then that result into the next carrier, etc.
+
+ Args:
+ payload: The data to hide (innermost secret)
+ carriers: List of (Image, filename) tuples - innermost carrier FIRST
+ config: StegConfig to use for all layers (or None for auto)
+ password: Optional encryption password
+
+ Returns:
+ Tuple of (final_image, layer_info_list)
+ """
+ if not carriers:
+ raise ValueError("At least one carrier image is required")
+
+ if config is None:
+ config = create_config(channels='RGBA', bits=2) # Default: good capacity
+
+ layer_info = []
+ current_data = payload
+
+ # Encode from innermost to outermost
+ for i, (carrier_img, carrier_name) in enumerate(carriers):
+ layer_num = i + 1
+
+ # Calculate capacity
+ capacity = _calculate_capacity_bytes(carrier_img, config)
+ data_size = len(current_data)
+
+ layer_info.append({
+ 'layer': layer_num,
+ 'carrier': carrier_name,
+ 'capacity': capacity,
+ 'payload_size': data_size,
+ 'fits': data_size <= capacity
+ })
+
+ if data_size > capacity:
+ raise ValueError(f"Layer {layer_num} ({carrier_name}): payload {data_size} bytes exceeds capacity {capacity} bytes")
+
+ # Encrypt if password provided (only innermost layer or all?)
+ data_to_encode = current_data
+ if password and i == 0: # Only encrypt the innermost payload
+ data_to_encode = crypto.encrypt(current_data, password)
+
+ # Encode current data into this carrier
+ encoded_img = encode(carrier_img, data_to_encode, config)
+
+ # If there are more carriers, convert this to PNG bytes for next layer
+ if i < len(carriers) - 1:
+ buffer = io.BytesIO()
+ encoded_img.save(buffer, format='PNG')
+ current_data = buffer.getvalue()
+ layer_info[-1]['output_size'] = len(current_data)
+ else:
+ # Final layer - return the image
+ layer_info[-1]['output_size'] = 'final'
+
+ return encoded_img, layer_info
+
+
+def _calculate_capacity_bytes(image: Image.Image, config) -> int:
+ """Calculate byte capacity for an image with given config (simplified, returns int)"""
+ width, height = image.size
+ channels = len(config.channels)
+ bits = config.bits_per_channel
+ # Account for STEG header overhead (~48 bytes)
+ raw_capacity = (width * height * channels * bits) // 8
+ return max(0, raw_capacity - 64) # Reserve space for header
+
+
+# ============== UI COMPONENTS ==============
+
+def create_header():
+ """Create the main header with ASCII art"""
+ with ui.column().classes('w-full items-center'):
+ ui.html(f'{BANNER_ASCII}', sanitize=False)
+ ui.label('🦕 Ultimate LSB Steganography Suite 🦕').classes('text-cyan-400 text-lg')
+
+
+def create_channel_selector(advanced: bool = False):
+ """Create channel selection UI"""
+ if advanced:
+ # All 15 presets
+ options = list(CHANNEL_PRESETS.keys())
+ else:
+ # Simple options
+ options = ['RGB', 'RGBA', 'R', 'G', 'B']
+ return options
+
+
+def create_capacity_display(capacity_label: ui.label, image: Image.Image, config_dict: dict):
+ """Update capacity display"""
+ if image is None:
+ capacity_label.set_text('No image loaded')
+ return
+
+ config = create_config(**config_dict)
+ cap = calculate_capacity(image, config)
+
+ text = f"📊 Capacity: {cap['human']} ({cap['usable_bytes']:,} bytes)"
+ capacity_label.set_text(text)
+ state.capacity_info = cap
+
+
+def create_file_picker(element_id: str, accept: str, endpoint: str, tab: str = 'encode'):
+ """Create a custom file picker that works with Python 3.9
+
+ Returns tuple of (html, javascript) - call ui.html() for html and ui.add_body_html() for js
+ Args:
+ element_id: Unique ID for the element
+ accept: File types to accept
+ endpoint: API endpoint to POST to
+ tab: Tab name to restore after reload ('encode', 'decode', 'analyze')
+ """
+ # Cleaner accept display
+ accept_display = accept.replace(',', ' · ').replace('.', '').upper() if accept != '*' else 'ANY FILE'
+
+ html = f"""
+
+
+ ↑ DROP FILE OR CLICK TO BROWSE
+
+
{accept_display}
+
+
+ """
+
+ js = f"""
+
+ """
+ return html, js
+
+
+# ============== API ENDPOINTS FOR FILE UPLOADS ==============
+
+@app.post('/api/upload/carrier')
+async def upload_carrier(request: Request):
+ """Handle carrier image upload"""
+ try:
+ body = await request.json()
+ b64_data = body.get('data', '')
+ filename = body.get('filename', 'image.png')
+ tab = body.get('tab', 'encode')
+
+ state.carrier_image = base64_to_image(b64_data)
+ state.carrier_path = filename
+ state.active_tab = tab
+ return JSONResponse({'success': True, 'filename': filename})
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/upload/payload')
+async def upload_payload(request: Request):
+ """Handle payload file upload"""
+ try:
+ body = await request.json()
+ b64_data = body.get('data', '')
+ filename = body.get('filename', 'file.bin')
+ tab = body.get('tab', 'encode')
+
+ # Decode base64 data
+ if ',' in b64_data:
+ b64_data = b64_data.split(',')[1]
+ state.encode_file_data = base64.b64decode(b64_data)
+ state.encode_file_name = filename
+ state.active_tab = tab
+ return JSONResponse({'success': True, 'filename': filename})
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/upload/decode')
+async def upload_decode(request: Request):
+ """Handle decode image upload"""
+ try:
+ body = await request.json()
+ b64_data = body.get('data', '')
+ filename = body.get('filename', 'image.png')
+ tab = body.get('tab', 'decode')
+
+ state.decode_image = base64_to_image(b64_data)
+ state.active_tab = tab
+
+ # Try auto-detection
+ detection = detect_encoding(state.decode_image)
+ state.detected_config = detection
+
+ return JSONResponse({
+ 'success': True,
+ 'filename': filename,
+ 'detected': detection is not None,
+ 'config': detection['config'] if detection else None,
+ 'payload_size': detection['original_length'] if detection else 0
+ })
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/upload/analyze')
+async def upload_analyze(request: Request):
+ """Handle analyze image upload"""
+ try:
+ body = await request.json()
+ b64_data = body.get('data', '')
+ filename = body.get('filename', 'image.png')
+ tab = body.get('tab', 'analyze')
+
+ state.analyze_image = base64_to_image(b64_data)
+ state.active_tab = tab
+ return JSONResponse({'success': True, 'filename': filename})
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/matryoshka/toggle')
+async def toggle_matryoshka(request: Request):
+ """🪆 Toggle Matryoshka mode"""
+ try:
+ body = await request.json()
+ active = body.get('active', False)
+ state.matryoshka_mode = active
+ return JSONResponse({
+ 'success': True,
+ 'matryoshka_active': state.matryoshka_mode,
+ 'depth': state.matryoshka_depth
+ })
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/matryoshka/decode')
+async def matryoshka_decode_api(request: Request):
+ """🪆 Perform recursive Matryoshka decode"""
+ try:
+ body = await request.json()
+ depth = body.get('depth', state.matryoshka_depth)
+ password = body.get('password', None)
+
+ if state.decode_image is None:
+ return JSONResponse({'success': False, 'error': 'No image loaded'})
+
+ results = matryoshka_decode(state.decode_image, max_depth=depth, password=password)
+ state.matryoshka_results = results
+
+ # Flatten results for JSON response
+ def flatten_results(res_list, level=0):
+ flat = []
+ for r in res_list:
+ flat.append({
+ 'depth': r.get('depth', level),
+ 'type': r.get('type', 'unknown'),
+ 'filename': r.get('filename'),
+ 'data_size': r.get('data_size', 0),
+ 'preview': r.get('preview', '')[:500],
+ 'has_nested': r.get('has_nested', False),
+ })
+ if r.get('nested_results'):
+ flat.extend(flatten_results(r['nested_results'], level + 1))
+ return flat
+
+ flat_results = flatten_results(results)
+
+ return JSONResponse({
+ 'success': True,
+ 'layers_found': len(flat_results),
+ 'results': flat_results
+ })
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/matryoshka/add_carrier')
+async def add_matryoshka_carrier(request: Request):
+ """🪆 Add a carrier image to the Matryoshka stack"""
+ try:
+ body = await request.json()
+ b64_data = body.get('data', '')
+ filename = body.get('filename', 'carrier.png')
+
+ img = base64_to_image(b64_data)
+ state.matryoshka_carriers.append((img, filename))
+
+ # Calculate capacity for this carrier
+ config = create_config(channels='RGBA', bits=2)
+ capacity = _calculate_capacity_bytes(img, config)
+
+ return JSONResponse({
+ 'success': True,
+ 'carrier_count': len(state.matryoshka_carriers),
+ 'filename': filename,
+ 'dimensions': f"{img.size[0]}x{img.size[1]}",
+ 'capacity': capacity
+ })
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.post('/api/matryoshka/clear_carriers')
+async def clear_matryoshka_carriers():
+ """🪆 Clear all Matryoshka carrier images"""
+ state.matryoshka_carriers = []
+ state.matryoshka_encode_result = None
+ return JSONResponse({'success': True, 'carrier_count': 0})
+
+
+@app.post('/api/matryoshka/encode')
+async def matryoshka_encode_api(request: Request):
+ """🪆 Perform recursive Matryoshka encode"""
+ try:
+ body = await request.json()
+ password = body.get('password')
+ channels = body.get('channels', 'RGBA')
+ bits = body.get('bits', 2)
+
+ if not state.matryoshka_carriers:
+ return JSONResponse({'success': False, 'error': 'No carrier images added'})
+
+ # Get payload - either from encode_file_data or encode_text
+ if state.encode_file_data:
+ payload = state.encode_file_data
+ # Prepend filename if available
+ if state.encode_file_name:
+ fn_bytes = state.encode_file_name.encode('utf-8')
+ payload = bytes([len(fn_bytes)]) + fn_bytes + payload
+ elif hasattr(state, 'encode_text') and state.encode_text:
+ payload = state.encode_text.encode('utf-8')
+ else:
+ return JSONResponse({'success': False, 'error': 'No payload to encode. Upload a file or enter text first.'})
+
+ config = create_config(channels=channels, bits=bits)
+
+ # Carriers are stored outermost-first in UI, but we need innermost-first for encoding
+ # So reverse the list
+ carriers_reversed = list(reversed(state.matryoshka_carriers))
+
+ result_img, layer_info = matryoshka_encode(
+ payload=payload,
+ carriers=carriers_reversed,
+ config=config,
+ password=password
+ )
+
+ state.matryoshka_encode_result = result_img
+
+ # Convert result to base64 for preview
+ buffer = io.BytesIO()
+ result_img.save(buffer, format='PNG')
+ result_b64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
+
+ return JSONResponse({
+ 'success': True,
+ 'layers': len(state.matryoshka_carriers),
+ 'layer_info': layer_info,
+ 'result_b64': f"data:image/png;base64,{result_b64}"
+ })
+ except Exception as e:
+ return JSONResponse({'success': False, 'error': str(e)})
+
+
+@app.get('/api/matryoshka/download')
+async def download_matryoshka_result():
+ """🪆 Download the Matryoshka encoded result"""
+ if state.matryoshka_encode_result is None:
+ return JSONResponse({'success': False, 'error': 'No result to download'})
+
+ buffer = io.BytesIO()
+ state.matryoshka_encode_result.save(buffer, format='PNG')
+ buffer.seek(0)
+
+ return Response(
+ content=buffer.getvalue(),
+ media_type='image/png',
+ headers={'Content-Disposition': 'attachment; filename="matryoshka_encoded.png"'}
+ )
+
+
+# ============== MAIN PAGE ==============
+
+@ui.page('/')
+async def main_page():
+ # Add custom CSS
+ ui.add_head_html(f'')
+ ui.add_head_html('')
+
+ # Matrix scanline overlay
+ ui.html('', sanitize=False)
+
+ with ui.column().classes('w-full max-w-6xl mx-auto p-4 gap-4'):
+ create_header()
+
+ # Tab navigation
+ with ui.tabs().classes('w-full') as tabs:
+ encode_tab = ui.tab('ENCODE', icon='lock').classes('text-green-400')
+ decode_tab = ui.tab('DECODE', icon='lock_open').classes('text-cyan-400')
+ analyze_tab = ui.tab('ANALYZE', icon='search').classes('text-yellow-400')
+ # 🪆 Hidden Matryoshka tab - revealed by easter egg
+ matryoshka_tab = ui.tab('🪆 MATRYOSHKA', icon='auto_awesome').classes('text-pink-400')
+ matryoshka_tab.props('id="matryoshka-tab"')
+ matryoshka_tab.style('display: none;') # Hidden until activated
+
+ # Determine initial tab from state (set by API) or URL hash
+ tab_map = {
+ 'encode': encode_tab,
+ 'decode': decode_tab,
+ 'analyze': analyze_tab,
+ 'matryoshka': matryoshka_tab,
+ }
+ initial_tab = tab_map.get(state.active_tab, encode_tab)
+
+ with ui.tab_panels(tabs, value=initial_tab).classes('w-full') as panels:
+
+ # ==================== ENCODE PANEL ====================
+ with ui.tab_panel(encode_tab):
+ await create_encode_panel()
+
+ # ==================== DECODE PANEL ====================
+ with ui.tab_panel(decode_tab):
+ await create_decode_panel()
+
+ # ==================== ANALYZE PANEL ====================
+ with ui.tab_panel(analyze_tab):
+ await create_analyze_panel()
+
+ # ==================== MATRYOSHKA PANEL ====================
+ with ui.tab_panel(matryoshka_tab):
+ await create_matryoshka_panel()
+
+ # JavaScript to read URL hash on load and sync tabs
+ ui.add_body_html("""
+
+ """)
+
+ # Footer
+ ui.html('.-.-.-.-<={LOVE PLINY}=>-.-.-.-.
', sanitize=False)
+
+ # 🪆 MATRYOSHKA MODE - Easter egg trigger zone (bottom-left pixel)
+ ui.html('', sanitize=False)
+
+ # Simple mode indicator (shown when active)
+ ui.html('''
+
+ 🪆 MATRYOSHKA MODE
+ click to deactivate
+
+ ''', sanitize=False)
+
+ # JavaScript to handle easter egg activation
+ ui.add_body_html("""
+
+ """)
+
+
+async def create_encode_panel():
+ """Create the encoding interface"""
+
+ # Use global state for persistence
+ global state
+
+ with ui.row().classes('w-full gap-4'):
+ # Left column - Image upload and preview
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('📷 CARRIER IMAGE').classes('text-green-400 text-lg font-bold')
+
+ # Show current image if loaded
+ if state.carrier_image:
+ b64 = image_to_base64(state.carrier_image)
+ ui.image(f'data:image/png;base64,{b64}').classes('w-full max-h-64 object-contain')
+ w, h = state.carrier_image.size
+ ui.label(f'{state.carrier_path or "Generated"} | {w}x{h}').classes('text-green-400 text-sm')
+
+ # Show capacity
+ try:
+ config = create_config(channels='RGB', bits=1)
+ cap = calculate_capacity(state.carrier_image, config)
+ ui.label(f"📊 Capacity: {cap['human']} ({cap['usable_bytes']:,} bytes)").classes('text-cyan-400')
+ except:
+ pass
+ else:
+ ui.image().classes('w-full max-h-64 object-contain')
+ ui.label('No image loaded').classes('text-gray-500 text-sm')
+ ui.label('📊 Capacity: --').classes('text-cyan-400')
+
+ # Custom file picker - uses API endpoint
+ picker_html, picker_js = create_file_picker('encode_carrier', '.png,.PNG,image/png', '/api/upload/carrier', tab='encode')
+ ui.html(picker_html, sanitize=False)
+ ui.add_body_html(picker_js)
+
+ # Generate image buttons
+ ui.label('Or generate a carrier:').classes('text-gray-500 text-sm mt-2')
+
+ # Size selector for generated images
+ with ui.row().classes('gap-2 items-center'):
+ gen_size_select = ui.select(
+ options=['512x512', '1024x1024', '2048x2048', '4096x4096'],
+ value='1024x1024',
+ label='Size'
+ ).props('dense').classes('w-24')
+
+ with ui.row().classes('gap-2'):
+ async def gen_image(color: str):
+ size_str = gen_size_select.value or '1024x1024'
+ w, h = map(int, size_str.split('x'))
+ state.carrier_image = generate_blank_image(w, h, color)
+ state.carrier_path = f'{color}_{size_str}.png'
+ state.active_tab = 'encode'
+ ui.notify(f'Generated {color} image ({size_str})', type='positive')
+ await asyncio.sleep(0.3)
+ await ui.run_javascript('window.location.hash = "encode"; window.location.reload();')
+
+ ui.button('🎲 Noise', on_click=lambda: gen_image('noise')).props('dense').classes('text-xs')
+ ui.button('⬛ Black', on_click=lambda: gen_image('black')).props('dense').classes('text-xs')
+ ui.button('🌈 Gradient', on_click=lambda: gen_image('gradient')).props('dense').classes('text-xs')
+ ui.button('⬜ White', on_click=lambda: gen_image('white')).props('dense').classes('text-xs')
+
+ # Right column - Data and settings
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('📝 PAYLOAD DATA').classes('text-green-400 text-lg font-bold')
+
+ # Data source toggle
+ with ui.row().classes('gap-2'):
+ data_source = ui.toggle(
+ {
+ 'text': '💬 Text',
+ 'file': '📁 File',
+ },
+ value='text'
+ ).classes('w-full')
+
+ # Text input
+ text_container = ui.column().classes('w-full')
+ with text_container:
+ text_input = ui.textarea(
+ label='Secret message',
+ placeholder='Enter your hidden message here...'
+ ).classes('w-full').props('rows=6 dark')
+
+ # File upload for payload
+ file_container = ui.column().classes('w-full hidden')
+ with file_container:
+ if state.encode_file_data and state.encode_file_name:
+ ui.label(f'📄 {state.encode_file_name} ({format_size(len(state.encode_file_data))})').classes('text-green-400')
+ else:
+ ui.label('No file selected').classes('text-gray-500')
+
+ payload_html, payload_js = create_file_picker('encode_payload', '*', '/api/upload/payload', tab='encode')
+ ui.html(payload_html, sanitize=False)
+ ui.add_body_html(payload_js)
+
+ def toggle_data_source():
+ if data_source.value == 'text':
+ text_container.classes(remove='hidden')
+ file_container.classes(add='hidden')
+ else:
+ text_container.classes(add='hidden')
+ file_container.classes(remove='hidden')
+
+ data_source.on('update:model-value', toggle_data_source)
+
+ ui.separator().classes('my-2')
+
+ # ===== ENCODING SETTINGS =====
+ ui.label('⚙️ ENCODING SETTINGS').classes('text-green-400 text-lg font-bold')
+
+ # Advanced mode toggle
+ advanced_toggle = ui.switch('🔧 Advanced Mode').classes('text-cyan-400')
+
+ # Basic settings (always visible)
+ with ui.row().classes('w-full gap-4'):
+ channel_select = ui.select(
+ label='Channels',
+ options=['RGB', 'RGBA', 'R', 'G', 'B'],
+ value='RGB'
+ ).classes('w-1/2')
+
+ bits_select = ui.select(
+ label='Bits/Channel',
+ options=[1, 2],
+ value=1
+ ).classes('w-1/2')
+
+ # Live capacity display
+ capacity_display = ui.html('', sanitize=False).classes('w-full')
+
+ def update_capacity_display():
+ """Update capacity display based on current settings"""
+ if state.carrier_image is None:
+ capacity_display.set_content(
+ ''
+ '📊 Capacity: Load an image to see capacity'
+ '
'
+ )
+ return
+
+ try:
+ if advanced_toggle.value:
+ ch = adv_channel_select.value or 'RGB'
+ b = adv_bits_select.value or 1
+ else:
+ ch = channel_select.value or 'RGB'
+ b = bits_select.value or 1
+
+ config = create_config(channels=ch, bits=b)
+ cap = calculate_capacity(state.carrier_image, config)
+
+ # Calculate fill percentage if we have data
+ data_size = 0
+ if data_source.value == 'text' and text_input.value:
+ data_size = len(text_input.value.encode('utf-8'))
+ elif data_source.value == 'file' and state.encode_file_data:
+ data_size = len(state.encode_file_data) + len(state.encode_file_name or '') + 1
+
+ # Don't cap at 100% - we need to detect overflow!
+ fill_pct = (data_size / max(1, cap['usable_bytes'])) * 100
+ bar_width = min(100, fill_pct)
+
+ if fill_pct > 100:
+ color = '#ff3333'
+ bar_color = '#ff3333'
+ status = 'OVER CAPACITY'
+ icon = '⛔'
+ elif fill_pct > 80:
+ color = '#ffd000'
+ bar_color = '#ffd000'
+ status = 'NEARLY FULL'
+ icon = '⚠️'
+ elif fill_pct > 0:
+ color = '#00ff41'
+ bar_color = '#00ff41'
+ status = 'READY'
+ icon = '✓'
+ else:
+ color = '#555'
+ bar_color = '#333'
+ status = 'NO DATA'
+ icon = '○'
+
+ capacity_display.set_content(
+ f''
+ f'
'
+ f'CAPACITY'
+ f'{cap["usable_bytes"]:,} bytes'
+ f'
'
+ f'
'
+ f'
'
+ f'{icon} {status}'
+ f'{format_size(data_size)} / {cap["human"]} ({fill_pct:.1f}%)'
+ f'
'
+ f'
'
+ )
+
+ # Store capacity for encode check
+ state.capacity_info = cap
+
+ except Exception as e:
+ capacity_display.set_content(f'Error: {e}
')
+
+ # Update capacity on various changes
+ if state.carrier_image:
+ update_capacity_display()
+
+ # Advanced settings (hidden by default)
+ advanced_container = ui.column().classes('w-full hidden gap-2')
+ with advanced_container:
+ ui.label('🎯 Channel & Bit Configuration').classes('text-yellow-400 text-sm')
+
+ with ui.row().classes('w-full gap-4'):
+ # All channel presets
+ adv_channel_select = ui.select(
+ label='All Channel Presets',
+ options=list(CHANNEL_PRESETS.keys()),
+ value='RGB'
+ ).classes('w-1/2')
+
+ # Full bits range
+ adv_bits_select = ui.select(
+ label='Bits (1-8)',
+ options=list(range(1, 9)),
+ value=1
+ ).classes('w-1/2')
+
+ with ui.row().classes('w-full gap-4'):
+ strategy_select = ui.select(
+ label='Strategy',
+ options=['interleaved', 'sequential', 'spread', 'randomized'],
+ value='interleaved'
+ ).classes('w-1/2')
+
+ bit_offset_select = ui.select(
+ label='Bit Offset (0=LSB, 7=MSB)',
+ options=list(range(8)),
+ value=0
+ ).classes('w-1/2')
+
+ ui.separator().classes('my-2')
+ ui.label('🎲 Randomization').classes('text-yellow-400 text-sm')
+
+ with ui.row().classes('w-full gap-4'):
+ seed_input = ui.number(
+ label='Random Seed (manual)',
+ value=None,
+ placeholder='Optional'
+ ).classes('w-1/2')
+
+ seed_password = ui.input(
+ label='Or derive seed from password',
+ placeholder='Enter passphrase...'
+ ).classes('w-1/2')
+
+ def derive_seed_from_password():
+ """Derive a numeric seed from password hash"""
+ if seed_password.value:
+ import hashlib
+ hash_bytes = hashlib.sha256(seed_password.value.encode()).digest()
+ derived = int.from_bytes(hash_bytes[:4], 'big')
+ seed_input.value = derived
+ ui.notify(f'Derived seed: {derived}', type='info')
+
+ ui.button('🔑 Derive Seed', on_click=derive_seed_from_password).props('dense').classes('text-xs')
+
+ ui.separator().classes('my-2')
+ ui.label('🛡️ Stealth & Obfuscation').classes('text-yellow-400 text-sm')
+
+ with ui.row().classes('w-full gap-4'):
+ compress_toggle = ui.switch('Compress data', value=True).classes('w-1/2')
+ invert_toggle = ui.switch('XOR invert bits', value=False).classes('w-1/2')
+
+ ui.html(
+ ''
+ '• Compression reduces payload size but adds header bytes
'
+ '• XOR invert flips all bits for additional obfuscation'
+ '
', sanitize=False
+ )
+
+ ui.separator().classes('my-2')
+ ui.label('📊 Encoding Statistics').classes('text-yellow-400 text-sm')
+
+ stats_display = ui.html('', sanitize=False)
+
+ def update_stats():
+ """Show encoding statistics preview"""
+ if state.carrier_image is None:
+ stats_display.set_content('Load carrier to see stats
')
+ return
+
+ w, h = state.carrier_image.size
+ total_pixels = w * h
+
+ ch = adv_channel_select.value or 'RGB'
+ b = adv_bits_select.value or 1
+ num_channels = len(CHANNEL_PRESETS.get(ch, []))
+
+ bits_per_pixel = num_channels * b
+ total_bits = total_pixels * bits_per_pixel
+ total_bytes = total_bits // 8
+
+ # Estimate pixels modified for a 1KB payload
+ sample_payload = 1024 * 8 # 1KB in bits
+ pixels_for_1kb = sample_payload // bits_per_pixel if bits_per_pixel else 0
+
+ stats_display.set_content(
+ f''
+ f'Image: {w}x{h} = {total_pixels:,} pixels
'
+ f'Channels: {ch} ({num_channels} channels)
'
+ f'Bits/pixel: {bits_per_pixel} ({b} bits × {num_channels} ch)
'
+ f'Raw capacity: {total_bytes:,} bytes
'
+ f'Pixels for 1KB: ~{pixels_for_1kb:,} ({pixels_for_1kb/total_pixels*100:.2f}% of image)'
+ f'
'
+ )
+
+ # Initial stats update
+ update_stats()
+
+ # Update stats when settings change
+ adv_channel_select.on('update:model-value', lambda: (update_stats(), update_capacity_display()))
+ adv_bits_select.on('update:model-value', lambda: (update_stats(), update_capacity_display()))
+
+ def toggle_advanced():
+ if advanced_toggle.value:
+ advanced_container.classes(remove='hidden')
+ # Sync values
+ adv_channel_select.value = channel_select.value
+ adv_bits_select.value = bits_select.value
+ update_stats()
+ update_capacity_display()
+ else:
+ advanced_container.classes(add='hidden')
+ update_capacity_display()
+
+ advanced_toggle.on('update:model-value', toggle_advanced)
+
+ # Also update capacity when basic settings change
+ channel_select.on('update:model-value', update_capacity_display)
+ bits_select.on('update:model-value', update_capacity_display)
+
+ # Encryption settings
+ ui.separator().classes('my-2')
+ ui.label('🔐 ENCRYPTION').classes('text-green-400 text-lg font-bold')
+
+ encrypt_toggle = ui.switch('Enable Encryption').classes('text-cyan-400')
+
+ encrypt_container = ui.column().classes('w-full hidden gap-2')
+ with encrypt_container:
+ password_input = ui.input(
+ label='Password',
+ password=True,
+ password_toggle_button=True
+ ).classes('w-full')
+
+ crypto_methods = crypto.get_available_methods()
+ encrypt_method = ui.select(
+ label='Method',
+ options=crypto_methods,
+ value=crypto_methods[0] if crypto_methods else 'xor'
+ ).classes('w-full')
+
+ def toggle_encrypt():
+ if encrypt_toggle.value:
+ encrypt_container.classes(remove='hidden')
+ else:
+ encrypt_container.classes(add='hidden')
+
+ encrypt_toggle.on('update:model-value', toggle_encrypt)
+
+ ui.separator().classes('my-4')
+
+ # ENCODE BUTTON & OUTPUT
+ with ui.row().classes('w-full justify-center gap-4'):
+ async def do_encode():
+ if state.carrier_image is None:
+ ui.notify('Please upload or generate a carrier image first', type='negative')
+ return
+
+ # Get data to encode
+ if data_source.value == 'text':
+ if not text_input.value:
+ ui.notify('Please enter some text to hide', type='negative')
+ return
+ data = text_input.value.encode('utf-8')
+ else:
+ if not state.encode_file_data:
+ ui.notify('Please upload a file to hide', type='negative')
+ return
+ # Prepend filename for extraction
+ filename_bytes = state.encode_file_name.encode('utf-8')
+ data = bytes([len(filename_bytes)]) + filename_bytes + state.encode_file_data
+
+ # Build config
+ if advanced_toggle.value:
+ config = create_config(
+ channels=adv_channel_select.value or 'RGB',
+ bits=adv_bits_select.value or 1,
+ strategy=strategy_select.value or 'interleaved',
+ bit_offset=bit_offset_select.value or 0,
+ seed=int(seed_input.value) if seed_input.value else None,
+ compress=compress_toggle.value,
+ )
+ else:
+ config = create_config(
+ channels=channel_select.value or 'RGB',
+ bits=bits_select.value or 1,
+ )
+
+ # Check capacity before proceeding
+ cap = calculate_capacity(state.carrier_image, config)
+ # Account for header overhead (~32 bytes + compression metadata)
+ estimated_size = len(data) + 64 # Conservative estimate with header
+ if estimated_size > cap['usable_bytes']:
+ ui.notify(
+ f'⛔ Data too large! Need {format_size(estimated_size)}, but capacity is {cap["human"]}. '
+ f'Try: more bits, more channels, or larger image.',
+ type='negative',
+ timeout=8000
+ )
+ return
+
+ # Encrypt if enabled
+ if encrypt_toggle.value:
+ if not password_input.value:
+ ui.notify('Please enter an encryption password', type='negative')
+ return
+ data = crypto.encrypt(data, password_input.value, encrypt_method.value)
+
+ # XOR invert if enabled (advanced mode only)
+ if advanced_toggle.value and invert_toggle.value:
+ data = bytes(b ^ 0xFF for b in data)
+
+ try:
+ result = encode(state.carrier_image, data, config)
+
+ # Calculate encoding stats
+ data_bits = len(data) * 8
+ if advanced_toggle.value:
+ ch = adv_channel_select.value or 'RGB'
+ b = adv_bits_select.value or 1
+ else:
+ ch = channel_select.value or 'RGB'
+ b = bits_select.value or 1
+ num_ch = len(CHANNEL_PRESETS.get(ch, []))
+ bits_per_pixel = num_ch * b
+ pixels_used = (data_bits // bits_per_pixel) + 1 if bits_per_pixel else 0
+ total_pixels = state.carrier_image.size[0] * state.carrier_image.size[1]
+ pct_used = (pixels_used / total_pixels) * 100 if total_pixels else 0
+
+ # Show result
+ b64 = image_to_base64(result)
+ result_preview.set_source(f'data:image/png;base64,{b64}')
+ result_container.classes(remove='hidden')
+
+ # Update result stats
+ result_stats.set_content(
+ f''
+ f'✓ Encoded {format_size(len(data))} into {pixels_used:,} pixels ({pct_used:.2f}% of image)
'
+ f'Config: {ch} @ {b} bits | Strategy: {strategy_select.value if advanced_toggle.value else "interleaved"}'
+ f'{"
🔐 Encrypted" if encrypt_toggle.value else ""}'
+ f'{"
🔀 XOR Inverted" if advanced_toggle.value and invert_toggle.value else ""}'
+ f'
'
+ )
+
+ # Store for download
+ state.encode_result = result
+
+ ui.notify('✓ Encoding successful!', type='positive')
+
+ except Exception as e:
+ ui.notify(f'Encoding failed: {str(e)}', type='negative')
+
+ ui.button('🔒 ENCODE', on_click=do_encode).classes(
+ 'bg-green-600 hover:bg-green-500 text-black font-bold px-8 py-2 text-lg'
+ )
+
+ # Result display
+ result_container = ui.column().classes('w-full hidden gap-2 mt-4')
+ with result_container:
+ ui.label('✓ ENCODED IMAGE').classes('text-green-400 text-lg font-bold')
+ result_stats = ui.html('', sanitize=False).classes('w-full')
+ result_preview = ui.image().classes('w-full max-h-96 object-contain')
+
+ async def download_result():
+ if state.encode_result is None:
+ return
+
+ # Create download
+ buffer = io.BytesIO()
+ state.encode_result.save(buffer, format='PNG')
+ buffer.seek(0)
+
+ ui.download(buffer.getvalue(), 'stego_output.png')
+
+ ui.button('💾 Download Encoded Image', on_click=download_result).classes(
+ 'bg-cyan-600 hover:bg-cyan-500 text-black font-bold'
+ )
+
+
+async def create_decode_panel():
+ """Create the decoding interface - Aperi'Solve inspired"""
+
+ global state
+
+ # Store scan results for UI updates
+ scan_results_data = {"results": [], "selected": None}
+
+ with ui.row().classes('w-full gap-4'):
+ # Left column - Image upload
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('🖼️ STEGO IMAGE').classes('text-cyan-400 text-lg font-bold')
+
+ # Show current image if loaded
+ if state.decode_image:
+ b64 = image_to_base64(state.decode_image)
+ ui.image(f'data:image/png;base64,{b64}').classes('w-full max-h-48 object-contain')
+ w, h = state.decode_image.size
+ ui.label(f'Loaded | {w}x{h} | {w*h:,} pixels').classes('text-green-400 text-sm')
+
+ # Quick detection status
+ if state.detected_config:
+ ui.html(
+ f''
+ f'✓ STEG v3 header detected
'
+ f'Channels: {state.detected_config["config"]["channels"]}
'
+ f'Payload: {format_size(state.detected_config["original_length"])}'
+ f'
', sanitize=False
+ )
+ else:
+ ui.html(
+ ''
+ '⚠ No STEG header - use Smart Scan to detect LSB data'
+ '
', sanitize=False
+ )
+ else:
+ ui.image().classes('w-full max-h-48 object-contain')
+ ui.label('No image loaded').classes('text-gray-500 text-sm')
+
+ # Custom file picker
+ decode_html, decode_js = create_file_picker('decode_image', '.png,.PNG,image/png', '/api/upload/decode', tab='decode')
+ ui.html(decode_html, sanitize=False)
+ ui.add_body_html(decode_js)
+
+ # Right column - Decode options
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('⚙️ DECODE OPTIONS').classes('text-cyan-400 text-lg font-bold')
+
+ # Decryption password (optional)
+ ui.label('🔓 Decryption (optional)').classes('text-sm text-gray-400')
+ dec_password_input = ui.input(
+ label='Password (leave empty if unencrypted)',
+ password=True,
+ password_toggle_button=True
+ ).classes('w-full')
+
+ ui.separator().classes('my-2')
+
+ # Manual decode settings
+ ui.label('📋 Manual Config (for targeted decode)').classes('text-sm text-gray-400')
+ with ui.row().classes('w-full gap-2'):
+ dec_channel_select = ui.select(
+ label='Channels',
+ options=list(CHANNEL_PRESETS.keys()),
+ value='RGB'
+ ).classes('w-1/3')
+
+ dec_bits_select = ui.select(
+ label='Bits',
+ options=list(range(1, 9)),
+ value=1
+ ).classes('w-1/3')
+
+ dec_strategy_select = ui.select(
+ label='Strategy',
+ options=['interleaved', 'sequential', 'spread', 'randomized'],
+ value='interleaved'
+ ).classes('w-1/3')
+
+ ui.separator().classes('my-4')
+
+ # ACTION BUTTONS
+ with ui.row().classes('w-full justify-center gap-4'):
+ async def do_smart_scan():
+ """Run Aperi'Solve-style multi-config scan"""
+ if state.decode_image is None:
+ ui.notify('Please upload an image first', type='negative')
+ return
+
+ ui.notify('🔍 Scanning all channel combinations...', type='info')
+
+ try:
+ password = dec_password_input.value if dec_password_input.value else None
+ results = smart_scan_image(state.decode_image, password)
+ scan_results_data["results"] = results
+
+ # Build results HTML table
+ html = ''
+ html += '
'
+ html += ''
+ html += '| Config | '
+ html += 'Channels | '
+ html += 'Bits | '
+ html += 'Status | '
+ html += 'Conf | '
+ html += 'Preview | '
+ html += '
'
+
+ for i, r in enumerate(results):
+ # Status color coding
+ if r["status"] == "STEG_DETECTED" or r["status"] == "STEG_HEADER":
+ status_color = "#00ff00"
+ status_icon = "✓"
+ row_bg = "rgba(0, 255, 0, 0.15)"
+ elif r["status"] == "TEXT_FOUND":
+ status_color = "#00ffff"
+ status_icon = "📝"
+ row_bg = "rgba(0, 255, 255, 0.15)"
+ elif r["status"] == "POSSIBLE_TEXT":
+ status_color = "#ffff00"
+ status_icon = "?"
+ row_bg = "rgba(255, 255, 0, 0.1)"
+ elif r["status"] == "MIXED_DATA":
+ status_color = "#ff9900"
+ status_icon = "~"
+ row_bg = "transparent"
+ else:
+ status_color = "#666666"
+ status_icon = "·"
+ row_bg = "transparent"
+
+ # Confidence bar
+ conf = r.get("confidence", 0)
+ conf_bar = f''
+
+ # Preview (escape HTML)
+ preview = r.get("preview", "")[:60]
+ preview_escaped = preview.replace("<", "<").replace(">", ">").replace("&", "&")
+ if len(r.get("preview", "")) > 60:
+ preview_escaped += "..."
+
+ channels_str = ",".join(r["channels"]) if isinstance(r["channels"], list) else str(r["channels"])
+
+ html += f''
+ html += f'| {r["name"]} | '
+ html += f'{channels_str} | '
+ html += f'{r["bits"]} | '
+ html += f'{status_icon} {r["status"]} | '
+ html += f'{conf_bar} | '
+ html += f'{preview_escaped} | '
+ html += '
'
+
+ html += '
'
+
+ # Summary
+ text_found = sum(1 for r in results if r["status"] in ["TEXT_FOUND", "STEG_DETECTED", "STEG_HEADER"])
+ possible = sum(1 for r in results if r["status"] == "POSSIBLE_TEXT")
+
+ summary = f''
+ summary += f'📊 Scanned {len(results)} configurations | '
+ summary += f'Found: {text_found} likely | '
+ summary += f'Possible: {possible}'
+ summary += '
'
+
+ scan_results_container.set_content(html + summary)
+ scan_results_container.classes(remove='hidden')
+
+ if text_found > 0:
+ ui.notify(f'✓ Found {text_found} potential hidden data!', type='positive')
+ else:
+ ui.notify('Scan complete - no obvious text found', type='warning')
+
+ except Exception as e:
+ ui.notify(f'Scan failed: {str(e)}', type='negative')
+
+ async def do_decode():
+ """Standard decode with current settings"""
+ if state.decode_image is None:
+ ui.notify('Please upload an image first', type='negative')
+ return
+
+ config = create_config(
+ channels=dec_channel_select.value or 'RGB',
+ bits=dec_bits_select.value or 1,
+ strategy=dec_strategy_select.value or 'interleaved',
+ )
+
+ try:
+ data = decode(state.decode_image, config)
+
+ # Decrypt if needed
+ if dec_password_input.value:
+ try:
+ data = crypto.decrypt(data, dec_password_input.value)
+ except Exception as e:
+ ui.notify(f'Decryption failed: {e}', type='warning')
+
+ # Process extracted data
+ process_extracted_data(data, output_text, file_download_btn)
+ ui.notify('✓ Decode successful!', type='positive')
+
+ except Exception as e:
+ ui.notify(f'Decode failed: {str(e)}', type='negative')
+ output_text.set_value(f"Error: {str(e)}\n\nTry using Smart Scan to detect the correct configuration.")
+
+ async def do_auto_decode():
+ """Auto-decode using detected STEG header"""
+ if state.decode_image is None:
+ ui.notify('Please upload an image first', type='negative')
+ return
+
+ try:
+ # Use auto-detection (config=None)
+ data = decode(state.decode_image, None)
+
+ # Decrypt if needed
+ if dec_password_input.value:
+ try:
+ data = crypto.decrypt(data, dec_password_input.value)
+ except Exception as e:
+ ui.notify(f'Decryption failed: {e}', type='warning')
+
+ # Process extracted data
+ process_extracted_data(data, output_text, file_download_btn)
+ ui.notify('✓ Auto-decode successful!', type='positive')
+
+ except Exception as e:
+ ui.notify(f'Auto-decode failed: {str(e)}. Try Smart Scan.', type='negative')
+ output_text.set_value(f"Auto-decode failed: {str(e)}\n\nThe image may not contain STEG v3 header.\nTry using Smart Scan to detect raw LSB data.")
+
+ ui.button('🔍 SMART SCAN', on_click=do_smart_scan).classes(
+ 'bg-purple-600 hover:bg-purple-500 text-white font-bold px-6 py-2'
+ ).props('unelevated')
+
+ ui.button('🤖 AUTO DECODE', on_click=do_auto_decode).classes(
+ 'bg-cyan-600 hover:bg-cyan-500 text-black font-bold px-6 py-2'
+ )
+
+ ui.button('📋 MANUAL DECODE', on_click=do_decode).classes(
+ 'bg-gray-600 hover:bg-gray-500 text-white font-bold px-6 py-2'
+ )
+
+ # 🪆 MATRYOSHKA MODE BUTTON (only shows when matryoshka mode is active)
+ matryoshka_btn_container = ui.row().classes('w-full justify-center gap-4 mt-2')
+
+ with matryoshka_btn_container:
+ async def do_matryoshka_decode():
+ """🪆 Perform recursive Matryoshka decode"""
+ if state.decode_image is None:
+ ui.notify('Please upload an image first', type='negative')
+ return
+
+ if not state.matryoshka_mode:
+ ui.notify('🪆 Matryoshka mode not active! Click the bottom-left corner to activate.', type='warning')
+ return
+
+ ui.notify(f'🪆 Starting Matryoshka decode (depth: {state.matryoshka_depth})...', type='info')
+
+ try:
+ password = dec_password_input.value if dec_password_input.value else None
+ results = matryoshka_decode(state.decode_image, max_depth=state.matryoshka_depth, password=password)
+ state.matryoshka_results = results
+
+ # Build nested results display
+ def build_results_html(res_list, indent=0):
+ html = ''
+ for r in res_list:
+ depth = r.get('depth', indent)
+ indent_px = depth * 24
+ doll_icon = '🪆' * (depth + 1) # More dolls for deeper layers
+
+ # Color based on type
+ if r.get('type') == 'nested_image' or r.get('type') == 'nested_image_raw':
+ color = '#ff00ff'
+ icon = '🖼️'
+ elif r.get('type') == 'text':
+ color = '#00ff00'
+ icon = '📝'
+ elif r.get('type') == 'file':
+ color = '#00ffff'
+ icon = '📁'
+ elif r.get('type') == 'no_data_found':
+ color = '#666666'
+ icon = '∅'
+ elif r.get('type') == 'max_depth_reached':
+ color = '#ffff00'
+ icon = '⚠️'
+ else:
+ color = '#888888'
+ icon = '?'
+
+ html += f'''
+
+
+ {doll_icon} Layer {depth} {icon}
+
+ {r.get('filename') or r.get('type', 'unknown')}
+
+
+
+ Size: {format_size(r.get('data_size', 0))}
+
+
+ {r.get('preview', '')[:200]}
+
+
+ '''
+
+ # Recurse into nested results
+ if r.get('nested_results'):
+ html += build_results_html(r['nested_results'], indent + 1)
+
+ return html
+
+ results_html = build_results_html(results)
+
+ # Count layers found
+ def count_layers(res_list):
+ count = len(res_list)
+ for r in res_list:
+ if r.get('nested_results'):
+ count += count_layers(r['nested_results'])
+ return count
+
+ total_layers = count_layers(results)
+
+ matryoshka_results_container.set_content(
+ f'''
+
+
+ 🪆 MATRYOSHKA DECODE RESULTS
+ ({total_layers} layers found)
+
+ {results_html}
+
+ '''
+ )
+ matryoshka_results_container.classes(remove='hidden')
+
+ # Also extract the deepest/last data for download
+ def get_deepest_data(res_list):
+ for r in reversed(res_list):
+ if r.get('nested_results'):
+ deep = get_deepest_data(r['nested_results'])
+ if deep:
+ return deep
+ if r.get('raw_data'):
+ return r
+ return None
+
+ deepest = get_deepest_data(results)
+ if deepest and deepest.get('raw_data'):
+ state.extracted_file = deepest['raw_data']
+ state.extracted_filename = deepest.get('filename', 'extracted_data.bin')
+ file_download_btn.classes(remove='hidden')
+
+ # Also show in output text
+ try:
+ text = deepest['raw_data'].decode('utf-8')
+ output_text.set_value(f"🪆 Deepest layer content:\n\n{text}")
+ except:
+ output_text.set_value(f"🪆 Deepest layer: {deepest.get('filename', 'binary data')} ({format_size(len(deepest['raw_data']))})")
+
+ ui.notify(f'🪆 Matryoshka decode complete! Found {total_layers} layers.', type='positive')
+
+ except Exception as e:
+ ui.notify(f'🪆 Matryoshka decode failed: {str(e)}', type='negative')
+ matryoshka_results_container.set_content(
+ f'Error: {str(e)}
'
+ )
+
+ matryoshka_decode_btn = ui.button('🪆 MATRYOSHKA DECODE', on_click=do_matryoshka_decode).classes(
+ 'bg-pink-600 hover:bg-pink-500 text-white font-bold px-6 py-2'
+ ).props('unelevated').style('display: none;')
+
+ # Show/hide based on matryoshka mode state
+ ui.add_body_html('''
+
+ ''')
+
+ # 🪆 MATRYOSHKA RESULTS CONTAINER
+ matryoshka_results_container = ui.html('', sanitize=False).classes('w-full hidden mt-4')
+
+ # SCAN RESULTS (hidden until scan runs)
+ ui.label('📊 SCAN RESULTS').classes('text-purple-400 text-lg font-bold mt-4')
+ scan_results_container = ui.html(
+ ''
+ 'Click "Smart Scan" to analyze all channel combinations'
+ '
', sanitize=False
+ ).classes('w-full')
+
+ ui.separator().classes('my-4')
+
+ # OUTPUT SECTION
+ ui.label('📤 EXTRACTED DATA').classes('text-cyan-400 text-lg font-bold')
+ output_text = ui.textarea(
+ label='Decoded output',
+ placeholder='Extracted data will appear here...'
+ ).classes('w-full terminal').props('rows=8 readonly dark')
+
+ with ui.row().classes('gap-2'):
+ async def download_file():
+ if state.extracted_file is None:
+ return
+ ui.download(state.extracted_file, state.extracted_filename)
+
+ file_download_btn = ui.button('💾 Download Extracted File', on_click=download_file).classes(
+ 'bg-green-600 hover:bg-green-500 text-black font-bold hidden'
+ )
+
+ async def download_raw():
+ if not output_text.value:
+ return
+ ui.download(output_text.value.encode('utf-8'), 'extracted_text.txt')
+
+ ui.button('📄 Download as Text', on_click=download_raw).classes(
+ 'bg-gray-600 hover:bg-gray-500 text-white font-bold'
+ )
+
+
+def process_extracted_data(data: bytes, output_text, file_download_btn):
+ """Process and display extracted data"""
+ try:
+ # Check if it's a file (starts with filename length byte)
+ if len(data) > 2 and data[0] < 255:
+ fname_len = data[0]
+ if fname_len > 0 and fname_len < 200:
+ try:
+ filename = data[1:1+fname_len].decode('utf-8')
+ file_data = data[1+fname_len:]
+ # Check if it looks like a valid filename
+ if '.' in filename and len(filename) < 200 and '/' not in filename:
+ state.extracted_file = file_data
+ state.extracted_filename = filename
+ output_text.set_value(
+ f"📁 Extracted file: {filename}\n"
+ f" Size: {format_size(len(file_data))}\n\n"
+ f"(Click 'Download Extracted File' to save)"
+ )
+ file_download_btn.classes(remove='hidden')
+ return
+ except:
+ pass
+
+ # Try as plain text
+ text = data.decode('utf-8')
+ output_text.set_value(text)
+ file_download_btn.classes(add='hidden')
+
+ except UnicodeDecodeError:
+ # Binary data - show hex preview
+ hex_preview = data[:512].hex()
+ formatted = ' '.join(hex_preview[i:i+2] for i in range(0, min(len(hex_preview), 512), 2))
+ output_text.set_value(
+ f"Binary data ({format_size(len(data))}):\n\n{formatted}"
+ + ("..." if len(data) > 256 else "")
+ )
+ state.extracted_file = data
+ state.extracted_filename = 'extracted_data.bin'
+ file_download_btn.classes(remove='hidden')
+
+
+async def create_matryoshka_panel():
+ """Create the Matryoshka (nested steganography) interface"""
+
+ global state
+
+ ui.html('''
+
+
🪆 MATRYOSHKA MODE 🪆
+
Russian Nesting Doll Steganography
+
+ ''', sanitize=False)
+
+ with ui.row().classes('w-full gap-6'):
+ # ==================== ENCODE SECTION ====================
+ with ui.column().classes('w-1/2 gap-3'):
+ ui.label('🔐 NESTED ENCODE').classes('text-pink-400 text-lg font-bold')
+
+ ui.html('Add carrier images (outermost first). Each layer wraps the previous.
', sanitize=False)
+
+ # Carrier list with capacity info
+ carrier_container = ui.column().classes('w-full gap-1 p-3').style(
+ 'background: rgba(255,0,255,0.05); border: 1px solid rgba(255,0,255,0.3); border-radius: 8px; min-height: 120px;'
+ )
+
+ def refresh_carriers():
+ carrier_container.clear()
+ with carrier_container:
+ if not state.matryoshka_carriers:
+ ui.html('No carriers added yet
', sanitize=False)
+ else:
+ config = create_config(channels='RGBA', bits=2)
+ for i, (img, name) in enumerate(state.matryoshka_carriers):
+ cap = _calculate_capacity_bytes(img, config)
+ layer_num = len(state.matryoshka_carriers) - i
+ with ui.row().classes('w-full items-center gap-2 p-2').style(
+ 'background: rgba(255,0,255,0.1); border-radius: 4px; margin-bottom: 4px;'
+ ):
+ ui.html(f'L{layer_num}', sanitize=False)
+ ui.label(f'{name[:25]}').classes('text-pink-300 text-sm flex-grow')
+ ui.label(f'{img.size[0]}x{img.size[1]}').classes('text-gray-500 text-xs')
+ ui.label(f'{cap//1024}KB').classes('text-pink-400 text-xs')
+
+ refresh_carriers()
+
+ # Carrier upload
+ carrier_upload_html, carrier_upload_js = create_drop_zone(
+ 'matryoshka_carrier', 'image/png,image/jpeg,image/webp',
+ '/api/matryoshka/add_carrier', 'matryoshka'
+ )
+ ui.html(carrier_upload_html, sanitize=False)
+ ui.add_body_html(carrier_upload_js)
+
+ with ui.row().classes('w-full gap-2'):
+ def clear_carriers():
+ state.matryoshka_carriers = []
+ refresh_carriers()
+ ui.notify('🪆 Carriers cleared', type='info')
+
+ ui.button('🗑️ Clear Carriers', on_click=clear_carriers).props('flat dense').classes('text-pink-400')
+
+ ui.separator().classes('my-2')
+
+ # Payload info
+ ui.html('PAYLOAD
', sanitize=False)
+ with ui.row().classes('w-full items-center gap-2 p-2').style('background: rgba(0,0,0,0.3); border-radius: 4px;'):
+ if state.encode_file_data:
+ ui.icon('description').classes('text-pink-400')
+ ui.label(f'{state.encode_file_name or "file"} ({len(state.encode_file_data)} bytes)').classes('text-pink-300 text-sm')
+ else:
+ ui.icon('text_fields').classes('text-gray-500')
+ ui.label('Set payload in ENCODE tab first').classes('text-gray-500 text-sm')
+
+ # Encode button
+ async def do_encode():
+ if not state.matryoshka_carriers:
+ ui.notify('Add carrier images first!', type='warning')
+ return
+ if not state.encode_file_data and not (hasattr(state, 'encode_text') and state.encode_text):
+ ui.notify('Set payload in ENCODE tab first', type='warning')
+ return
+
+ ui.notify('🪆 Encoding nested layers...', type='info')
+ ui.run_javascript('''
+ fetch('/api/matryoshka/encode', {
+ method: 'POST',
+ headers: {'Content-Type': 'application/json'},
+ body: JSON.stringify({channels: 'RGBA', bits: 2})
+ })
+ .then(r => r.json())
+ .then(data => {
+ if (data.success) {
+ window.location.href = '/api/matryoshka/download';
+ } else {
+ alert('Error: ' + data.error);
+ }
+ });
+ ''')
+
+ ui.button('🪆 ENCODE MATRYOSHKA', on_click=do_encode).props('unelevated').classes(
+ 'w-full mt-4 bg-pink-900 text-pink-200'
+ ).style('font-size: 14px; font-weight: bold;')
+
+ # ==================== DECODE SECTION ====================
+ with ui.column().classes('w-1/2 gap-3'):
+ ui.label('🔓 NESTED DECODE').classes('text-pink-400 text-lg font-bold')
+
+ ui.html('Upload an image and recursively extract all nested layers.
', sanitize=False)
+
+ # Depth control
+ with ui.row().classes('w-full items-center gap-3'):
+ ui.label('Max Depth:').classes('text-pink-400')
+ depth_slider = ui.slider(min=1, max=11, value=state.matryoshka_depth, step=1).classes('flex-grow').props('color="pink"')
+ depth_label = ui.label(f'{state.matryoshka_depth}').classes('text-pink-300 font-bold text-lg')
+
+ def update_depth(e):
+ state.matryoshka_depth = int(depth_slider.value)
+ depth_label.set_text(str(state.matryoshka_depth))
+
+ depth_slider.on('update:model-value', update_depth)
+
+ # Image upload for decode
+ decode_upload_html, decode_upload_js = create_drop_zone(
+ 'matryoshka_decode', 'image/png',
+ '/api/upload/decode', 'matryoshka'
+ )
+ ui.html(decode_upload_html, sanitize=False)
+ ui.add_body_html(decode_upload_js)
+
+ # Current decode image preview
+ if state.decode_image:
+ b64 = image_to_base64(state.decode_image)
+ ui.image(f'data:image/png;base64,{b64}').classes('w-full max-h-48 object-contain mt-2')
+
+ # Decode button
+ async def do_decode():
+ if not state.decode_image:
+ ui.notify('Upload an image first', type='warning')
+ return
+
+ ui.notify('🪆 Decoding nested layers...', type='info')
+ ui.run_javascript(f'''
+ fetch('/api/matryoshka/decode', {{
+ method: 'POST',
+ headers: {{'Content-Type': 'application/json'}},
+ body: JSON.stringify({{depth: {state.matryoshka_depth}}})
+ }})
+ .then(r => r.json())
+ .then(data => {{
+ if (data.success) {{
+ alert('🪆 Found ' + data.layers_found + ' nested layers!');
+ window.location.reload();
+ }} else {{
+ alert('Error: ' + data.error);
+ }}
+ }});
+ ''')
+
+ ui.button('🪆 DECODE MATRYOSHKA', on_click=do_decode).props('unelevated').classes(
+ 'w-full mt-4 bg-pink-900 text-pink-200'
+ ).style('font-size: 14px; font-weight: bold;')
+
+ # Results display
+ if state.matryoshka_results:
+ ui.separator().classes('my-3')
+ ui.label('📦 EXTRACTED LAYERS').classes('text-pink-400 font-bold')
+
+ for result in state.matryoshka_results:
+ with ui.card().classes('w-full p-2 mt-2').style('background: rgba(255,0,255,0.1); border: 1px solid rgba(255,0,255,0.3);'):
+ depth = result.get('depth', 0)
+ rtype = result.get('type', 'unknown')
+ preview = result.get('preview', '')[:200]
+
+ ui.html(f'''
+ Layer {depth}: {rtype}
+ {preview}
+ ''', sanitize=False)
+
+
+async def create_analyze_panel():
+ """Create the analysis interface"""
+
+ global state
+
+ with ui.row().classes('w-full gap-4'):
+ # Left column - Image upload
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('🔍 IMAGE TO ANALYZE').classes('text-yellow-400 text-lg font-bold')
+
+ # Show current image if loaded
+ if state.analyze_image:
+ b64 = image_to_base64(state.analyze_image)
+ ui.image(f'data:image/png;base64,{b64}').classes('w-full max-h-64 object-contain')
+ w, h = state.analyze_image.size
+ ui.label(f'Loaded | {w}x{h}').classes('text-green-400 text-sm')
+ else:
+ ui.image().classes('w-full max-h-64 object-contain')
+ ui.label('No image loaded').classes('text-gray-500 text-sm')
+
+ # Custom file picker - uses API endpoint
+ analyze_html, analyze_js = create_file_picker('analyze_image', '.png,.PNG,image/png', '/api/upload/analyze', tab='analyze')
+ ui.html(analyze_html, sanitize=False)
+ ui.add_body_html(analyze_js)
+
+ # Right column - Results
+ with ui.column().classes('w-1/2 gap-2'):
+ ui.label('📊 ANALYSIS RESULTS').classes('text-yellow-400 text-lg font-bold')
+
+ analysis_output = ui.html('', sanitize=False).classes('w-full terminal p-4')
+
+ ui.separator().classes('my-4')
+
+ with ui.row().classes('w-full justify-center'):
+ async def do_analyze():
+ if state.analyze_image is None:
+ ui.notify('Please upload an image first', type='negative')
+ return
+
+ try:
+ analysis = analyze_image(state.analyze_image)
+ detection = detect_encoding(state.analyze_image)
+
+ # Format output
+ html = ''
+ html += '╔══════════════════════════════════════╗\n'
+ html += '║ STEGANALYSIS REPORT ║\n'
+ html += '╠══════════════════════════════════════╣\n'
+
+ html += f'║ Dimensions: {analysis["dimensions"]["width"]}x{analysis["dimensions"]["height"]}\n'
+ html += f'║ Total Pixels: {analysis["total_pixels"]:,}\n'
+ html += f'║ Format: {analysis["format"] or "PNG"}\n'
+ html += '╠══════════════════════════════════════╣\n'
+
+ # Detection result
+ if detection:
+ html += '║ ✓ STEG HEADER DETECTED\n'
+ html += f'║ Channels: {detection["config"]["channels"]}\n'
+ html += f'║ Bits: {detection["config"]["bits_per_channel"]}\n'
+ html += f'║ Strategy: {detection["config"]["strategy"]}\n'
+ html += f'║ Payload: {format_size(detection["original_length"])}\n'
+ else:
+ html += '║ No STEG header detected\n'
+
+ html += '╠══════════════════════════════════════╣\n'
+
+ # Statistical analysis
+ det = analysis['detection']
+ if det['level'] == 'HIGH':
+ color = '#ff4444'
+ elif det['level'] == 'MEDIUM':
+ color = '#ffff00'
+ else:
+ color = '#00ff00'
+
+ html += f'║ Detection Level: {det["level"]}\n'
+ html += f'║ Confidence: {det["confidence"]*100:.1f}%\n'
+ html += f'║ {det["recommendation"]}\n'
+
+ html += '╠══════════════════════════════════════╣\n'
+ html += '║ CAPACITY BY CONFIG:\n'
+ for cfg, cap in analysis['capacity_by_config'].items():
+ html += f'║ {cfg}: {cap}\n'
+
+ html += '╠══════════════════════════════════════╣\n'
+ html += '║ CHANNEL ANALYSIS:\n'
+ for ch_name, ch_data in analysis['channels'].items():
+ chi = ch_data['chi_square_indicator']
+ chi_color = '#ff4444' if chi > 0.5 else '#ffff00' if chi > 0.2 else '#00ff00'
+ html += f'║ {ch_name}: μ={ch_data["mean"]:.1f} σ={ch_data["std"]:.1f} '
+ html += f'χ²={chi:.3f}\n'
+
+ html += '╚══════════════════════════════════════╝'
+ html += ''
+
+ analysis_output.set_content(html)
+ ui.notify('Analysis complete!', type='positive')
+
+ except Exception as e:
+ ui.notify(f'Analysis failed: {str(e)}', type='negative')
+
+ ui.button('🔬 ANALYZE', on_click=do_analyze).classes(
+ 'bg-yellow-600 hover:bg-yellow-500 text-black font-bold px-8 py-2 text-lg'
+ )
+
+
+# ============== MAIN ==============
+
+def main():
+ """Run the web UI"""
+ print(STEGO_ASCII)
+ print("\n🦕 Starting STEGOSAURUS WRECKS Web UI...")
+ print(" Open http://localhost:8080 in your browser\n")
+
+ ui.run(
+ title='STEGOSAURUS WRECKS',
+ favicon='🦕',
+ dark=True,
+ port=8080,
+ reload=False,
+ )
+
+
+if __name__ in {"__main__", "__mp_main__"}:
+ main()
diff --git a/wrangler.jsonc b/wrangler.jsonc
new file mode 100644
index 0000000..bc41213
--- /dev/null
+++ b/wrangler.jsonc
@@ -0,0 +1,10 @@
+{
+ "name": "st3gg",
+ "compatibility_date": "2026-03-30",
+ "assets": {
+ "directory": "./"
+ },
+ // Only index.html and f5stego-lib.js need to be served
+ // All other files (Python, examples, etc.) are development-only
+ "rules": []
+}