diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 6294c1e..e0c4365 100644 --- a/.github/workflows/lint.yml +++ b/.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 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 16a1f1f..b8fa7af 100644 --- a/.pre-commit-config.yaml +++ b/.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: @@ -25,4 +31,4 @@ repos: hooks: - id: mypy args: ["--disallow-any-generics", "--disallow-untyped-def", "--follow-imports", "skip", "--ignore-missing-imports", "--disable-error-code", "unused-awaitable"] - language: system \ No newline at end of file + language: system diff --git a/README.md b/README.md index 285ca5f..b84a9ea 100644 --- a/README.md +++ b/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. diff --git a/aperisolve/analyzers/base_analyzer.py b/aperisolve/analyzers/base_analyzer.py new file mode 100644 index 0000000..5f20a05 --- /dev/null +++ b/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 diff --git a/aperisolve/analyzers/binwalk.py b/aperisolve/analyzers/binwalk.py index 9b8cfac..caf755a 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, 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.""" - - 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/decomposer.py b/aperisolve/analyzers/decomposer.py index cff14d5..9946c6e 100644 --- a/aperisolve/analyzers/decomposer.py +++ b/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) - 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 = 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) + analyzer = DecomposerAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/analyzers/exiftool.py b/aperisolve/analyzers/exiftool.py index 5ed107d..d35d216 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, 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.""" - - 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..85a667b 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, 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.""" - - 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/image_resize.py b/aperisolve/analyzers/image_resize.py index 17db9b5..4b4a12a 100644 --- a/aperisolve/analyzers/image_resize.py +++ b/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( - png_bytes: bytearray, - 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") +class ResizeAnalyzer(SubprocessAnalyzer): + """Analyzer for image resize.""" - full_png_data = ( - png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End] - + width_bytes - + height_bytes - + png_bytes[24:] - ) + png: PNG - with output_path.open("wb") as out_f: - out_f.write(full_png_data) + def __init__(self, input_img: Path, output_dir: Path) -> None: + 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]: - """ - Look up PNG images in the database by their CRC value. + Returns: + List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their + 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: - png (PNG): A PNG object containing the CRC value to search for. + def _search_height_crc(self) -> List[PNG]: + """ + Search for PNG images with matching CRC by iterating through possible heights. - Returns: - List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their packed - representations. - """ - with app.app_context(): - imgs = IHDR.query.filter_by(crc=png.crc).all() + 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. + + 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 img in imgs: - packed = img.packed - results.append(PNG.from_packed(packed, crc=png.crc)) - return results - - -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." + for height in range(100, 3501): + candidate = PNG( + size=(self.png.width, height), + bit_depth=self.png.bit_depth, + color_type=self.png.color_type, + interlace=self.png.interlace, ) - update_data( - output_dir, - { - "image_resize": { + if candidate.crc == self.png.crc: + results.append(candidate) + return results + + 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(): - 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 = [] - saved_img_urls = [] - - db_matches = lookup_crc(png) - if not db_matches: - db_matches = search_height_crc(png) - if not db_matches: - logs.append("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 - ) + # No match found + if not (db_matches := self._lookup_crc()) and not ( + db_matches := self._search_height_crc() + ): + return { + "status": "error", + "error": "Failure: No matching dimensions found.", + } + # 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 - output_data: dict[str, Any] = { - "image_resize": { + return { "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() diff --git a/aperisolve/analyzers/openstego.py b/aperisolve/analyzers/openstego.py index aa6e989..2588150 100644 --- a/aperisolve/analyzers/openstego.py +++ b/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.""" - - image_name = input_img.name - try: - stderr = "" - # Try to extract the embedded file - extracted_dir = output_dir / "openstego" - extracted_dir.mkdir(parents=True, exist_ok=True) - - for algo in ["AES128", "AES256"]: - cmd: list[str] = [ - "openstego", - "extract", - "-a", - "randomlsb", - "--cryptalgo", - algo, - "-sf", - "../" + str(image_name), - "-xd", - str(extracted_dir), - "-p", - password, - ] - - data = subprocess.run( - cmd, - cwd=output_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - - # Check if extraction was successful - if data.returncode != 0: - stderr += data.stderr.replace( - f'"../{image_name}" ', "" - ) # hide file name - err = { - "openstego": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - # Find the extracted file - extracted_files = list(extracted_dir.glob("*")) - if not extracted_files: - err = { - "openstego": { - "status": "error", - "error": "No file extracted, password may be incorrect.", - } - } - update_data(output_dir, err) - return None - - # Filter stdout for success messages - stdout = [] - for f in extracted_files: - stdout.append(f"Recovered file: {f.name}") - - # Zip extracted files - zip_data = subprocess.run( - ["7z", "a", "../openstego.7z", "*"], - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - timeout=MAX_PENDING_TIME, - ) - - stderr += zip_data.stderr - - if len(stderr) > 0: - err = { - "openstego": { - "status": "error", - "error": stderr, - } - } - update_data(output_dir, err) - return None - - # Remove the extracted directory - shutil.rmtree(extracted_dir) - - update_data( - output_dir, - { - "openstego": { - "status": "ok", - "output": stdout, - "download": f"/download/{output_dir.name}/openstego", - } - }, - ) - - except Exception as e: - update_data(output_dir, {"openstego": {"status": "error", "error": str(e)}}) - return None + analyzer = OpenStegoAnalyzer(input_img, output_dir) + if password: + analyzer.analyze(password) + else: + analyzer.analyze() diff --git a/aperisolve/analyzers/outguess.py b/aperisolve/analyzers/outguess.py index 9fc6a78..e99d996 100644 --- a/aperisolve/analyzers/outguess.py +++ b/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( - 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, input_img: Path, output_dir: Path) -> None: + super().__init__("outguess", input_img, output_dir, 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, - ] - else: - cmd = ["outguess", "-r", "../../" + str(image_name), outguess_file] + return ["outguess", "-k", password, "-r", self.img, out] + 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..530c5d6 100644 --- a/aperisolve/analyzers/pngcheck.py +++ b/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.""" - - 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..c1785c2 100644 --- a/aperisolve/analyzers/steghide.py +++ b/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( - 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, input_img: Path, output_dir: Path) -> None: + super().__init__("steghide", input_img, output_dir, 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_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: - 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 embedded_filename is None and extracted_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) + 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 (stdout + "\n" + stderr).split("\n"): + if "wrote extracted data to" in line: + out.append(line) + return out - 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", ""] + 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)." + return stderr - 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 - zip_data = subprocess.run( - ["7z", "a", "../steghide.7z", "*"], - cwd=extracted_dir, - capture_output=True, - text=True, - check=False, - 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 +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..4d25a58 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, 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.""" - - 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..cb5306a 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 TemplateAnalyzer(SubprocessAnalyzer): + """Analyzer for .""" - image_name = input_img.name + def __init__(self, input_img: Path, output_dir: Path) -> None: + super().__init__("", input_img, output_dir, 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 = TemplateAnalyzer(input_img, output_dir) + if password: + analyzer.analyze(password) + else: + analyzer.analyze() diff --git a/aperisolve/analyzers/utils.py b/aperisolve/analyzers/utils.py index 3387106..a2ee0d0 100644 --- a/aperisolve/analyzers/utils.py +++ b/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 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) +from typing import Iterable, Self 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( - ">IIBBBBB", ihdr_data - ) + width, height, depth, color, _, _, interlace = struct.unpack(">IIBBBBB", ihdr_data) return cls( size=(width, height), diff --git a/aperisolve/analyzers/zsteg.py b/aperisolve/analyzers/zsteg.py index 0435841..df67ef4 100644 --- a/aperisolve/analyzers/zsteg.py +++ b/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.""" - - 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() diff --git a/aperisolve/app.py b/aperisolve/app.py index 734097c..ae869ea 100644 --- a/aperisolve/app.py +++ b/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( - os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024) -) +app.config["MAX_CONTENT_LENGTH"] = int(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 ( - "." not in image.filename - or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS - ): + if "." 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_FOLDER / str(image.hash) / str(submission.hash) / "results.json" - ) + result_path = 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/") @app.route("/image//") -def get_image( - hash_val: Optional[str] = None, img_name: Optional[str] = None -) -> Response: +def get_image(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 ( - output_file.suffix.lower() not in IMAGE_EXTENSIONS - ): + if (not output_file.exists()) or (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) diff --git a/aperisolve/cleanup.py b/aperisolve/cleanup.py index 5a63f96..032a7b6 100644 --- a/aperisolve/cleanup.py +++ b/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_FOLDER / str(submission.image_hash) / str(submission.hash) - ) + result_path = 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 diff --git a/aperisolve/models.py b/aperisolve/models.py index dc47053..22d6d34 100644 --- a/aperisolve/models.py +++ b/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 - IHDR(packed=p.packed, crc=p.crc) - ) + db.session.add(IHDR(packed=p.packed, crc=p.crc)) # pylint: disable=no-member count += 1 if count % 5000 == 0: diff --git a/aperisolve/sentry.py b/aperisolve/sentry.py index 13b5871..c0307a5 100644 --- a/aperisolve/sentry.py +++ b/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( - os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1") -) +SENTRY_PROFILES_SAMPLE_RATE = float(os.environ.get("SENTRY_PROFILES_SAMPLE_RATE", "0.1")) SENTRY_RELEASE = os.environ.get("SENTRY_RELEASE", "1.0.0") diff --git a/aperisolve/static/css/aperisolve.css b/aperisolve/static/css/aperisolve.css index 4279412..62d1901 100644 --- a/aperisolve/static/css/aperisolve.css +++ b/aperisolve/static/css/aperisolve.css @@ -379,6 +379,10 @@ td { height: 400px; } +.a-foremost textarea { + height: 100px; +} + /* ======== Footer ======== */ footer { @@ -420,4 +424,4 @@ footer a i { } #httperror h1, #httperror h2{ text-align: center; -} \ No newline at end of file +} diff --git a/aperisolve/static/js/aperisolve.js b/aperisolve/static/js/aperisolve.js index 0294ead..06ec2e8 100644 --- a/aperisolve/static/js/aperisolve.js +++ b/aperisolve/static/js/aperisolve.js @@ -254,9 +254,8 @@ function parseResult(result) { // Parse text output if (typeof result[tool]["output"] === "string") { - analyzer.innerHTML += `
${escapeHtml(
-        result[tool]["output"]
-      )}
`; + output = escapeHtml(result[tool]["output"]); + analyzer.innerHTML += ``; } else if (Array.isArray(result[tool]["output"])) { if (result[tool]["output"].length > 0) { var texarea_content = `
`; @@ -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 += ``; texarea_content += ``; texarea_content += `
`; diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 72b7861..8ab66ef 100644 --- a/aperisolve/workers.py +++ b/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 diff --git a/compose.dev.yml b/compose.dev.yml index 6f46315..77baee1 100644 --- a/compose.dev.yml +++ b/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