Merge pull request #128 from Zeecka/enhancement/refactor_workers

Refactoring Workers

.github/workflows/lint.yml

@@ -34,6 +34,7 @@ jobs:
         uses: wearerequired/lint-action@v2
         with:
           black: true
+          black_args: "--line-length=100"
           flake8: true
           flake8_args: "--ignore=E203,E501,W503"
           pylint: true

.pre-commit-config.yaml

@@ -9,6 +9,12 @@ repos:
   rev: 25.11.0
   hooks:
   - id: black
+    args: ['--line-length', '100']
+- repo: https://github.com/pycqa/isort
+  rev: 5.11.2
+  hooks:
+  - id: isort
+    args: ['--profile', 'black']
 - repo: https://github.com/PyCQA/flake8
   rev: 7.3.0
   hooks:

README.md

@@ -208,10 +208,13 @@ Contributions are welcome! Please:
 > [!IMPORTANT]
 > Please follow the code style and run linters before submitting. The project adheres to:
 >
-> - Black
+> - Black (line length 100)
+> - Isort (profile black)
 > - Flake8 (ignoring E203, E501, W503)
 > - Pylint (ignoring W0718, R0903, R0801)
 > - Mypy (ignoring unused-awaitable)
+>
+> See [.pre-commit-config.yaml](.pre-commit-config.yaml) file.
 
 CI will run these checks on each PR.
 

aperisolve/analyzers/base_analyzer.py

