OpenStego

This commit is contained in:
Zeecka
2025-12-27 14:38:01 +01:00
parent 0949996a4f
commit c816c0bcd9
8 changed files with 142 additions and 9 deletions
+115
View File
@@ -0,0 +1,115 @@
"""OpenStego Analyzer for Image Submissions."""
import shutil
import subprocess
from pathlib import Path
from typing import Optional
from .utils import MAX_PENDING_TIME, update_data
def analyze_openstego(input_img: Path, output_dir: Path, password: str = "") -> None:
"""Analyze an image submission using openstego."""
image_name = input_img.name
try:
stderr = ""
# Try to extract the embedded file
extracted_dir = output_dir / "openstego"
extracted_dir.mkdir(parents=True, exist_ok=True)
for algo in ["AES128", "AES256"]:
cmd: list[str] = [
"openstego",
"extract",
"-a",
"randomlsb",
"--cryptalgo",
algo,
"-sf",
"../" + str(image_name),
"-xd",
str(extracted_dir),
"-p",
password,
]
data = subprocess.run(
cmd,
cwd=output_dir,
capture_output=True,
text=True,
check=False,
timeout=MAX_PENDING_TIME,
)
# Check if extraction was successful
if data.returncode != 0:
stderr += data.stderr.replace(
f'"../{image_name}" ', ""
) # hide file name
err = {
"openstego": {
"status": "error",
"error": stderr,
}
}
update_data(output_dir, err)
return None
# Find the extracted file
extracted_files = list(extracted_dir.glob("*"))
if not extracted_files:
err = {
"openstego": {
"status": "error",
"error": "No file extracted, password may be incorrect.",
}
}
update_data(output_dir, err)
return None
# Filter stdout for success messages
stdout = []
for f in extracted_files:
stdout.append(f"Recovered file: {f.name}")
# Zip extracted files
zip_data = subprocess.run(
["7z", "a", "../openstego.7z", "*"],
cwd=extracted_dir,
capture_output=True,
text=True,
check=False,
timeout=MAX_PENDING_TIME,
)
stderr += zip_data.stderr
if len(stderr) > 0:
err = {
"openstego": {
"status": "error",
"error": stderr,
}
}
update_data(output_dir, err)
return None
# Remove the extracted directory
shutil.rmtree(extracted_dir)
update_data(
output_dir,
{
"openstego": {
"status": "ok",
"output": stdout,
"download": f"/download/{output_dir.name}/openstego",
}
},
)
except Exception as e:
update_data(output_dir, {"openstego": {"status": "error", "error": str(e)}})
return None
+1 -1
View File
@@ -3,6 +3,6 @@
from pathlib import Path
IMAGE_EXTENSIONS = [".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".tiff"]
WORKER_FILES = ["binwalk", "foremost", "steghide", "zsteg", "image_resize"]
WORKER_FILES = ["binwalk", "foremost", "steghide", "zsteg", "image_resize", "openstego"]
RESULT_FOLDER = Path(__file__).parent.resolve() / "results"
+2 -1
View File
@@ -6,6 +6,7 @@ const TOOL_ORDER = [
"outguess",
"pngcheck",
"steghide",
"openstego",
"zsteg",
"strings",
"image_resize",
@@ -286,7 +287,7 @@ function parseResult(result) {
}
// Parse text output
if (typeof result[tool]["output"] === "string") {
analyzer.innerHTML += `<pre>${escapeHtml(
+2
View File
@@ -9,6 +9,7 @@ from .analyzers.decomposer import analyze_decomposer
from .analyzers.exiftool import analyze_exiftool
from .analyzers.foremost import analyze_foremost
from .analyzers.image_resize import analyze_image_resize
from .analyzers.openstego import analyze_openstego
from .analyzers.outguess import analyze_outguess
from .analyzers.pngcheck import analyze_pngcheck
from .analyzers.steghide import analyze_steghide
@@ -77,6 +78,7 @@ def analyze_image(submission_hash: str) -> None:
(analyze_strings, img_path, result_path),
(analyze_pngcheck, img_path, result_path),
(analyze_steghide, img_path, result_path, submission.password),
(analyze_openstego, img_path, result_path, submission.password),
(analyze_zsteg, img_path, result_path),
(analyze_image_resize, img_path, result_path),
]