Add MCP server for AI agent integration

Expose 13 steganography tools via Model Context Protocol (stdio):
- stegg_encode/decode: LSB steg with 15 channel presets, 4 strategies, AES-256-GCM
- stegg_analyze: chi-square anomaly detection with verdict scoring
- stegg_detect: STEG v3 header auto-detection
- stegg_capacity: carrier capacity calculation
- stegg_inject_chunk/read_chunks: PNG metadata chunk injection and reading
- stegg_inject_exif: EXIF/PNG metadata field injection via PIL
- stegg_injection_filename: prompt-injection filename generation for AI red-teaming
- stegg_jailbreak_templates: list available jailbreak prompt templates
- stegg_analysis_tool/list_analysis_tools: access to 264+ detection functions
- stegg_crypto_status: encryption method availability check

Install with: pip install stegg[mcp]
Run with: stegg-mcp (or python mcp_server.py)
This commit is contained in:
GangGreenTemperTatum 2026-06-15 08:25:42 -04:00
parent 7db0938950
commit c337ab7095
No known key found for this signature in database
GPG Key ID: 97A6E04DE3B622D3
2 changed files with 684 additions and 3 deletions

678
mcp_server.py Normal file
View File

@ -0,0 +1,678 @@
#!/usr/bin/env python3
"""
ST3GG MCP Server Steganography toolkit for AI agents.
Exposes encode, decode, analyze, inject, and detection capabilities
via the Model Context Protocol (stdio transport).
"""
import base64
import json
import os
import sys
import tempfile
from pathlib import Path
from typing import Any, Optional
from mcp.server.fastmcp import FastMCP
# ---------------------------------------------------------------------------
# Bootstrap: add repo root to sys.path so local modules resolve
# ---------------------------------------------------------------------------
_REPO_ROOT = Path(__file__).resolve().parent
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
from PIL import Image
from steg_core import (
encode,
decode,
create_config,
calculate_capacity,
analyze_image,
detect_encoding,
CHANNEL_PRESETS,
)
from analysis_tools import (
execute_action,
list_available_tools,
detect_file_type,
png_full_analysis,
)
from injector import (
generate_injection_filename,
get_template_names,
get_jailbreak_template,
get_jailbreak_names,
inject_text_chunk,
inject_itxt_chunk,
inject_private_chunk,
read_png_chunks,
extract_text_chunks,
inject_metadata_pil,
)
try:
from crypto import encrypt, decrypt, get_available_methods, crypto_status
except Exception:
encrypt = decrypt = None
def get_available_methods():
return ["none", "xor"]
def crypto_status():
return {"cryptography_available": False, "available_methods": ["xor"]}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _load_image(image_path: str) -> Image.Image:
"""Load an image from disk, raising a clear error on failure."""
p = Path(image_path).expanduser().resolve()
if not p.exists():
raise FileNotFoundError(f"Image not found: {p}")
return Image.open(p)
def _resolve_output(output_path: Optional[str], input_path: str, suffix: str = "_steg") -> Path:
"""Determine output path, defaulting to <input>_steg.png next to input."""
if output_path:
return Path(output_path).expanduser().resolve()
inp = Path(input_path).expanduser().resolve()
return inp.parent / f"{inp.stem}{suffix}.png"
# ---------------------------------------------------------------------------
# MCP Server
# ---------------------------------------------------------------------------
mcp = FastMCP(
"stegg",
instructions=(
"ST3GG steganography toolkit. Encode/decode hidden data in images, "
"analyze files for steganographic content, inject metadata/chunks, "
"and generate prompt-injection filenames for AI red-teaming."
),
)
# ---- Encode ---------------------------------------------------------------
@mcp.tool()
def stegg_encode(
image_path: str,
payload_text: str = "",
payload_file: str = "",
output_path: str = "",
channels: str = "RGB",
bits_per_channel: int = 1,
strategy: str = "interleaved",
seed: int = 0,
password: str = "",
compress: bool = True,
) -> str:
"""Hide data inside an image using LSB steganography.
Supports 15 channel presets (R, G, B, A, RG, RB, ... RGBA),
1-8 bits per channel, 4 embedding strategies, and optional
AES-256-GCM encryption.
Args:
image_path: Path to carrier image (PNG recommended).
payload_text: Text message to hide (mutually exclusive with payload_file).
payload_file: Path to file whose bytes to hide.
output_path: Where to write the stegged image. Defaults to <input>_steg.png.
channels: Channel preset one of R, G, B, A, RG, RB, RA, GB, GA, BA,
RGB, RGA, RBA, GBA, RGBA.
bits_per_channel: Bits to use per channel (1-8). Higher = more capacity,
more visual distortion.
strategy: Embedding strategy sequential, interleaved, spread, randomized.
seed: Random seed for the randomized strategy (0 = auto).
password: Optional encryption password (AES-256-GCM if available, else XOR).
compress: Whether to zlib-compress the payload before encoding.
Returns:
JSON with output_path, payload_bytes, capacity info, and encryption status.
"""
image = _load_image(image_path)
# Resolve payload
if payload_file:
p = Path(payload_file).expanduser().resolve()
if not p.exists():
return json.dumps({"error": f"Payload file not found: {p}"})
payload = p.read_bytes()
elif payload_text:
payload = payload_text.encode("utf-8")
else:
return json.dumps({"error": "Provide payload_text or payload_file"})
out = _resolve_output(output_path or "", image_path)
config = create_config(
channels=channels,
bits=bits_per_channel,
compress=compress,
strategy=strategy,
seed=seed if seed else None,
)
capacity = calculate_capacity(image, config)
if len(payload) > capacity["usable_bytes"]:
return json.dumps({
"error": f"Payload too large: {len(payload)} bytes > {capacity['usable_bytes']} available",
"capacity": capacity["human"],
})
encrypted = False
if password and encrypt:
payload = encrypt(payload, password)
encrypted = True
encode(image, payload, config, str(out))
return json.dumps({
"output_path": str(out),
"payload_bytes": len(payload),
"capacity": capacity["human"],
"channels": channels,
"bits_per_channel": bits_per_channel,
"strategy": strategy,
"encrypted": encrypted,
"compressed": compress,
})
# ---- Decode ---------------------------------------------------------------
@mcp.tool()
def stegg_decode(
image_path: str,
output_path: str = "",
auto_detect: bool = True,
channels: str = "RGB",
bits_per_channel: int = 1,
strategy: str = "interleaved",
seed: int = 0,
password: str = "",
) -> str:
"""Extract hidden data from a steganographic image.
By default auto-detects the encoding config from the STEG header.
Falls back to manual config if no header is found.
Args:
image_path: Path to the encoded image.
output_path: Optional path to write extracted binary data.
auto_detect: Try to detect encoding config from header (default True).
channels: Channel preset if not auto-detecting.
bits_per_channel: Bits per channel if not auto-detecting.
strategy: Strategy if not auto-detecting.
seed: Seed if not auto-detecting.
password: Decryption password if the payload was encrypted.
Returns:
JSON with extracted text (UTF-8) or hex preview for binary data,
plus byte count, config detected, and output_path if saved.
"""
image = _load_image(image_path)
config = None
detected = False
if auto_detect:
detection = detect_encoding(image)
if detection:
detected = True
config = None # let decode() use header
else:
config = create_config(
channels=channels,
bits=bits_per_channel,
strategy=strategy,
seed=seed if seed else None,
)
else:
config = create_config(
channels=channels,
bits=bits_per_channel,
strategy=strategy,
seed=seed if seed else None,
)
data = decode(image, config)
if password and decrypt:
data = decrypt(data, password)
result: dict[str, Any] = {
"bytes": len(data),
"auto_detected": detected,
}
if output_path:
out = Path(output_path).expanduser().resolve()
out.write_bytes(data)
result["output_path"] = str(out)
# Try UTF-8
try:
text = data.decode("utf-8")
result["text"] = text
result["encoding"] = "utf-8"
except UnicodeDecodeError:
result["hex_preview"] = data[:512].hex()
result["encoding"] = "binary"
return json.dumps(result)
# ---- Analyze --------------------------------------------------------------
@mcp.tool()
def stegg_analyze(
image_path: str,
full: bool = False,
) -> str:
"""Analyze an image for steganographic indicators.
Runs chi-square analysis on each channel's LSB distribution,
calculates capacity estimates, and optionally runs the full
264-function analysis suite.
Args:
image_path: Path to the image to analyze.
full: Run the full analysis suite (PNG only, more detailed).
Returns:
JSON with channel stats, anomaly indicators, capacity estimates,
and a verdict (normal / possible / high probability).
"""
image = _load_image(image_path)
analysis = analyze_image(image)
# Compact summary
channels_summary = {}
max_indicator = 0.0
for ch_name, ch_data in analysis["channels"].items():
lsb = ch_data["lsb_ratio"]
indicator = ch_data.get("chi_square_indicator", 0.0)
max_indicator = max(max_indicator, indicator)
channels_summary[ch_name] = {
"mean": round(ch_data["mean"], 2),
"std": round(ch_data["std"], 2),
"lsb_zeros_pct": round(lsb["zeros"] * 100, 1),
"lsb_ones_pct": round(lsb["ones"] * 100, 1),
"chi_square": round(ch_data.get("chi_square", 0.0), 4),
"chi_square_indicator": round(indicator, 4),
"anomaly": "HIGH" if indicator > 0.3 else ("slight" if indicator > 0.1 else "normal"),
}
if max_indicator > 0.3:
verdict = "HIGH PROBABILITY of hidden data"
elif max_indicator > 0.1:
verdict = "Possible hidden data (slight anomaly)"
else:
verdict = "No obvious steganographic indicators"
result: dict[str, Any] = {
"dimensions": analysis["dimensions"],
"mode": analysis["mode"],
"total_pixels": analysis["total_pixels"],
"channels": channels_summary,
"capacity": analysis["capacity_by_config"],
"verdict": verdict,
}
# Optional full PNG analysis
if full:
p = Path(image_path).expanduser().resolve()
raw = p.read_bytes()
try:
full_result = png_full_analysis(raw)
if isinstance(full_result, dict):
result["full_analysis"] = full_result
except Exception as e:
result["full_analysis_error"] = str(e)
return json.dumps(result)
# ---- Capacity -------------------------------------------------------------
@mcp.tool()
def stegg_capacity(
image_path: str,
channels: str = "RGB",
bits_per_channel: int = 1,
) -> str:
"""Calculate how much data an image can hold with given settings.
Args:
image_path: Path to the carrier image.
channels: Channel preset.
bits_per_channel: Bits per channel (1-8).
Returns:
JSON with capacity in bytes and human-readable form.
"""
image = _load_image(image_path)
config = create_config(channels=channels, bits=bits_per_channel)
cap = calculate_capacity(image, config)
return json.dumps({
"usable_bytes": cap["usable_bytes"],
"human": cap["human"],
"total_pixels": image.width * image.height,
"channels": channels,
"bits_per_channel": bits_per_channel,
})
# ---- Detect ---------------------------------------------------------------
@mcp.tool()
def stegg_detect(
image_path: str,
) -> str:
"""Quick check: does this image contain a STEG v3 header?
Attempts auto-detection of the ST3GG encoding header to determine
if the image was encoded with this toolkit.
Args:
image_path: Path to the image to check.
Returns:
JSON with detected (bool) and config details if found.
"""
image = _load_image(image_path)
detection = detect_encoding(image)
if detection:
return json.dumps({"detected": True, "config": detection})
return json.dumps({"detected": False})
# ---- PNG Chunk Injection --------------------------------------------------
@mcp.tool()
def stegg_inject_chunk(
image_path: str,
output_path: str,
chunk_type: str = "tEXt",
keyword: str = "Comment",
text: str = "",
compressed: bool = False,
) -> str:
"""Inject a text chunk into a PNG image.
Useful for hiding data in metadata, prompt injection via image
metadata, or adding custom PNG chunks for red-teaming.
Args:
image_path: Path to source PNG.
output_path: Where to write the modified PNG.
chunk_type: PNG chunk type tEXt, zTXt, iTXt, or a 4-char private type.
keyword: Chunk keyword (e.g. Comment, Description, Author).
text: Text content to inject.
compressed: Use zTXt compression (only for tEXt/zTXt).
Returns:
JSON with output_path and chunk details.
"""
p = Path(image_path).expanduser().resolve()
if not p.exists():
return json.dumps({"error": f"Image not found: {p}"})
raw = p.read_bytes()
if chunk_type == "iTXt":
modified = inject_itxt_chunk(raw, keyword, text)
elif len(chunk_type) == 4 and chunk_type not in ("tEXt", "zTXt", "iTXt"):
modified = inject_private_chunk(raw, chunk_type, text.encode("utf-8"))
else:
modified = inject_text_chunk(raw, keyword, text, compressed=compressed)
out = Path(output_path).expanduser().resolve()
out.write_bytes(modified)
return json.dumps({
"output_path": str(out),
"chunk_type": chunk_type,
"keyword": keyword,
"text_length": len(text),
"compressed": compressed,
})
# ---- Read PNG Chunks ------------------------------------------------------
@mcp.tool()
def stegg_read_chunks(
image_path: str,
) -> str:
"""Read and list all chunks in a PNG image.
Extracts text chunks (tEXt, zTXt, iTXt) and lists all chunk types
with sizes. Useful for inspecting images for hidden metadata.
Args:
image_path: Path to the PNG image.
Returns:
JSON with chunk list and extracted text content.
"""
p = Path(image_path).expanduser().resolve()
if not p.exists():
return json.dumps({"error": f"Image not found: {p}"})
raw = p.read_bytes()
chunks = read_png_chunks(raw)
text_chunks = extract_text_chunks(raw)
# Compact chunk summary
chunk_summary = []
for c in chunks:
chunk_summary.append({
"type": c.get("type", "?"),
"size": c.get("length", 0),
"offset": c.get("offset", 0),
})
return json.dumps({
"chunks": chunk_summary,
"text_content": text_chunks,
"total_chunks": len(chunks),
})
# ---- EXIF Injection -------------------------------------------------------
@mcp.tool()
def stegg_inject_exif(
image_path: str,
output_path: str,
comment: str = "",
author: str = "",
description: str = "",
title: str = "",
custom_fields: str = "",
) -> str:
"""Inject EXIF/metadata fields into an image via PIL.
Args:
image_path: Path to source image.
output_path: Where to write the modified image.
comment: Image comment field.
author: Author / artist field.
description: Image description.
title: Image title.
custom_fields: JSON object of additional key-value pairs to inject
as PNG text chunks (e.g. '{"Software": "evil"}').
Returns:
JSON with output_path and injected fields.
"""
p = Path(image_path).expanduser().resolve()
if not p.exists():
return json.dumps({"error": f"Image not found: {p}"})
metadata: dict[str, str] = {}
if comment:
metadata["Comment"] = comment
if author:
metadata["Author"] = author
if description:
metadata["Description"] = description
if title:
metadata["Title"] = title
if custom_fields:
try:
extra = json.loads(custom_fields)
metadata.update(extra)
except json.JSONDecodeError:
return json.dumps({"error": "custom_fields must be valid JSON"})
if not metadata:
return json.dumps({"error": "Provide at least one metadata field"})
image = Image.open(p)
_, png_bytes = inject_metadata_pil(image, metadata)
out = Path(output_path).expanduser().resolve()
out.write_bytes(png_bytes)
return json.dumps({
"output_path": str(out),
"injected_fields": list(metadata.keys()),
"field_count": len(metadata),
})
# ---- Prompt Injection Filenames ------------------------------------------
@mcp.tool()
def stegg_injection_filename(
template: str = "universal_decoder",
channels: str = "RGB",
count: int = 1,
) -> str:
"""Generate prompt-injection filenames for AI red-teaming.
Creates filenames designed to trigger LLMs into decoding steganographic
content when processing uploaded images.
Args:
template: Template name chatgpt_decoder, claude_decoder,
gemini_decoder, universal_decoder, system_override,
roleplay_trigger, dev_mode, subtle, custom.
channels: Channel string to embed in the filename.
count: Number of filenames to generate.
Returns:
JSON with list of generated filenames and template used.
"""
filenames = [
generate_injection_filename(template, channels)
for _ in range(count)
]
return json.dumps({
"template": template,
"channels": channels,
"filenames": filenames,
})
# ---- Jailbreak Templates -------------------------------------------------
@mcp.tool()
def stegg_jailbreak_templates() -> str:
"""List available jailbreak prompt templates and their previews.
These templates can be encoded into images as hidden payloads
for AI red-teaming scenarios.
Returns:
JSON with template names and previews.
"""
templates = {}
for name in get_jailbreak_names():
content = get_jailbreak_template(name)
templates[name] = content[:120] + ("..." if len(content) > 120 else "")
return json.dumps({"templates": templates, "count": len(templates)})
# ---- Analysis Tools -------------------------------------------------------
@mcp.tool()
def stegg_analysis_tool(
file_path: str,
action: str,
) -> str:
"""Run a specific analysis tool from the 264-function analysis suite.
Use stegg_list_analysis_tools to see available actions. Each tool
returns structured results with suspicion scoring and confidence.
Args:
file_path: Path to the file to analyze.
action: Analysis action name (e.g. png_chi_square_analysis,
rs_analysis, detect_homoglyph_steg, jpeg_decode, etc.).
Returns:
JSON with analysis results including suspicious flag and confidence.
"""
p = Path(file_path).expanduser().resolve()
if not p.exists():
return json.dumps({"error": f"File not found: {p}"})
data = p.read_bytes()
result = execute_action(action, data)
if hasattr(result, "to_dict"):
return json.dumps(result.to_dict(), default=str)
return json.dumps({"result": str(result)})
@mcp.tool()
def stegg_list_analysis_tools() -> str:
"""List all available analysis tool actions.
Returns the full registry of 264+ analysis functions organized by
file type (PNG, JPEG, audio, text, archive, etc.).
Returns:
JSON with sorted list of action names.
"""
tools = list_available_tools()
return json.dumps({"tools": tools, "count": len(tools)})
# ---- Crypto Status --------------------------------------------------------
@mcp.tool()
def stegg_crypto_status() -> str:
"""Check available encryption methods.
Returns whether AES-256-GCM is available (requires cryptography package)
and lists all available encryption methods.
Returns:
JSON with crypto availability and method list.
"""
status = crypto_status()
if isinstance(status, dict):
return json.dumps(status)
return json.dumps({"status": str(status)})
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()