@@ -0,0 +1,179 @@
+# flake8: noqa: E203,E501,W503
+# pylint: disable=C0413,W0718,W0613,R0903,R0801
+# mypy: disable-error-code=unused-awaitable
+"""Base Analyzer class for image analysis tools."""
+
+import fcntl
+import json
+import os
+import subprocess
+import threading
+from abc import ABC
+from pathlib import Path
+from shutil import rmtree
+from typing import Any, Optional, overload
+
+_thread_lock = threading.Lock()
+MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600"))  # 10 minutes by default
+
+
+class SubprocessAnalyzer(ABC):
+    """Analyzer that runs a subprocess command."""
+
+    name: str
+    input_img: Path
+    output_dir: Path
+    has_archive: bool
+    cmd: list[str] | None = None
+    img: str
+    make_folder: bool = True
+
+    def __init__(
+        self, name: str, input_img: Path, output_dir: Path, has_archive: bool = False
+    ) -> None:
+        self.name = name
+        self.input_img = input_img
+        self.img = f"../{self.input_img.name}"
+        self.output_dir = output_dir
+        self.has_archive = has_archive
+
+    def run_command(
+        self, cmd: list[str], cwd: Optional[Path] = None
+    ) -> subprocess.CompletedProcess[str]:
+        """Run a subprocess command."""
+        return subprocess.run(
+            cmd,
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=MAX_PENDING_TIME,
+        )
+
+    def generate_archive(self, output_dir: Path, extracted_dir: Path) -> str:
+        """Zip the extracted files and remove the directory."""
+        zip_data = self.run_command(["7z", "a", f"../{self.name}.7z", "*"], cwd=extracted_dir)
+        rmtree(extracted_dir)
+        return zip_data.stderr
+
+    def update_result(self, result: dict[str, Any]) -> None:
+        """Thread-safe and process-safe JSON update using lock file and atomic replace."""
+        json_file = self.output_dir / "results.json"
+        new_data = {self.name: result}
+        lock_file = json_file.with_suffix(json_file.suffix + ".lock")
+        tmp_file = json_file.with_suffix(json_file.suffix + ".tmp")
+
+        json_file.parent.mkdir(parents=True, exist_ok=True)
+
+        with _thread_lock:  # synchronizes across threads
+            with open(lock_file, "w", encoding="utf-8") as lock:
+                fcntl.flock(lock, fcntl.LOCK_EX)  # synchronizes across processes
+
+                try:
+                    # Read existing JSON
+                    data: dict[Any, Any] = {}
+                    if json_file.exists():
+                        try:
+                            with open(json_file, "r", encoding="utf-8") as f:
+                                data = json.load(f)
+                        except json.JSONDecodeError:
+                            data = {}
+
+                    # Update with new data
+                    data.update(new_data)
+
+                    # Write safely to a temp file
+                    with open(tmp_file, "w", encoding="utf-8") as f:
+                        json.dump(data, f, sort_keys=False)
+
+                    os.replace(tmp_file, json_file)  # ensures file write is atomic
+                finally:
+                    fcntl.flock(lock, fcntl.LOCK_UN)
+
+    def get_extracted_dir(self) -> Path:
+        """Get the extracted directory path. Can be overridden but work as it is."""
+        return self.output_dir / self.name
+
+    @overload
+    def build_cmd(self) -> list[str]: ...
+
+    @overload
+    def build_cmd(self, password: str) -> list[str]: ...
+
+    def build_cmd(self, password: Optional[str] = None) -> list[str]:
+        """Build the command to run. Can be overridden."""
+        if self.cmd is None:
+            raise NotImplementedError("cmd must be set or build_cmd overridden")
+        return self.cmd
+
+    def get_results(self, password: Optional[str] = None) -> dict[str, Any]:
+        """Get results of command before returning."""
+        result: dict[str, Any]
+        extracted_dir = None
+        if self.has_archive:
+            extracted_dir = self.get_extracted_dir()
+            if self.make_folder:
+                extracted_dir.mkdir(parents=True, exist_ok=True)
+
+        cmd: list[str]
+        if password:
+            cmd = list(map(str, self.build_cmd(password)))
+        else:
+            cmd = list(map(str, self.build_cmd()))
+        data = self.run_command(cmd, cwd=self.output_dir)
+        returncode = data.returncode
+        stderr = data.stderr
+        stdout = data.stdout
+
+        zip_exist = False
+        if extracted_dir and extracted_dir.exists() and any(extracted_dir.iterdir()):
+            self.generate_archive(self.output_dir, extracted_dir)  # keep archive
+            zip_exist = True
+
+        if self.is_error(returncode, stdout, stderr, zip_exist):
+            result = {
+                "status": "error",
+            }
+            result["error"] = self.process_error(stdout, stderr)
+        else:
+            result = {
+                "status": "ok",
+                "output": self.process_output(stdout, stderr),
+            }
+            note: Optional[str] = self.process_note(stdout, stderr)
+            if note:
+                result["note"] = note
+            if zip_exist:
+                result["download"] = f"/download/{self.output_dir.name}/{self.name}"
+        return result
+
+    @overload
+    def analyze(self) -> None: ...
+
+    @overload
+    def analyze(self, password: str) -> None: ...
+
+    def analyze(self, password: Optional[str] = None) -> None:
+        """Run the subprocess command and handle results."""
+        result: dict[str, Any]
+        try:
+            result = self.get_results(password)
+            self.update_result(result)
+        except Exception as e:
+            self.update_result({"status": "error", "error": str(e)})
+
+    def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool:
+        """Check if the result is an error."""
+        return len(stderr) > 0
+
+    def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
+        """Process the stdout into a list of lines."""
+        return [line for line in stdout.split("\n") if line] if stdout else []
+
+    def process_error(self, stdout: str, stderr: str) -> str:
+        """Process the stderr."""
+        return stderr
+
+    def process_note(self, stdout: str, stderr: str) -> Optional[str]:
+        """Process the stdout for informational purposes."""
+        return None

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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("binwalk", input_img, output_dir, 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
 
 
 def analyze_binwalk(input_img: Path, output_dir: Path) -> None:
     """Analyze an image submission using binwalk."""
-
+    analyzer = BinwalkAnalyzer(input_img, output_dir)
-    image_name = input_img.name
+    analyzer.analyze()
-    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

aperisolve/analyzers/decomposer.py

@@ -4,78 +4,92 @@
 """Bits Decomposer Analyzer for Image Submissions."""
 
 from pathlib import Path
+from typing import Any, Optional
 
 import numpy as np
 from PIL import Image
 
-from .utils import update_data
+from .base_analyzer import SubprocessAnalyzer
+
+
+class DecomposerAnalyzer(SubprocessAnalyzer):
+    """Analyzer for bits decomposer."""
+
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
+        super().__init__("decomposer", input_img, output_dir)
+
+    def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
+        """Analyze an image submission using bits decomposition."""
+        try:
+            img = Image.open(self.input_img)
+            converted = False
+
+            if img.mode == "P":
+                img = img.convert("RGB")
+                converted = True
+
+            img_np = np.array(img)
+
+            # Handle grayscale, RGB, RGBA, etc.
+            if len(img_np.shape) == 2:
+                channels = 1
+            else:
+                channels = img_np.shape[2]
+
+            if channels > 1:
+                channel_names = ["Red", "Green", "Blue", "Alpha"]
+            else:
+                channel_names = ["Grayscale"]
+
+            image_json = {}
+
+            # Add Superimposed RGB bit planes
+            if channels >= 3:
+                superimposed_json = []
+                for bit in range(8):
+                    bit_mask = 1 << bit
+                    rgb_planes = [((img_np[..., c] & bit_mask) >> bit) * 255 for c in range(3)]
+                    rgb_image = np.stack(rgb_planes, axis=-1).astype(np.uint8)
+                    rgb_img = Image.fromarray(rgb_image, mode="RGB")
+                    img_name = f"superimposed_bit_{bit}.png"
+                    out_path = self.output_dir / img_name
+                    dl_path = Path(self.output_dir.name) / img_name
+                    superimposed_json.append("/image/" + str(dl_path))
+                    rgb_img.save(out_path)
+
+                image_json["Superimposed"] = superimposed_json
+
+            # Process individual channels
+            for c in range(channels):
+                channel_data = img_np[..., c] if channels > 1 else img_np
+                channel_label = channel_names[c]
+                channel_json = []
+                for bit in range(8):
+                    bit_mask = 1 << bit
+                    bit_plane = ((channel_data & bit_mask) >> bit) * 255
+                    bit_img = Image.fromarray(bit_plane.astype(np.uint8), mode="L")
+                    img_name = f"{channel_label}_bit_{bit}.png"
+                    out_path = self.output_dir / img_name
+                    dl_path = Path(self.output_dir.name) / img_name
+                    channel_json.append("/image/" + str(dl_path))
+                    bit_img.save(out_path)
+                image_json[channel_label] = channel_json
+
+            output = {
+                "status": "ok",
+                "images": image_json,
+            }
+            if converted:
+                output["note"] = (
+                    "Image contains a color palette and was converted to RGB for processing."
+                )
+
+            return output
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
 
 
 def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
     """Analyze an image submission using bits decomposition."""
-    img = Image.open(input_img)
+    analyzer = DecomposerAnalyzer(input_img, output_dir)
-    converted = False
+    analyzer.analyze()
-
-    if img.mode == "P":
-        img = img.convert("RGB")
-        converted = True
-
-    img_np = np.array(img)
-
-    # Handle grayscale, RGB, RGBA, etc.
-    if len(img_np.shape) == 2:
-        channels = 1
-    else:
-        channels = img_np.shape[2]
-
-    if channels > 1:
-        channel_names = ["Red", "Green", "Blue", "Alpha"]
-    else:
-        channel_names = ["Grayscale"]
-
-    image_json = {}
-
-    # Add Superimposed RGB bit planes
-    if channels >= 3:
-        superimposed_json = []
-        for bit in range(8):
-            bit_mask = 1 << bit
-            rgb_planes = [((img_np[..., c] & bit_mask) >> bit) * 255 for c in range(3)]
-            rgb_image = np.stack(rgb_planes, axis=-1).astype(np.uint8)
-            rgb_img = Image.fromarray(rgb_image, mode="RGB")
-            img_name = f"superimposed_bit_{bit}.png"
-            out_path = output_dir / img_name
-            dl_path = Path(output_dir.name) / img_name
-            superimposed_json.append("/image/" + str(dl_path))
-            rgb_img.save(out_path)
-
-        image_json["Superimposed"] = superimposed_json
-
-    # Process individual channels
-    for c in range(channels):
-        channel_data = img_np[..., c] if channels > 1 else img_np
-        channel_label = channel_names[c]
-        channel_json = []
-        for bit in range(8):
-            bit_mask = 1 << bit
-            bit_plane = ((channel_data & bit_mask) >> bit) * 255
-            bit_img = Image.fromarray(bit_plane.astype(np.uint8), mode="L")
-            img_name = f"{channel_label}_bit_{bit}.png"
-            out_path = output_dir / img_name
-            dl_path = Path(output_dir.name) / img_name
-            channel_json.append("/image/" + str(dl_path))
-            bit_img.save(out_path)
-        image_json[channel_label] = channel_json
-
-    output = {
-        "decomposer": {
-            "status": "ok",
-            "images": image_json,
-        }
-    }
-    if converted:
-        output["decomposer"][
-            "note"
-        ] = "Image contains a color palette and was converted to RGB for processing."
-
-    update_data(output_dir, output)

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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("exiftool", input_img, output_dir)
+        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."""
-
+    analyzer = ExiftoolAnalyzer(input_img, output_dir)
-    try:
+    analyzer.analyze()
-        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)}})

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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("foremost", input_img, output_dir, 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."""
-
+    analyzer = ForemostAnalyzer(input_img, output_dir)
-    try:
+    analyzer.analyze()
-        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)}})

