From 78f8d867ea8f72486709286645aedc0f04a585ec Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 1 Jan 2026 23:31:42 +0100 Subject: [PATCH] Update analysers (WIP) --- aperisolve/analyzers/binwalk.py | 84 +++--------- aperisolve/analyzers/exiftool.py | 57 +++----- aperisolve/analyzers/foremost.py | 72 +++------- aperisolve/analyzers/outguess.py | 103 +++------------ aperisolve/analyzers/pngcheck.py | 59 ++------- aperisolve/analyzers/steghide.py | 154 +++++++--------------- aperisolve/analyzers/strings.py | 47 ++----- aperisolve/analyzers/template_analyzer.py | 85 +++--------- aperisolve/analyzers/zsteg.py | 54 +++----- 9 files changed, 189 insertions(+), 526 deletions(-) diff --git a/aperisolve/analyzers/binwalk.py b/aperisolve/analyzers/binwalk.py index 9b8cfac..eb04845 100644 --- a/aperisolve/analyzers/binwalk.py +++ b/aperisolve/analyzers/binwalk.py @@ -3,75 +3,27 @@ # mypy: disable-error-code=unused-awaitable """Binwalk Analyzer for Image Submissions.""" -import shutil -import subprocess from pathlib import Path -from typing import Any -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class BinwalkAnalyzer(SubprocessAnalyzer): + """Analyzer for binwalk.""" + + def __init__(self, *args): + super().__init__("binwalk", *args, has_archive=True) + self.cmd = ["binwalk", "--matryoshka", "-e", self.img, "--run-as=root"] + self.make_folder = False # Binwalk already create its own folder + + def get_extracted_dir(self) -> Path: + return self.output_dir / f"_{self.input_img.name}.extracted" + + def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + return len(stderr) > 0 and not zip_exist and False def analyze_binwalk(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using binwalk.""" - - image_name = input_img.name - extracted_dir = output_dir / f"_{image_name}.extracted" - - try: - stderr = "" - # Run binwalk - data = subprocess.run( - ["binwalk", "--matryoshka", "-e", "../" + str(image_name), "--run-as=root"], - cwd=output_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - stderr += data.stderr - - zip_exist = False - if extracted_dir.exists(): - # Zip extracted files - zip_data = subprocess.run( - ["7z", "a", "../binwalk.7z", "*"], - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - zip_exist = True - stderr += zip_data.stderr - - # Remove the extracted directory - if extracted_dir.exists(): - shutil.rmtree(extracted_dir) - - # Only report error if stderr exists AND extraction failed. - # If zip_exist is True, binwalk worked (despite warnings), so we proceed. - if len(stderr) > 0 and not zip_exist: - err: dict[str, dict[str, Any]] = { - "binwalk": { - "status": "error", - "output": data.stdout.split("\n") if data else [], - "error": stderr, - } - } - update_data(output_dir, err) - return None - - output_data: dict[str, dict[str, Any]] = { - "binwalk": { - "status": "ok", - "output": data.stdout.split("\n") if data else [], - } - } - if zip_exist: - output_data["binwalk"]["download"] = f"/download/{output_dir.name}/binwalk" - - update_data(output_dir, output_data) - - except Exception as e: - update_data(output_dir, {"binwalk": {"status": "error", "error": str(e)}}) - return None + analyzer = BinwalkAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/exiftool.py b/aperisolve/analyzers/exiftool.py index 5ed107d..ae42bae 100644 --- a/aperisolve/analyzers/exiftool.py +++ b/aperisolve/analyzers/exiftool.py @@ -3,46 +3,29 @@ # mypy: disable-error-code=unused-awaitable """Exiftool Analyzer for Image Submissions.""" -import subprocess from pathlib import Path -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class ExiftoolAnalyzer(SubprocessAnalyzer): + """Analyzer for exiftool.""" + + def __init__(self, *args): + super().__init__("exiftool", *args) + self.cmd = ["exiftool", "-a", "-u", "-g1", self.img] + + def process_output(self, stdout: str, stderr: str) -> dict[str, str]: + """Process the stdout into a list of lines.""" + metadata: dict[str, str] = {} + for line in stdout.split("\n"): + if ":" in line: + key, value = line.split(":", 1) + metadata[key.strip()] = value.strip() + return metadata def analyze_exiftool(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using exiftool.""" - - try: - data = subprocess.run( - ["exiftool", input_img], - capture_output=True, - text=True, - check=True, - timeout=MAX_PENDING_TIME, - ) - - metadata: dict[str, str] = {} - for line in data.stdout.split("\n"): - if ":" in line: - key, value = line.split(":", 1) - metadata[key.strip()] = value.strip() - - if data.stderr: - update_data( - output_dir, - { - "exiftool": { - "status": "error", - "output": metadata, - "error": data.stderr, - } - }, - ) - return - - update_data( - output_dir, - {"exiftool": {"status": "ok", "output": metadata}}, - ) - except Exception as e: - update_data(output_dir, {"exiftool": {"status": "error", "error": str(e)}}) + analyzer = ExiftoolAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/foremost.py b/aperisolve/analyzers/foremost.py index eb0fc8c..45f96f6 100644 --- a/aperisolve/analyzers/foremost.py +++ b/aperisolve/analyzers/foremost.py @@ -3,60 +3,30 @@ # mypy: disable-error-code=unused-awaitable """Foremost Analyzer for Image Submissions.""" -import shutil -import subprocess from pathlib import Path -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class ForemostAnalyzer(SubprocessAnalyzer): + """Analyzer for foremost.""" + + def __init__(self, *args): + super().__init__("foremost", *args, has_archive=True) + self.cmd = ["foremost", "-o", str(self.get_extracted_dir()), "-i", self.img] + + def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + """Check if the result is an error.""" + return len(stderr) > 60 + + def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]: + """Process the stdout/stderr.""" + if "Processing" in stderr and "|*|" in stderr: + return stderr.strip().split("\n") + return [] def analyze_foremost(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using foremost.""" - - try: - foremost_dir = output_dir / "foremost" - foremost_dir.mkdir(parents=True, exist_ok=True) - - stderr = "" - # Run foremost - data = subprocess.run( - ["foremost", "-o", foremost_dir, "-i", input_img], - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - # Note: foremost use stderr as standard output for results :o) - # So we can't use it as error detection : - # stderr += data.stderr - - # Zip extracted files - zip_data = subprocess.run( - ["7z", "a", "../foremost.7z", "*"], - cwd=foremost_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - - stderr += zip_data.stderr - - # Remove the extracted directory - shutil.rmtree(foremost_dir) - - data_strings = data.stdout.split("\n") if data else [] - data_strings = [s for s in data_strings if s] - - update_data( - output_dir, - { - "foremost": { - "status": "ok", - "output": data_strings, - "download": f"/download/{output_dir.name}/foremost", - } - }, - ) - except Exception as e: - update_data(output_dir, {"foremost": {"status": "error", "error": str(e)}}) + analyzer = ForemostAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/outguess.py b/aperisolve/analyzers/outguess.py index 9fc6a78..738879e 100644 --- a/aperisolve/analyzers/outguess.py +++ b/aperisolve/analyzers/outguess.py @@ -3,97 +3,32 @@ # mypy: disable-error-code=unused-awaitable """Outguess Analyzer for Image Submissions.""" -import shutil -import subprocess from pathlib import Path from typing import Optional -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer -def analyze_outguess( - input_img: Path, output_dir: Path, password: Optional[str] = None -) -> None: - """Analyze an image submission using outguess.""" +class OutguessAnalyzer(SubprocessAnalyzer): + """Analyzer for outguess.""" - image_name = input_img.name - try: - outguess_dir = output_dir / "outguess" - outguess_dir.mkdir(parents=True, exist_ok=True) - outguess_file = "outguess.data" - stderr = "" - cmd: list[str] + def __init__(self, *args): + super().__init__("outguess", *args, has_archive=True) + + def build_cmd(self, password: Optional[str] = None) -> list[str]: + extracted_dir = self.get_extracted_dir() + assert extracted_dir is not None # since has_archive is True + out = str(extracted_dir / "outguess.data") if password: - cmd = [ - "outguess", - "-k", - password, - "-r", - "../../" + str(image_name), - outguess_file, - ] + return ["outguess", "-k", password, "-r", self.img, out] else: - cmd = ["outguess", "-r", "../../" + str(image_name), outguess_file] + return ["outguess", "-r", self.img, out] - data = subprocess.run( - cmd, - cwd=outguess_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - if data.returncode != 0: - stderr += data.stderr - err = { - "outguess": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - # Zip extracted files - zip_data = subprocess.run( - ["7z", "a", "../outguess.7z", "*"], - cwd=outguess_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - - stderr += zip_data.stderr - - if len(stderr) > 0: - err = { - "outguess": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - # Remove the extracted directory - shutil.rmtree(outguess_dir) - - data_strings = data.stderr.split("\n") if data else [] - data_strings = [s for s in data_strings if s] - - update_data( - output_dir, - { - "outguess": { - "status": "ok", - "output": data_strings, - "download": f"/download/{output_dir.name}/outguess", - } - }, - ) - - except Exception as e: - update_data(output_dir, {"outguess": {"status": "error", "error": str(e)}}) - return None +def analyze_outguess(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: + """Analyze an image submission using outguess.""" + analyzer = OutguessAnalyzer(input_img, output_dir) + if password: + analyzer.analyze(password) + else: + analyzer.analyze() diff --git a/aperisolve/analyzers/pngcheck.py b/aperisolve/analyzers/pngcheck.py index 730e67c..36ee677 100644 --- a/aperisolve/analyzers/pngcheck.py +++ b/aperisolve/analyzers/pngcheck.py @@ -3,54 +3,23 @@ # mypy: disable-error-code=unused-awaitable """Pngcheck Analyzer for Image Submissions.""" -import subprocess from pathlib import Path -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class PngcheckAnalyzer(SubprocessAnalyzer): + """Analyzer for pngcheck.""" + + def __init__(self, *args): + super().__init__("pngcheck", *args) + self.cmd = ["pngcheck", "-v", self.img] + + def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + return "this is neither a PNG or JNG image nor a MNG stream" in stdout def analyze_pngcheck(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using pngcheck.""" - - try: - data = subprocess.run( - ["pngcheck", "-v", input_img], - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - # Error based on https://github.com/pnggroup/pngcheck/issues/59 - # if data.stderr: - # update_data( - # output_dir, {"pngcheck": {"status": "error", "error": data.stderr}} - # ) - # return - - if "this is neither a PNG or JNG image nor a MNG stream" in data.stdout: - update_data( - output_dir, - { - "pngcheck": { - "status": "error", - "error": "ERRORS DETECTED: this is neither a PNG or JNG image nor a " - "MNG stream.", - } - }, - ) - return - - data_pngcheck = data.stdout.split("\n") if data else [] - data_pngcheck = [s for s in data_pngcheck if s] - - update_data( - output_dir, - { - "pngcheck": { - "status": "ok", - "output": data_pngcheck, - } - }, - ) - except Exception as e: - update_data(output_dir, {"pngcheck": {"status": "error", "error": str(e)}}) + analyzer = PngcheckAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/steghide.py b/aperisolve/analyzers/steghide.py index 0602bd6..0454e13 100644 --- a/aperisolve/analyzers/steghide.py +++ b/aperisolve/analyzers/steghide.py @@ -4,127 +4,67 @@ """Steghide Analyzer for Image Submissions.""" import re -import shutil -import subprocess from pathlib import Path from typing import Optional -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer -def analyze_steghide( - input_img: Path, output_dir: Path, password: Optional[str] = None -) -> None: - """Analyze an image submission using steghide.""" +class SteghideAnalyzer(SubprocessAnalyzer): + """Analyzer for steghide.""" - image_name = input_img.name - try: - stderr = "" - # Get embedded file information first - cmd: list[str] = ["steghide", "info", "../" + str(image_name)] - if password: - cmd += ["-p", password] - else: - cmd += ["-p", ""] + def __init__(self, *args): + super().__init__("steghide", *args, has_archive=True) - data = subprocess.run( - cmd, - cwd=output_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) + def build_cmd(self, password: Optional[str] = None) -> list[str]: + if password is None: + password = "" + + # First, get the embedded file name, and return the test cmd if failed + cmd_test = ["steghide", "info", self.img, "-p", password] + data = self.run_command(cmd_test, cwd=self.output_dir) + err = self.is_error(data.returncode, data.stdout, data.stderr, False) + if err: + return cmd_test match = re.search(r'embedded file "([^"]+)"', data.stdout) + assert match is not None + hidden = match.group(1) + cmd = ["steghide", "extract", "-sf", self.img, "-xf", hidden, "-p", password] + return cmd + + def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + """Check if the result is an error.""" + + match = re.search(r'embedded file "([^"]+)"', stdout) embedded_filename = match.group(1) if match else None # if bad file format or wrong password, raise an error - if data.returncode != 0 or embedded_filename is None: - stderr += data.stderr.replace(f'"../{image_name}" ', "") # hide file name - # the file format of the file \"../foobar.png\" is not supported. - # become - # the file format of the file is not supported. - err = { - "steghide": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None + return returncode != 0 or embedded_filename is None - # The passphrase is correct, we can extract the file - # prevent path traversal - embedded_file = output_dir / "steghide" / str(Path(embedded_filename).name) - extracted_dir = embedded_file.parent - extracted_dir.mkdir(parents=True, exist_ok=True) - - cmd_ext: list[str] = [ - "steghide", - "extract", - "-sf", - "../../" + str(image_name), - "-xf", - str(embedded_file.name), - ] - if password: - cmd_ext += ["-p", password] - else: - cmd_ext += ["-p", ""] - - data = subprocess.run( - cmd_ext, - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - # Filter out lines that indicate success - stdout = [] - for line in data.stderr.split("\n"): + def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]: + """Process the stdout into a list of lines.""" + out = [] + for line in stderr.split("\n"): if line.startswith('wrote extracted data to "'): - stdout.append(line) - elif line: - stderr += line + "\n" + out.append(line) + return out - # Zip extracted files - zip_data = subprocess.run( - ["7z", "a", "../steghide.7z", "*"], - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) + def process_error(self, stdout: str, stderr: str) -> str: + """Process stderr.""" + if "the file format of the file" in stderr and "not supported" in stderr: + return "The file format of the file is not supported (JPEG or BMP only)." + out = "" + for line in stderr.split("\n"): + if line and not line.startswith('wrote extracted data to "'): + out += line + "\n" + return out - stderr += zip_data.stderr - if len(stderr) > 0: - err = { - "steghide": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - # Remove the extracted directory - shutil.rmtree(extracted_dir) - - update_data( - output_dir, - { - "steghide": { - "status": "ok", - "output": stdout, - "download": f"/download/{output_dir.name}/steghide", - } - }, - ) - - except Exception as e: - update_data(output_dir, {"steghide": {"status": "error", "error": str(e)}}) - return None +def analyze_steghide(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: + """Analyze an image submission using steghide.""" + analyzer = SteghideAnalyzer(input_img, output_dir) + if password: + analyzer.analyze(password) + else: + analyzer.analyze() diff --git a/aperisolve/analyzers/strings.py b/aperisolve/analyzers/strings.py index 6757fdb..0363814 100644 --- a/aperisolve/analyzers/strings.py +++ b/aperisolve/analyzers/strings.py @@ -3,45 +3,20 @@ # mypy: disable-error-code=unused-awaitable """Strings Analyzer for Image Submissions.""" -import subprocess from pathlib import Path -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class StringsAnalyzer(SubprocessAnalyzer): + """Analyzer for strings.""" + + def __init__(self, *args): + super().__init__("strings", *args) + self.cmd = ["strings", self.img] def analyze_strings(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using strings.""" - - try: - data = subprocess.run( - ["strings", input_img], - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - if data.stderr: - update_data( - output_dir, {"strings": {"status": "error", "error": data.stderr}} - ) - return - - # strings_dir = output_dir / "strings" - # strings_dir.mkdir(parents=True, exist_ok=True) - # with open(strings_dir / "strings.txt", "w", encoding="utf-8") as f: - # subprocess.run(["strings", input_img], stdout=f, check=True) - - data_strings = data.stdout.split("\n") if data else [] - data_strings = [s for s in data_strings if s] - - update_data( - output_dir, - { - "strings": { - "status": "ok", - "output": data_strings, - } - }, - ) - except Exception as e: - update_data(output_dir, {"strings": {"status": "error", "error": str(e)}}) + analyzer = StringsAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/template_analyzer.py b/aperisolve/analyzers/template_analyzer.py index 0008497..8f33a34 100644 --- a/aperisolve/analyzers/template_analyzer.py +++ b/aperisolve/analyzers/template_analyzer.py @@ -1,79 +1,30 @@ # flake8: noqa: E203,E501,W503 # pylint: disable=C0413,W0718,R0903,R0801 # mypy: disable-error-code=unused-awaitable -"""Custom Analyzer for Image Submissions.""" +""" Analyzer for Image Submissions.""" -import shutil -import subprocess from pathlib import Path +from typing import Optional -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer -def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None: - """Analyze an image submission using myanalyzer.""" +class TemplateAnalyze(SubprocessAnalyzer): + """Analyzer for .""" - image_name = input_img.name + def __init__(self, *args): + super().__init__("", *args, has_archive=True) - # Dir generated by the tool - extracted_dir = output_dir / "myanalyzer_output_dir" + def build_cmd(self, password: Optional[str] = None) -> list[str]: + if password: + return ["", "-p", password, "-i", self.img] + return ["", "-i", self.img] - try: - stderr, stdout = "", "" - # Run myanalyzer - data = subprocess.run( - ["myanalyzer", "-e", "../" + str(image_name)], - cwd=output_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - - stderr += data.stderr - stdout += data.stdout - - if len(stderr) > 0: # Return error if error has been raised - err = { - "myanalyzer": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - zip_exist = False - if extracted_dir.exists(): # If result file(s) exist, zip them - subprocess.run( - ["7z", "a", "../myanalyzer.7z", "*"], - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - zip_exist = True - - if extracted_dir.exists(): # Remove the extracted directory - shutil.rmtree(extracted_dir) - - output = data.stdout.split("\n") if data else [] - output_data = { - "myanalyzer": { - "status": "ok", - "output": output, - } - } - - if zip_exist: - output_data["myanalyzer"][ - "download" - ] = f"/download/{output_dir.name}/myanalyzer" - - update_data(output_dir, output_data) - - except Exception as e: - update_data(output_dir, {"myanalyzer": {"status": "error", "error": str(e)}}) - return None +def analyze_template(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: + """Analyze an image submission using .""" + analyzer = TemplateAnalyze(input_img, output_dir) + if password: + analyzer.analyze(password) + else: + analyzer.analyze() diff --git a/aperisolve/analyzers/zsteg.py b/aperisolve/analyzers/zsteg.py index 0435841..3546efb 100644 --- a/aperisolve/analyzers/zsteg.py +++ b/aperisolve/analyzers/zsteg.py @@ -1,43 +1,31 @@ # flake8: noqa: E203,E501,W503 # pylint: disable=C0413,W0718,R0903,R0801 # mypy: disable-error-code=unused-awaitable -"""Zsteg Analyzer for Image Submissions.""" +"""Strings Analyzer for Image Submissions.""" -import subprocess from pathlib import Path -from .utils import MAX_PENDING_TIME, update_data +from .base_analyzer import SubprocessAnalyzer + + +class ZstegAnalyzer(SubprocessAnalyzer): + """Analyzer for zsteg.""" + + def __init__(self, *args, **kwargs): + super().__init__("zsteg", *args, **kwargs) + self.cmd = ["zsteg", self.img] + + def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + return bool(stderr) or "PNG::NotSupported" in stdout[:100] + + def process_error(self, stdout: str, stderr: str) -> str: + """Process the stderr.""" + if "PNG::NotSupported" in stdout[:100]: + return "The file format of the file is not supported (PNG only)." + return stderr def analyze_zsteg(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using zsteg.""" - - try: - data = subprocess.run( - ["zsteg", input_img], - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - if data.stderr or "PNG::NotSupported" in data.stdout[:100]: - error_msg = data.stderr or data.stdout - update_data(output_dir, {"zsteg": {"status": "error", "error": error_msg}}) - return - - # zsteg_dir = output_dir / "zsteg" - # zsteg_dir.mkdir(parents=True, exist_ok=True) - # with open(zsteg_dir / "zsteg.txt", "w", encoding="utf-8") as f: - # subprocess.run(["zsteg", input_img], stdout=f, check=True) - - update_data( - output_dir, - { - "zsteg": { - "status": "ok", - "output": data.stdout.split("\n") if data else [], - } - }, - ) - except Exception as e: - update_data(output_dir, {"zsteg": {"status": "error", "error": str(e)}}) + analyzer = ZstegAnalyzer(input_img, output_dir) + analyzer.analyze()