View File

@ -8,7 +8,7 @@ version = "3.0.0"
description = "Steganography toolkit — hide anything in any file, across every modality"
readme = "README.md"
license = {text = "AGPL-3.0-or-later"}
requires-python = ">=3.9"
requires-python = ">=3.10"
authors = [
{name = "ST3GG Contributors"}
]
@ -21,7 +21,6 @@ classifiers = [
"Intended Audience :: Science/Research",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
@ -48,6 +47,8 @@ web = ["nicegui>=1.4.0", "fastapi>=0.100.0"]
web-legacy = ["streamlit>=1.28.0"]
# AES-256-GCM encryption
crypto = ["cryptography>=41.0.0"]
# MCP server for AI agent integration
mcp = ["mcp[cli]>=1.0.0"]
# Everything
all = [
"textual>=0.40.0",
@ -55,12 +56,14 @@ all = [
"fastapi>=0.100.0",
"streamlit>=1.28.0",
"cryptography>=41.0.0",
"mcp[cli]>=1.0.0",
]
[project.scripts]
stegg = "cli:main_cli"
stegg-tui = "tui:main"
stegg-web = "webui:main"
stegg-mcp = "mcp_server:main"
[project.urls]
Homepage = "https://ste.gg"
@ -69,7 +72,7 @@ Documentation = "https://github.com/elder-plinius/st3gg#readme"
Issues = "https://github.com/elder-plinius/st3gg/issues"
[tool.setuptools]
py-modules = ["steg_core", "crypto", "analysis_tools", "cli", "tui", "webui", "app", "injector", "ascii_art"]
py-modules = ["steg_core", "crypto", "analysis_tools", "cli", "tui", "webui", "app", "injector", "ascii_art", "mcp_server"]
[tool.setuptools.package-data]
"*" = ["index.html", "f5stego-lib.js", "_headers", "wrangler.jsonc"]