aperisolve/analyzers/image_resize.py

@@ -4,195 +4,167 @@
 """Image size bruteforcer for common PNG sizes."""
 
 from pathlib import Path
-from typing import Any, List
+from typing import Any, List, Optional
 
 from ..app import app
 from ..models import IHDR
-from .utils import PNG, update_data
+from .base_analyzer import SubprocessAnalyzer
+from .utils import PNG
 
 
-def write_recovered_png(
+class ResizeAnalyzer(SubprocessAnalyzer):
-    png_bytes: bytearray,
+    """Analyzer for image resize."""
-    width: int,
-    height: int,
-    output_path: Path,
-) -> None:
-    """
-    Rebuilds a PNG by patching IHDR width/height and writes it to disk.
-    """
-    height_bytes = height.to_bytes(4, byteorder="big")
-    width_bytes = width.to_bytes(4, byteorder="big")
 
-    full_png_data = (
+    png: PNG
-        png_bytes[:16]  # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
-        + width_bytes
-        + height_bytes
-        + png_bytes[24:]
-    )
 
-    with output_path.open("wb") as out_f:
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
-        out_f.write(full_png_data)
+        super().__init__("image_resize", input_img, output_dir, has_archive=True)
 
+    def _lookup_crc(self) -> List[PNG]:
+        """
+        Look up PNG images in the database by their CRC value.
 
-def lookup_crc(png: PNG) -> List[PNG]:
+        Returns:
-    """
+            List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their
-    Look up PNG images in the database by their CRC value.
+                       packed representations.
+        """
+        with app.app_context():
+            imgs = IHDR.query.filter_by(crc=self.png.crc).all()
+            results = []
+            for img in imgs:
+                packed = img.packed
+                results.append(PNG.from_packed(packed, crc=self.png.crc))
+        return results
 
-    Args:
+    def _search_height_crc(self) -> List[PNG]:
-        png (PNG): A PNG object containing the CRC value to search for.
+        """
+        Search for PNG images with matching CRC by iterating through possible heights.
 
-    Returns:
+        This function attempts to find alternative PNG configurations that produce the same
-        List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their packed
+        CRC checksum as the input PNG by systematically varying the height parameter while
-                   representations.
+        keeping all other properties constant.
-    """
+
-    with app.app_context():
+        Returns:
-        imgs = IHDR.query.filter_by(crc=png.crc).all()
+            List[PNG]: A list of PNG objects with matching CRC values. Returns an empty list
+                    if no matches are found.
+
+        Note:
+            The function iterates through heights from 100 to 3500 (inclusive).
+            This may be computationally expensive for large ranges.
+        """
         results = []
