Initial commit

aperisolve/analyzers/binwalk.py

@@ -0,0 +1,68 @@
+"""Binwalk Analyzer for Image Submissions."""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+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", "-e", "../" + str(image_name), "--run-as=root"],
+            cwd=output_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        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,
+            )
+            zip_exist = True
+            stderr += zip_data.stderr
+
+        if len(stderr) > 0:
+            err = {
+                "binwalk": {
+                    "status": "error",
+                    "error": stderr,
+                }
+            }
+            update_data(output_dir, err)
+            return None
+
+        # Remove the extracted directory
+        if extracted_dir.exists():
+            shutil.rmtree(extracted_dir)
+
+        output_data = {
+            "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

@@ -0,0 +1,69 @@
+"""Bits Decomposer Analyzer for Image Submissions."""
+
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+from .utils import update_data
+
+
+def analyze_decomposer(input_img: Path, output_dir: Path) -> None:
+    """Analyze an image submission using bits decomposition."""
+    img = Image.open(input_img)
+    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
+
+    update_data(
+        output_dir,
+        {
+            "decomposer": {
+                "status": "ok",
+                "images": image_json,
+            }
+        },
+    )

aperisolve/analyzers/exiftool.py

@@ -0,0 +1,38 @@
+"""Exiftool Analyzer for Image Submissions."""
+
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+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
+        )
+
+        if data.stderr:
+            update_data(
+                output_dir, {"strings": {"status": "error", "error": data.stderr}}
+            )
+            return
+
+        metadata: dict[str, str] = {}
+        for line in data.stdout.split("\n"):
+            if ":" in line:
+                key, value = line.split(":", 1)
+                metadata[key.strip()] = value.strip()
+
+        # exiftool_dir = output_dir / "exiftool"
+        # exiftool_dir.mkdir(parents=True, exist_ok=True)
+        # update_data(exiftool_dir, metadata, "data.json")
+
+        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

@@ -0,0 +1,57 @@
+"""Foremost Analyzer for Image Submissions."""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+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,
+        )
+        # Note: foremost use stderr as standard output for results :o)
+        # So we can't use it as error dectection :
+        # stderr += data.stderr
+
+        # Zip extracted files
+        zip_data = subprocess.run(
+            ["7z", "a", "../foremost.7z", "*"],
+            cwd=foremost_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        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/outguess.py

@@ -0,0 +1,94 @@
+"""Outguess Analyzer for Image Submissions."""
+
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Optional
+
+from .utils import update_data
+
+
+def analyze_outguess(
+    input_img: Path, output_dir: Path, password: Optional[str] = None
+) -> None:
+    """Analyze an image submission using stehide."""
+
+    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]
+        if password:
+            cmd = [
+                "outguess",
+                "-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,
+        )
+
+        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,
+        )
+
+        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/steghide.py

@@ -0,0 +1,118 @@
+"""Steghide Analyzer for Image Submissions."""
+
+import re
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Optional
+
+from .utils import update_data
+
+
+def analyze_steghide(
+    input_img: Path, output_dir: Path, password: Optional[str] = None
+) -> None:
+    """Analyze an image submission using stehide."""
+
+    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", ""]
+
+        data = subprocess.run(
+            cmd,
+            cwd=output_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        match = re.search(r'embedded file "([^"]+)"', data.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
+
+        # 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_filename),
+        ]
+        if password:
+            cmd_ext += ["-p", password]
+        else:
+            cmd_ext += ["-p", ""]
+
+        data = subprocess.run(
+            cmd_ext,
+            cwd=output_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        stderr += data.stderr
+
+        # Zip extracted files
+        zip_data = subprocess.run(
+            ["7z", "a", "../steghide.7z", "*"],
+            cwd=extracted_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        stderr += zip_data.stderr
+
+        if len(stderr) > 0:
+            err = {
+                "binwalk": {
+                    "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": data.stdout.split("\n") if data else [],
+                    "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

@@ -0,0 +1,40 @@
+"""Strings Analyzer for Image Submissions."""
+
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+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
+        )
+        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

@@ -0,0 +1,74 @@
+"""Custom Analyzer for Image Submissions."""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+def analyze_myanalyzer(input_img: Path, output_dir: Path) -> None:
+    """Analyze an image submission using myanalyzer."""
+
+    image_name = input_img.name
+
+    # Dir generated by the tool
+    extracted_dir = output_dir / "myanalyzer_output_dir"
+
+    try:
+        stderr, stdout = "", ""
+
+        # Run myanalyzer
+        data = subprocess.run(
+            ["myanalyzer", "-e", "../" + str(image_name)],
+            cwd=output_dir,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+        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,
+            )
+            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

@@ -0,0 +1,46 @@
+"""Utility functions for analyzers modules."""
+
+import fcntl
+import json
+import os
+import threading
+from pathlib import Path
+from typing import Any
+
+_thread_lock = threading.Lock()
+
+
+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)

aperisolve/analyzers/zsteg.py

@@ -0,0 +1,36 @@
+"""Zsteg Analyzer for Image Submissions."""
+
+import subprocess
+from pathlib import Path
+
+from .utils import update_data
+
+
+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
+        )
+        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)}})