-        for img in imgs:
+        for height in range(100, 3501):
-            packed = img.packed
+            candidate = PNG(
-            results.append(PNG.from_packed(packed, crc=png.crc))
+                size=(self.png.width, height),
-    return results
+                bit_depth=self.png.bit_depth,
-
+                color_type=self.png.color_type,
-
+                interlace=self.png.interlace,
-def search_height_crc(png: PNG) -> List[PNG]:
-    """
-    Search for PNG images with matching CRC by iterating through possible heights.
-
-    This function attempts to find alternative PNG configurations that produce the same
-    CRC checksum as the input PNG by systematically varying the height parameter while
-    keeping all other properties constant.
-
-    Args:
-        png (PNG): The reference PNG object whose CRC checksum will be matched.
-
-    Returns:
-        List[PNG]: A list of PNG objects with matching CRC values. Returns an empty list
-                   if no matches are found.
-
-    Note:
-        The function iterates through heights from 100 to 3500 (inclusive).
-        This may be computationally expensive for large ranges.
-    """
-    results = []
-    for height in range(100, 3501):
-        candidate = PNG(
-            size=(png.width, height),
-            bit_depth=png.bit_depth,
-            color_type=png.color_type,
-            interlace=png.interlace,
-        )
-        if candidate.crc == png.crc:
-            results.append(candidate)
-    return results
-
-
-def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
-    """
-    Analyze a PNG image and recover resized versions based on CRC matching.
-
-    This function attempts to find matching image dimensions for a given PNG file
-    by looking up its CRC (Cyclic Redundancy Check) in a database. If matches are
-    found, it recovers and saves PNG images with the matched dimensions.
-
-    Args:
-        input_img (Path): Path to the input PNG image file to analyze.
-        output_dir (Path): Path to the directory where recovered PNG images will be saved.
-
-    Returns:
-        None
-
-    Raises:
-        Handles exceptions gracefully by updating output data with error status.
-
-    Side Effects:
-        - Creates output directory if it doesn't exist.
-        - Saves recovered PNG images to the output directory.
-        - Updates data in the output directory with analysis results via update_data().
-
-    Notes:
-        - If the PNG has invalid structure or IHDR chunk is not first, an error is logged.
-        - The function performs a two-step lookup: first by CRC, then by height/CRC if no matches
-          found.
-        - Recovered images are named as "recovered_{width}x{height}.png".
-        - Analysis results (logs, status, and image URLs) are persisted via update_data().
-    """
-
-    # 1. Setup Paths
-    input_img = Path(input_img)
-    output_dir = Path(output_dir)
-
-    # 3. Initialize Logs
-    logs = []
-
-    try:
-        with input_img.open("rb") as image:
-            img_bytes = bytearray(image.read())
-            try:
-                png = PNG.from_bytearray(img_bytes)
-            except ValueError:
-                png = None
-
-        if png is None:
-            logs.append(
-                "Failure: IHDR chunk is not the first chunk, or PNG has invalid structure."
             )
-            update_data(
+            if candidate.crc == self.png.crc:
-                output_dir,
+                results.append(candidate)
-                {
+        return results
-                    "image_resize": {
+
+    def _write_recovered_png(self, png_bytes: bytearray, png_match: PNG) -> str:
+        """
+        Rebuilds a PNG by patching IHDR width/height and writes it to disk.
+        """
+
+        img_name = f"recovered_{png_match.width}x{png_match.height}.png"
+        output_path = self.output_dir / img_name
+
+        height_bytes = png_match.height.to_bytes(4, byteorder="big")
+        width_bytes = png_match.width.to_bytes(4, byteorder="big")
+
+        full_png_data = (
+            png_bytes[:16]  # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
+            + width_bytes
+            + height_bytes
+            + png_bytes[24:]
+        )
+
+        with output_path.open("wb") as out_f:
+            out_f.write(full_png_data)
+
+        return img_name
+
+    def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
+        """
+        Analyze a PNG image and recover resized versions based on CRC matching.
+
+        This function attempts to find matching image dimensions for a given PNG file
+        by looking up its CRC (Cyclic Redundancy Check) in a database. If matches are
+        found, it recovers and saves PNG images with the matched dimensions.
+
+        Returns:
+            dict[str, Any]: A dictionary containing the analysis status, any error message,
+
+        Raises:
+            Handles exceptions gracefully by updating output data with error status.
+
+        Side Effects:
+            - Creates output directory if it doesn't exist.
+            - Saves recovered PNG images to the output directory.
+            - Updates data in the output directory with analysis results via update_data().
+
+        Notes:
+            - If the PNG has invalid structure or IHDR chunk is not first, an error is logged.
+            - The function performs a two-step lookup: first by CRC, then by height/CRC if no
+              matches found.
+            - Recovered images are named as "recovered_{width}x{height}.png".
+            - Analysis results (logs, status, and image URLs) are persisted via update_data().
+        """
+        logs = []
+
+        try:
+            with self.input_img.open("rb") as image:
+                img_bytes = bytearray(image.read())
+                try:
+                    self.png = PNG.from_bytearray(img_bytes)
+                except ValueError:
+                    return {
                         "status": "error",
                         "error": "IHDR chunk is not the first chunk, or PNG has invalid structure.",
                     }
-                },
-            )
-            return
 
-        logs.append(f"Target CRC found: 0x{png.crc:08x}")
+            # Already correct
+            if self.png.crc == self.png.compute_crc():
+                return {
+                    "status": "ok",
+                    "note": f"PNG is already valid with dimensions "
+                    f"{self.png.width}x{self.png.height} and crc 0x{self.png.crc:08x}.",
+                }
 
-        if png.crc == png.compute_crc():
+            saved_img_urls = []
-            note = (
-                f"PNG is already valid with dimensions {png.width}x{png.height} "
-                f"and crc 0x{png.crc:08x}."
-            )
-            output_no_compute: dict[str, Any] = {
-                "image_resize": {"status": "ok", "note": note}
-            }
-            update_data(output_dir, output_no_compute)
-            return
 
-        saved_img_urls = []
+            # No match found
-
+            if not (db_matches := self._lookup_crc()) and not (
-        db_matches = lookup_crc(png)
+                db_matches := self._search_height_crc()
-        if not db_matches:
+            ):
-            db_matches = search_height_crc(png)
+                return {
-        if not db_matches:
+                    "status": "error",
-            logs.append("Failure: No matching dimensions found.")
+                    "error": "Failure: No matching dimensions found.",
-        else:
+                }
-            # Create parent directories if they don't exist
-            output_dir.mkdir(parents=True, exist_ok=True)
-
-            # Iterate through ALL candidates found
-            for im_match in db_matches:
-                img_name = f"recovered_{im_match.width}x{im_match.height}.png"
-                output_path = output_dir / img_name
-                write_recovered_png(
-                    img_bytes, im_match.width, im_match.height, output_path
-                )
 
+            # Match found, saving image(s)
+            logs.append(f"Target CRC found: 0x{self.png.crc:08x}")
+            self.output_dir.mkdir(parents=True, exist_ok=True)
+            for png_match in db_matches:
+                img_name = self._write_recovered_png(img_bytes, png_match)
                 logs.append(f"Image saved: {img_name}")
                 # Append the formatted URL to our list
-                saved_img_urls.append("/image/" + str(Path(output_dir.name) / img_name))
+                saved_img_urls.append("/image/" + str(Path(self.output_dir.name) / img_name))
 
-        # 4. Final Data Update
+            return {
-        output_data: dict[str, Any] = {
-            "image_resize": {
                 "status": "ok",
                 "output": logs,
                 "png_images": saved_img_urls,
             }
-        }
 
-        update_data(output_dir, output_data)
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
 
-    except Exception as e:
+
-        update_data(output_dir, {"image_resize": {"status": "error", "error": str(e)}})
+def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
+    """Analyze an image submission using ResizeAnalyzer."""
+    analyzer = ResizeAnalyzer(input_img, output_dir)
+    analyzer.analyze()

aperisolve/analyzers/openstego.py

@@ -3,115 +3,63 @@
 # mypy: disable-error-code=unused-awaitable
 """OpenStego Analyzer for Image Submissions."""
 
-import shutil
-import subprocess
 from pathlib import Path
+from typing import Any, Optional
 
-from .utils import MAX_PENDING_TIME, update_data
+from .base_analyzer import SubprocessAnalyzer
+
+
+class OpenStegoAnalyzer(SubprocessAnalyzer):
+    """Analyzer for openstego."""
+
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
+        super().__init__("openstego", input_img, output_dir, has_archive=True)
+        self.algo = 0
+
+    def build_cmd(self, password: Optional[str] = None) -> list[str]:
+        """Iterator that return command for OpenStego for every algorithms"""
+        if password is None:
+            password = ""
+
+        algorithms = ["AES128", "AES256"]
+        algo = algorithms[self.algo]
+        self.algo += 1
+        cmd = ["openstego", "extract", "-a", "randomlsb", "--cryptalgo", algo, "-sf"]
+        cmd += [self.img, "-xd", str(self.get_extracted_dir()), "-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."""
+        return (
+            len(list(self.get_extracted_dir().glob("*"))) == 0 and "Extracted file: " not in stderr
+        )
+
+    def analyze(self, password: Optional[str] = None) -> None:
+        """Run the subprocess command and handle results."""
+        result: dict[str, Any]
+        try:
+            result = self.get_results(password)
+            if result["status"] == "error":  # check second algorithm if failed
+                result = self.get_results(password)
+            self.update_result(result)
+        except Exception as e:
+            self.update_result({"status": "error", "error": str(e)})
+
+    def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
+        """Process the stdout/stderr."""
+        return stderr
+
+    def process_error(self, stdout: str, stderr: str) -> str:
+        """Process the stderr."""
+        if "OpenStego is a steganography application" in stderr:
+            return "OpenStego needs a password to work."
+        return stderr
 
 
 def analyze_openstego(input_img: Path, output_dir: Path, password: str = "") -> None:
     """Analyze an image submission using openstego."""
-
+    analyzer = OpenStegoAnalyzer(input_img, output_dir)
-    image_name = input_img.name
+    if password:
-    try:
+        analyzer.analyze(password)
-        stderr = ""
+    else:
-        # Try to extract the embedded file
+        analyzer.analyze()
-        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

aperisolve/analyzers/outguess.py

@@ -3,97 +3,31 @@
 # 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(
+class OutguessAnalyzer(SubprocessAnalyzer):
-    input_img: Path, output_dir: Path, password: Optional[str] = None
+    """Analyzer for outguess."""
-) -> None:
-    """Analyze an image submission using outguess."""
 
-    image_name = input_img.name
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
-    try:
+        super().__init__("outguess", input_img, output_dir, has_archive=True)
-        outguess_dir = output_dir / "outguess"
+
-        outguess_dir.mkdir(parents=True, exist_ok=True)
+    def build_cmd(self, password: Optional[str] = None) -> list[str]:
-        outguess_file = "outguess.data"
+        extracted_dir = self.get_extracted_dir()
-        stderr = ""
+        assert extracted_dir is not None  # since has_archive is True
-        cmd: list[str]
+        out = str(extracted_dir / "outguess.data")
         if password:
-            cmd = [
+            return ["outguess", "-k", password, "-r", self.img, out]
-                "outguess",
+        return ["outguess", "-r", self.img, out]
-                "-k",
-                password,
-                "-r",
-                "../../" + str(image_name),
-                outguess_file,
-            ]
-        else:
-            cmd = ["outguess", "-r", "../../" + str(image_name), outguess_file]
 
-        data = subprocess.run(
-            cmd,
-            cwd=outguess_dir,
-            capture_output=True,
-            text=True,
-            check=False,
-            timeout=MAX_PENDING_TIME,
-        )
 
-        if data.returncode != 0:
+def analyze_outguess(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
-            stderr += data.stderr
+    """Analyze an image submission using outguess."""
-            err = {
+    analyzer = OutguessAnalyzer(input_img, output_dir)
-                "outguess": {
+    if password:
-                    "status": "error",
+        analyzer.analyze(password)
-                    "error": stderr,
+    else:
-                }
+        analyzer.analyze()
-            }
-            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

aperisolve/analyzers/pngcheck.py

@@ -3,54 +3,29 @@
 # 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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("pngcheck", input_img, output_dir)
+        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 process_error(self, stdout: str, stderr: str) -> str:
+        """Process the stderr."""
+        if "this is neither a PNG or JNG image nor a MNG stream" in stdout:
+            return "The file format of the file is not supported (PNG or JNG only)."
+        return stdout
 
 
 def analyze_pngcheck(input_img: Path, output_dir: Path) -> None:
     """Analyze an image submission using pngcheck."""
-
+    analyzer = PngcheckAnalyzer(input_img, output_dir)
-    try:
+    analyzer.analyze()
-        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)}})

aperisolve/analyzers/steghide.py

@@ -4,127 +4,68 @@
 """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(
+class SteghideAnalyzer(SubprocessAnalyzer):
-    input_img: Path, output_dir: Path, password: Optional[str] = None
+    """Analyzer for steghide."""
-) -> None:
-    """Analyze an image submission using steghide."""
 
-    image_name = input_img.name
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
-    try:
+        super().__init__("steghide", input_img, output_dir, has_archive=True)
-        stderr = ""
-        # Get embedded file information first
-        cmd: list[str] = ["steghide", "info", "../" + str(image_name)]
-        if password:
-            cmd += ["-p", password]
-        else:
-            cmd += ["-p", ""]
 
-        data = subprocess.run(
+    def build_cmd(self, password: Optional[str] = None) -> list[str]:
-            cmd,
+        if password is None:
-            cwd=output_dir,
+            password = ""
-            capture_output=True,
+
-            text=True,
+        # First, get the embedded file name, and return the test cmd if failed
-            check=False,
+        cmd_test = ["steghide", "info", self.img, "-p", password]
-            timeout=MAX_PENDING_TIME,
+        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_file = match.group(1)
+        safe_hidden_file = Path(hidden_file).name
+        outfile = str(self.get_extracted_dir() / safe_hidden_file)
+        cmd = ["steghide", "extract", "-sf", self.img, "-xf", outfile, "-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 + stderr)
         embedded_filename = match.group(1) if match else None
 
+        match = re.search(r'wrote extracted data to "([^"]+)".*', stdout + stderr)
+        extracted_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:
+        return embedded_filename is None and extracted_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
 
-        # The passphrase is correct, we can extract the file
+    def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]:
-        # prevent path traversal
+        """Process the stdout into a list of lines."""
-        embedded_file = output_dir / "steghide" / str(Path(embedded_filename).name)
+        out = []
-        extracted_dir = embedded_file.parent
+        for line in (stdout + "\n" + stderr).split("\n"):
-        extracted_dir.mkdir(parents=True, exist_ok=True)
+            if "wrote extracted data to" in line:
+                out.append(line)
+        return out
 
-        cmd_ext: list[str] = [
+    def process_error(self, stdout: str, stderr: str) -> str:
-            "steghide",
+        """Process stderr."""
-            "extract",
+        if "the file format of the file" in stderr and "not supported" in stderr:
-            "-sf",
+            return "The file format of the file is not supported (JPEG or BMP only)."
-            "../../" + str(image_name),
+        return stderr
-            "-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"):
-            if line.startswith('wrote extracted data to "'):
-                stdout.append(line)
-            elif line:
-                stderr += line + "\n"
 
-        # Zip extracted files
+def analyze_steghide(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
-        zip_data = subprocess.run(
+    """Analyze an image submission using steghide."""
-            ["7z", "a", "../steghide.7z", "*"],
+    analyzer = SteghideAnalyzer(input_img, output_dir)
-            cwd=extracted_dir,
+    if password:
-            capture_output=True,
+        analyzer.analyze(password)
-            text=True,
+    else:
-            check=False,
+        analyzer.analyze()
-            timeout=MAX_PENDING_TIME,
-        )
-
-        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

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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("strings", input_img, output_dir)
+        self.cmd = ["strings", self.img]
 
 
 def analyze_strings(input_img: Path, output_dir: Path) -> None:
     """Analyze an image submission using strings."""
-
+    analyzer = StringsAnalyzer(input_img, output_dir)
-    try:
+    analyzer.analyze()
-        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)}})

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."""
+"""<toolname> 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:
+class TemplateAnalyzer(SubprocessAnalyzer):
-    """Analyze an image submission using myanalyzer."""
+    """Analyzer for <toolname>."""
 
-    image_name = input_img.name
+    def __init__(self, input_img: Path, output_dir: Path) -> None:
+        super().__init__("<toolname>", input_img, output_dir, has_archive=True)
 
-    # Dir generated by the tool
+    def build_cmd(self, password: Optional[str] = None) -> list[str]:
-    extracted_dir = output_dir / "myanalyzer_output_dir"
+        if password:
+            return ["<toolname>", "-p", password, "-i", self.img]
+        return ["<toolname>", "-i", self.img]
 
-    try:
-        stderr, stdout = "", ""
 
-        # Run myanalyzer
+def analyze_template(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None:
-        data = subprocess.run(
+    """Analyze an image submission using <toolname>."""
-            ["myanalyzer", "-e", "../" + str(image_name)],
+    analyzer = TemplateAnalyzer(input_img, output_dir)
-            cwd=output_dir,
+    if password:
-            capture_output=True,
+        analyzer.analyze(password)
-            text=True,
+    else:
-            check=False,
+        analyzer.analyze()
-            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

aperisolve/analyzers/utils.py

@@ -3,56 +3,11 @@
 # mypy: disable-error-code=unused-awaitable
 """Utility functions for analyzers modules."""
 
-import fcntl
-import json
 import math
-import os
 import struct
-import threading
 import zlib
 from functools import cached_property
-from pathlib import Path
+from typing import Iterable, Self
-from typing import Any, Iterable, Self
-
-_thread_lock = threading.Lock()
-
-MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600"))  # 10 minutes by default
-
-
-def update_data(
-    output_dir: Path, new_data: dict[Any, Any], json_filename: str = "results.json"
-) -> None:
-    """Thread-safe and process-safe JSON update using lock file and atomic replace."""
-    json_file = Path(output_dir) / json_filename
-    lock_file = json_file.with_suffix(json_file.suffix + ".lock")
-    tmp_file = json_file.with_suffix(json_file.suffix + ".tmp")
-
-    json_file.parent.mkdir(parents=True, exist_ok=True)
-
-    with _thread_lock:  # synchronizes across threads
-        with open(lock_file, "w", encoding="utf-8") as lock:
-            fcntl.flock(lock, fcntl.LOCK_EX)  # synchronizes across processes
-
-            try:
-                # Read existing JSON
-                data: dict[Any, Any] = {}
-                if json_file.exists():
-                    try:
-                        with open(json_file, "r", encoding="utf-8") as f:
-                            data = json.load(f)
-                    except json.JSONDecodeError:
-                        data = {}
-
-                # Update with new data
-                data.update(new_data)
-
-                # Write safely to a temp file
-                with open(tmp_file, "w", encoding="utf-8") as f:
-                    json.dump(data, f, sort_keys=False)
-
-                os.replace(tmp_file, json_file)  # ensures file write is atomic
-            finally:
-                fcntl.flock(lock, fcntl.LOCK_UN)
 
 
 def get_resolutions() -> list[tuple[int, int]]:
@@ -291,9 +246,7 @@ class PNG:
         crc = struct.unpack(">I", data[offset : offset + 4])[0]
 
         # Unpack IHDR fields
-        width, height, depth, color, _, _, interlace = struct.unpack(
+        width, height, depth, color, _, _, interlace = struct.unpack(">IIBBBBB", ihdr_data)
-            ">IIBBBBB", ihdr_data
-        )
 
         return cls(
             size=(width, height),

aperisolve/analyzers/zsteg.py

@@ -3,41 +3,29 @@
 # mypy: disable-error-code=unused-awaitable
 """Zsteg 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, input_img: Path, output_dir: Path) -> None:
+        super().__init__("zsteg", input_img, output_dir)
+        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."""
-
+    analyzer = ZstegAnalyzer(input_img, output_dir)
-    try:
+    analyzer.analyze()
-        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)}})

aperisolve/app.py

@@ -28,9 +28,7 @@ app: Flask = Flask(__name__)
 app.json.sort_keys = False  # type: ignore
 app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URI")
 app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
-app.config["MAX_CONTENT_LENGTH"] = int(
+app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024))
-    os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024)
-)
 RESULT_FOLDER.mkdir(parents=True, exist_ok=True)
 
 db.init_app(app)
@@ -112,10 +110,7 @@ def upload_image() -> tuple[Response, int]:
     if image.filename is None or image.filename == "":
         return jsonify({"error": "No image provided"}), 400
 
-    if (
+    if "." not in image.filename or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS:
-        "." not in image.filename
-        or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS
-    ):
         return jsonify({"error": "Unsupported file type"}), 400
 
     # Compute hashes and prepare submission
@@ -230,9 +225,7 @@ def get_result(hash_val: str) -> tuple[Response, int]:
     submission = Submission.query.get_or_404(hash_val)
     image = Image.query.get_or_404(submission.image_hash)  # type: ignore
 
-    result_path = (
+    result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
-        RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
-    )
 
     if not result_path.exists():
         return jsonify({"error": "Results not ready yet..."}), 425
@@ -261,9 +254,7 @@ def download_output(hash_val: str, tool: str) -> Response:
 
 @app.route("/image/<img_name>")
 @app.route("/image/<hash_val>/<img_name>")
-def get_image(
+def get_image(hash_val: Optional[str] = None, img_name: Optional[str] = None) -> Response:
-    hash_val: Optional[str] = None, img_name: Optional[str] = None
-) -> Response:
     """
     Download the image with a specific name (usually ones generated with
     decomposer) for a given submission hash.
@@ -283,9 +274,7 @@ def get_image(
         image = Image.query.filter_by(hash=hash_val).first_or_404()
         output_file = Path(image.file)  # type: ignore
 
-    if (not output_file.exists()) or (
+    if (not output_file.exists()) or (output_file.suffix.lower() not in IMAGE_EXTENSIONS):
-        output_file.suffix.lower() not in IMAGE_EXTENSIONS
-    ):
         return abort(404, description="Image not found or unsupported format")
 
     return send_file(output_file, as_attachment=True)

aperisolve/cleanup.py

@@ -49,9 +49,7 @@ def cleanup_old_entries() -> None:
             db.session.delete(submission)  # pylint: disable=no-member
         elif submission.status == "done":  # type: ignore
             # Search for buggy results, delete
-            result_path = (
+            result_path = RESULT_FOLDER / str(submission.image_hash) / str(submission.hash)
-                RESULT_FOLDER / str(submission.image_hash) / str(submission.hash)
-            )
             result_file = result_path / "results.json"
             if not result_file.exists():
                 db.session.delete(submission)  # pylint: disable=no-member

aperisolve/models.py

@@ -97,9 +97,7 @@ def fill_ihdr_db() -> None:
             if exists:  # Skip if already present
                 continue
 
-            db.session.add(  # pylint: disable=no-member
+            db.session.add(IHDR(packed=p.packed, crc=p.crc))  # pylint: disable=no-member
-                IHDR(packed=p.packed, crc=p.crc)
-            )
             count += 1
 
             if count % 5000 == 0:

aperisolve/sentry.py

@@ -14,9 +14,7 @@ from sentry_sdk.integrations.threading import ThreadingIntegration
 SENTRY_DSN = os.environ.get("SENTRY_DSN")
 SENTRY_ENVIRONMENT = os.environ.get("SENTRY_ENVIRONMENT", "development")
 SENTRY_TRACES_SAMPLE_RATE = float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", "0.1"))
-SENTRY_PROFILES_SAMPLE_RATE = float(
+SENTRY_PROFILES_SAMPLE_RATE = float(os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1"))
-    os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1")
-)
 SENTRY_RELEASE = os.environ.get("SENTRY_RELEASE", "1.0.0")
 
 

aperisolve/static/css/aperisolve.css

@@ -379,6 +379,10 @@ td {
     height: 400px;
 }
 
+.a-foremost textarea {
+    height: 100px;
+}
+
 /* ======== Footer ======== */
 
 footer {

aperisolve/static/js/aperisolve.js

@@ -254,9 +254,8 @@ function parseResult(result) {
 
     // Parse text output
     if (typeof result[tool]["output"] === "string") {
-      analyzer.innerHTML += `<pre>${escapeHtml(
+      output = escapeHtml(result[tool]["output"]);
-        result[tool]["output"]
+      analyzer.innerHTML += `<div class="alert alert-success" role="alert">${output}</div>`;
-      )}</pre>`;
     } else if (Array.isArray(result[tool]["output"])) {
       if (result[tool]["output"].length > 0) {
         var texarea_content = `<div class="textarea-container">`;
@@ -264,6 +263,7 @@ function parseResult(result) {
         for (const line of result[tool]["output"]) {
           texarea_content += escapeHtml(`${line}\n`);
         }
+        texarea_content = texarea_content.trim();
         texarea_content += `</textarea>`;
         texarea_content += `<i class="fas fa-copy copy-icon"></i>`;
         texarea_content += `</div>`;

aperisolve/workers.py

@@ -89,12 +89,12 @@ def analyze_image(submission_hash: str) -> None:
                 (analyze_decomposer, img_path, result_path),
                 (analyze_exiftool, img_path, result_path),
                 (analyze_foremost, img_path, result_path),
-                (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),
+                (analyze_openstego, img_path, result_path, submission.password),
+                (analyze_pngcheck, img_path, result_path),
+                (analyze_strings, img_path, result_path),
+                (analyze_steghide, img_path, result_path, submission.password),
+                (analyze_zsteg, img_path, result_path),
             ]
 
             # Deep analysis only

compose.dev.yml

@@ -38,6 +38,7 @@ services:
     environment:
       - FLASK_DEBUG=true
       - FLASK_ENV=development
+      - CLEAR_AT_RESTART=0  # Always false for worker to prevent double clearing
 
   redis:
     image: redis:7