diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 1156ce5..4469413 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -20,19 +20,16 @@ jobs: with: python-version: 3.12 - - name: Install Python dependencies - run: | - pip install -r aperisolve/requirements-dev.txt - pip install -r aperisolve/requirements.txt + - name: Set up uv + uses: astral-sh/setup-uv@v6 - - name: Run linters - uses: wearerequired/lint-action@v2 - with: - black: true - black_args: "--line-length=100" - flake8: true - flake8_args: "--ignore=E203,E501,W503" - pylint: true - pylint_args: "--disable W0511,W0718,R0801,R0903,R0914" - mypy: true - mypy_args: "--disallow-any-generics --disallow-untyped-def --follow-imports skip --ignore-missing-imports --disable-error-code unused-awaitable" + - name: Sync Python dependencies + run: uv sync --extra dev + + - name: Run Ruff + run: | + uv run ruff check . + uv run ruff format . --check + + - name: Run ty + run: uv run ty check aperisolve diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index a32e511..83e51ba 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -8,6 +8,8 @@ on: jobs: docker-releases: runs-on: ubuntu-latest + env: + RELEASE_VERSION: ${{ github.ref_name }} steps: - name: Get release version diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7e52a0a..2e9c714 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -27,6 +27,7 @@ Want to add support for a new steganography or forensics tool? We'd love that! > Check out our detailed tutorial: [Adding a New Analyzer](docs/adding-analyzer.md) **Quick checklist for new analyzers:** + - [ ] Create analyzer file in `analyzers/` - [ ] Extend `SubprocessAnalyzer` base class - [ ] Register in `workers.py` @@ -37,6 +38,7 @@ Want to add support for a new steganography or forensics tool? We'd love that! ### πŸ› Bug Fixes Found a bug? Please: + - Check if it's already reported in [Issues](../../issues) - If not, open a new issue with reproduction steps - Feel free to submit a PR with the fix! @@ -44,6 +46,7 @@ Found a bug? Please: ### πŸ“š Documentation Help us improve: + - Fix typos or unclear explanations - Add examples and use cases - Improve installation instructions @@ -52,6 +55,7 @@ Help us improve: ### 🎨 UI/UX Improvements Contributions to the web interface are welcome: + - Better error messages - Improved result presentation - Mobile responsiveness @@ -63,14 +67,12 @@ Contributions to the web interface are welcome: > Follow the project’s code style and run linters before submitting any code. This project enforces: -- **Black** : line length 100 -- **Isort** : profile black -- **Flake8** : ignoring E203, E501, W503 -- **Pylint** : ignoring W0511, W0718, R0801, R0903, R0914 -- **Mypy** : ignoring unused-awaitable + +- **Ruff** : Check + Format (line length 100) +- **ty** : Type checking > [!TIP] -> All tool configurations (Black, Isort, Flake8, Mypy, Pylint) are centralized in pyproject.toml. +> All tool configurations (Ruff, ty) are centralized in pyproject.toml. > You can run each tool directly and it will automatically pick up the configuration. ### Setup @@ -80,12 +82,9 @@ This project enforces: ```bash # Create & activate a virtual environment -python3 -m venv .venv -source .venv/bin/activate - -# Install dependencies -pip install -r aperisolve/requirements.txt -pip install -r aperisolve/requirements-dev.txt +uv venv +# Install project + development dependencies from pyproject.toml +uv sync --extra dev ``` ### Running Tools Manually @@ -95,67 +94,39 @@ pip install -r aperisolve/requirements-dev.txt Run this script [lint.sh](lint.sh) at the project root folder. ``` -πŸ–€ Running Black... -All done! ✨ 🍰 ✨ -25 files left unchanged. -πŸ“¦ Running Isort... -Skipped 2 files -🐍 Running Flake8... -πŸ” Running Mypy... -Success: no issues found in 25 source files -⚑ Running Pylint... - ------------------------------------- -Your code has been rated at 10.00/10 +🧹 Running Ruff lint checks... +🎨 Running Ruff format checks... +πŸ” Running ty type checks... βœ… All checks passed! ``` +Equivalent manual commands: + +```bash +ruff check . +ruff format . --check +ty check aperisolve +``` + **Option 2 - Pre-Commit hook** Create a pre-commit hooks file [.pre-commit-config.yaml](#) at the project root folder. ```bash repos: -- repo: https://github.com/psf/black - rev: 25.12.0 +- repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.13.0 hooks: - - id: black - args: ['--line-length', '100'] -- repo: https://github.com/pycqa/isort - rev: 7.0.0 - hooks: - - id: isort - args: ['--profile', 'black'] -- repo: https://github.com/PyCQA/flake8 - rev: 7.3.0 - hooks: - - id: flake8 - args: ['--extend-ignore','E203,E501,W503'] -- repo: https://github.com/pylint-dev/pylint - rev: v4.0.4 - hooks: - - id: pylint - args: ['--disable', 'W0511,W0718,R0801,R0903,R0914'] - language: system + - id: ruff-check - repo: local hooks: - - id: mypy - name: mypy - entry: mypy + - id: ty-check + name: ty-check + entry: ty check types: [python] language: system pass_filenames: true - args: - [ - "--disallow-any-generics", - "--disallow-untyped-def", - "--follow-imports", - "skip", - "--ignore-missing-imports", - "--disable-error-code", - "unused-awaitable", - ] ``` Then run the following command so each staged files will be checked while commited. @@ -173,20 +144,20 @@ pre-commit run --all-files > [!NOTE] > Continuous Integration (CI) runs all checks on every PR. Ensure your code passes before submitting. - ## Pull Request Guidelines Use clear, descriptive titles: + - βœ… `Add stegdetect analyzer` - βœ… `Fix binwalk extraction error handling` - βœ… `Update documentation for password-protected tools` - ❌ `Update code` - ❌ `Fix bug` - ### Commit Messages Write clear, concise commit messages: + ```bash # Good git commit -m "Add zsteg analyzer for PNG steganography detection" @@ -204,12 +175,11 @@ git commit -m "fix" If your contribution needs new dependencies: -1. **Python packages**: Add to `requirements.txt` +1. **Python packages**: Add to `pyproject.toml` (`[project.dependencies]` or `[project.optional-dependencies.dev]`) 2. **System tools**: Add to `Dockerfile` 3. **Explain why** in your PR description 4. **Keep dependencies minimal** - avoid adding large libraries for small features - ## Useful commands ```bash @@ -234,6 +204,7 @@ docker cp aperisolve-web:/app/aperisolve/results/filename.ext /path/to/backup/fi > [!WARNING] > If switching between dev and production compose files, remove the `results` directory or mounted volume to avoid conflicts: +> > ```bash > rm -rf aperisolve/results > ``` @@ -250,6 +221,7 @@ RUN apt-get update && apt-get install -y \ ``` Or for tools requiring compilation: + ```dockerfile # Builder stage example (jphide) RUN git clone https://github.com/h3xx/jphs.git /tmp/jphs && \ diff --git a/Dockerfile b/Dockerfile index bbd483e..51294af 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,6 +43,9 @@ FROM python:3.14-slim WORKDIR /app +# Install uv binary +COPY --from=ghcr.io/astral-sh/uv:0.8.22 /uv /uvx /bin/ + # Install runtime + stego tools RUN apt-get update && apt-get install -y --no-install-recommends \ expect \ @@ -69,11 +72,12 @@ RUN dpkg -i /tmp/openstego.deb || apt-get install -f -y --no-install-recommends && rm -rf /tmp/openstego.deb \ && rm -rf /var/lib/apt/lists/* -# Copy app +# Copy Python package metadata and app +COPY pyproject.toml README.md /app/ COPY aperisolve/ /app/aperisolve/ -# Install Python dependencies -RUN pip install --no-cache-dir -r /app/aperisolve/requirements.txt +# Install Python dependencies from pyproject.toml +RUN uv pip install --system --no-cache /app # Copy jphide and jsteg binaries COPY --from=builder /usr/local/bin/jphide /usr/local/bin/jphide diff --git a/README.md b/README.md index 2464716..281c457 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,9 @@ ![Lint](https://github.com/Zeecka/AperiSolve/actions/workflows/lint.yml/badge.svg) ![Docker](https://img.shields.io/badge/docker-compose-blue?logo=docker) ![Python Version](https://img.shields.io/badge/python-3.12%2B-blue) -![Black Code Style](https://img.shields.io/badge/code%20style-black-000000.svg) +![UV](https://img.shields.io/badge/package-uv-46aef7) +![Ruff](https://img.shields.io/badge/lint-ruff-46aef7) +![ty](https://img.shields.io/badge/types-ty-5a45ff) ![License](https://img.shields.io/github/license/Zeecka/AperiSolve) ![GitHub Issues](https://img.shields.io/github/issues/Zeecka/AperiSolve) ![GitHub Pull Requests](https://img.shields.io/github/issues-pr/Zeecka/AperiSolve) diff --git a/aperisolve/__init__.py b/aperisolve/__init__.py index 2d22b7f..c49b831 100644 --- a/aperisolve/__init__.py +++ b/aperisolve/__init__.py @@ -1,8 +1,5 @@ -"""Main Aperi'Solve module""" +"""Main Aperi'Solve module.""" -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable from . import app, config, models, workers __all__ = ["app", "config", "models", "workers"] diff --git a/aperisolve/analyzers/__init__.py b/aperisolve/analyzers/__init__.py index 538f764..b6b1fa1 100644 --- a/aperisolve/analyzers/__init__.py +++ b/aperisolve/analyzers/__init__.py @@ -1,3 +1 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable +"""Analyzer implementations for AperiSolve.""" diff --git a/aperisolve/analyzers/base_analyzer.py b/aperisolve/analyzers/base_analyzer.py index 6597f99..5e64e31 100644 --- a/aperisolve/analyzers/base_analyzer.py +++ b/aperisolve/analyzers/base_analyzer.py @@ -1,19 +1,16 @@ -# 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 asyncio 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 +from subprocess import CompletedProcess +from typing import Any, overload -from ..config import MAX_PENDING_TIME +from aperisolve.config import MAX_PENDING_TIME _thread_lock = threading.Lock() @@ -30,8 +27,14 @@ class SubprocessAnalyzer(ABC): make_folder: bool = True def __init__( - self, name: str, input_img: Path, output_dir: Path, has_archive: bool = False + self, + name: str, + input_img: Path, + output_dir: Path, + *, + has_archive: bool = False, ) -> None: + """Initialize analyzer metadata and execution context.""" self.name = name self.input_img = input_img self.img = f"../{self.input_img.name}" @@ -39,19 +42,37 @@ class SubprocessAnalyzer(ABC): self.has_archive = has_archive def run_command( - self, cmd: list[str], cwd: Optional[Path] = None - ) -> subprocess.CompletedProcess[str]: + self, + cmd: list[str], + cwd: Path | None = None, + ) -> 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: Optional[Path] = None) -> str: + async def _run() -> CompletedProcess[str]: + process = await asyncio.create_subprocess_exec( + *cmd, + cwd=str(cwd) if cwd is not None else None, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout_raw, stderr_raw = await asyncio.wait_for( + process.communicate(), + timeout=MAX_PENDING_TIME, + ) + returncode = process.returncode + if returncode is None: + msg = "Subprocess exited without a return code" + raise RuntimeError(msg) + return CompletedProcess( + args=cmd, + returncode=returncode, + stdout=stdout_raw.decode("utf-8", errors="replace"), + stderr=stderr_raw.decode("utf-8", errors="replace"), + ) + + return asyncio.run(_run()) + + def generate_archive(self, _output_dir: Path, extracted_dir: Path | None = None) -> str: """Zip the extracted files and remove the directory.""" if extracted_dir is None: extracted_dir = self.get_extracted_dir() @@ -68,30 +89,29 @@ class SubprocessAnalyzer(ABC): 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 + with _thread_lock, lock_file.open("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 = {} + try: + # Read existing JSON + data: dict[Any, Any] = {} + if json_file.exists(): + try: + with json_file.open(encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError: + data = {} - # Update with new data - data.update(new_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) + # Write safely to a temp file + with tmp_file.open("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) + tmp_file.replace(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.""" @@ -103,13 +123,15 @@ class SubprocessAnalyzer(ABC): @overload def build_cmd(self, password: str) -> list[str]: ... - def build_cmd(self, password: Optional[str] = None) -> list[str]: + def build_cmd(self, password: str | None = None) -> list[str]: """Build the command to run. Can be overridden.""" + _ = password if self.cmd is None: - raise NotImplementedError("cmd must be set or build_cmd overridden") + msg = "cmd must be set or build_cmd overridden" + raise NotImplementedError(msg) return self.cmd - def get_results(self, password: Optional[str] = None) -> dict[str, Any]: + def get_results(self, password: str | None = None) -> dict[str, Any]: """Get results of command before returning.""" result: dict[str, Any] extracted_dir = None @@ -128,12 +150,12 @@ class SubprocessAnalyzer(ABC): stderr = data.stderr stdout = data.stdout - zip_exist = False + zip_exists = 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 + zip_exists = True - if self.is_error(returncode, stdout, stderr, zip_exist): + if self.is_error(returncode, stdout, stderr, zip_exist=zip_exists): result = { "status": "error", } @@ -143,10 +165,10 @@ class SubprocessAnalyzer(ABC): "status": "ok", "output": self.process_output(stdout, stderr), } - note: Optional[str] = self.process_note(stdout, stderr) + note: str | None = self.process_note(stdout, stderr) if note: result["note"] = note - if zip_exist: + if zip_exists: result["download"] = f"/download/{self.output_dir.name}/{self.name}" return result @@ -156,28 +178,32 @@ class SubprocessAnalyzer(ABC): @overload def analyze(self, password: str) -> None: ... - def analyze(self, password: Optional[str] = None) -> None: + def analyze(self, password: str | None = 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: + except (RuntimeError, ValueError, OSError, TimeoutError) as e: self.update_result({"status": "error", "error": str(e)}) raise - def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: """Check if the result is an error.""" + _ = returncode, stdout, zip_exist 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.""" + _ = stderr 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.""" + _ = stdout return stderr - def process_note(self, stdout: str, stderr: str) -> Optional[str]: + def process_note(self, stdout: str, stderr: str) -> str | None: """Process the stdout for informational purposes.""" + _ = stdout, stderr return None diff --git a/aperisolve/analyzers/binwalk.py b/aperisolve/analyzers/binwalk.py index caf755a..97aeeb7 100644 --- a/aperisolve/analyzers/binwalk.py +++ b/aperisolve/analyzers/binwalk.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Binwalk Analyzer for Image Submissions.""" from pathlib import Path @@ -12,14 +9,18 @@ class BinwalkAnalyzer(SubprocessAnalyzer): """Analyzer for binwalk.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the binwalk analyzer.""" 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 the binwalk extraction directory.""" return self.output_dir / f"_{self.input_img.name}.extracted" - def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: + """Check whether binwalk execution failed.""" + _ = returncode, stdout return len(stderr) > 0 and not zip_exist diff --git a/aperisolve/analyzers/color_remapping.py b/aperisolve/analyzers/color_remapping.py index 802e480..b26639c 100644 --- a/aperisolve/analyzers/color_remapping.py +++ b/aperisolve/analyzers/color_remapping.py @@ -1,21 +1,24 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801,W0612,R0912 -# mypy: disable-error-code=unused-awaitable """Color Remapping Analyzer for Image Submissions.""" from pathlib import Path -from typing import Any, Optional +from typing import Any import numpy as np from PIL import Image from .base_analyzer import SubprocessAnalyzer +RGB_CHANNEL_COUNT = 3 +RGBA_CHANNEL_COUNT = 4 +RANDOM_REMAPPING_COUNT = 8 +GRAYSCALE_DIMENSIONS = 2 + class ColorRemappingAnalyzer(SubprocessAnalyzer): """Analyzer for color remapping.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the color remapping analyzer.""" super().__init__("color_remapping", input_img, output_dir) def _normalize_image(self, img_np: np.ndarray) -> tuple[np.ndarray, int]: @@ -23,41 +26,46 @@ class ColorRemappingAnalyzer(SubprocessAnalyzer): Returns: Tuple of (normalized image array, number of channels) + """ - if len(img_np.shape) == 2: + if len(img_np.shape) == GRAYSCALE_DIMENSIONS: channels = 1 img_np = np.expand_dims(img_np, axis=-1) else: channels = img_np.shape[2] - if channels < 3: + if channels < RGB_CHANNEL_COUNT: # Grayscale to RGB - img_np = np.repeat(img_np, 3, axis=2) - channels = 3 - elif channels > 4: + img_np = np.repeat(img_np, RGB_CHANNEL_COUNT, axis=2) + channels = RGB_CHANNEL_COUNT + elif channels > RGBA_CHANNEL_COUNT: # More than 4 channels, take first 3 - img_np = img_np[..., :3] - channels = 3 + img_np = img_np[..., :RGB_CHANNEL_COUNT] + channels = RGB_CHANNEL_COUNT return img_np, channels def _create_remapped_image(self, img_np: np.ndarray, color_map: np.ndarray) -> Image.Image: """Create a single remapped image using the color map.""" remapped = np.zeros_like(img_np) - for c in range(min(3, img_np.shape[2])): # Apply to RGB channels + for c in range(min(RGB_CHANNEL_COUNT, img_np.shape[2])): # Apply to RGB channels remapped[..., c] = color_map[img_np[..., c]] # Keep alpha channel if present - if img_np.shape[2] == 4: + if img_np.shape[2] == RGBA_CHANNEL_COUNT: remapped[..., 3] = img_np[..., 3] remapped_img = Image.fromarray(remapped.astype(np.uint8), mode="RGBA") else: - remapped_img = Image.fromarray(remapped[..., :3].astype(np.uint8), mode="RGB") + remapped_img = Image.fromarray( + remapped[..., :RGB_CHANNEL_COUNT].astype(np.uint8), + mode="RGB", + ) return remapped_img - def get_results(self, password: Optional[str] = None) -> dict[str, Any]: + def get_results(self, password: str | None = None) -> dict[str, Any]: """Analyze an image submission using color remapping.""" + _ = password img = Image.open(self.input_img) converted = False @@ -71,9 +79,10 @@ class ColorRemappingAnalyzer(SubprocessAnalyzer): image_json = [] # Generate 8 random color remappings - for i in range(8): + rng = np.random.default_rng() + for i in range(RANDOM_REMAPPING_COUNT): # Create a random permutation of color values (0-255) - color_map = np.random.randint(0, 256, size=256, dtype=np.uint8) + color_map = rng.integers(0, 256, size=256, dtype=np.uint8) remapped_img = self._create_remapped_image(img_np, color_map) # Save the remapped image diff --git a/aperisolve/analyzers/decomposer.py b/aperisolve/analyzers/decomposer.py index a3ff02a..478755f 100644 --- a/aperisolve/analyzers/decomposer.py +++ b/aperisolve/analyzers/decomposer.py @@ -1,25 +1,27 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Bits Decomposer Analyzer for Image Submissions.""" from pathlib import Path -from typing import Any, Optional +from typing import Any import numpy as np from PIL import Image from .base_analyzer import SubprocessAnalyzer +RGB_CHANNEL_THRESHOLD = 3 +GRAYSCALE_DIMENSIONS = 2 + class DecomposerAnalyzer(SubprocessAnalyzer): """Analyzer for bits decomposer.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the decomposer analyzer.""" super().__init__("decomposer", input_img, output_dir) - def get_results(self, _: Optional[str] = None) -> dict[str, Any]: + def get_results(self, password: str | None = None) -> dict[str, Any]: """Analyze an image submission using bits decomposition.""" + _ = password img = Image.open(self.input_img) converted = False @@ -30,20 +32,14 @@ class DecomposerAnalyzer(SubprocessAnalyzer): img_np = np.array(img) # Handle grayscale, RGB, RGBA, etc. - if len(img_np.shape) == 2: - channels = 1 - else: - channels = img_np.shape[2] + channels = 1 if len(img_np.shape) == GRAYSCALE_DIMENSIONS else img_np.shape[2] - if channels > 1: - channel_names = ["Red", "Green", "Blue", "Alpha"] - else: - channel_names = ["Grayscale"] + channel_names = ["Red", "Green", "Blue", "Alpha"] if channels > 1 else ["Grayscale"] image_json = {} # Add Superimposed RGB bit planes - if channels >= 3: + if channels >= RGB_CHANNEL_THRESHOLD: superimposed_json = [] for bit in range(8): bit_mask = 1 << bit diff --git a/aperisolve/analyzers/exiftool.py b/aperisolve/analyzers/exiftool.py index d35d216..3c99b88 100644 --- a/aperisolve/analyzers/exiftool.py +++ b/aperisolve/analyzers/exiftool.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Exiftool Analyzer for Image Submissions.""" from pathlib import Path @@ -12,11 +9,13 @@ class ExiftoolAnalyzer(SubprocessAnalyzer): """Analyzer for exiftool.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the exiftool analyzer.""" 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.""" + _ = stderr metadata: dict[str, str] = {} for line in stdout.split("\n"): if ":" in line: diff --git a/aperisolve/analyzers/file.py b/aperisolve/analyzers/file.py index bb194c0..6c6735c 100644 --- a/aperisolve/analyzers/file.py +++ b/aperisolve/analyzers/file.py @@ -1,7 +1,5 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """File Analyzer for Image Submissions.""" + from pathlib import Path from .base_analyzer import SubprocessAnalyzer @@ -11,11 +9,13 @@ class FileAnalyzer(SubprocessAnalyzer): """Analyzer for the `file` command.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the file analyzer.""" super().__init__("file", input_img, output_dir) self.cmd = ["file", "-b", self.img] def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]: """Process the stdout into a list of lines.""" + _ = stderr return stdout diff --git a/aperisolve/analyzers/foremost.py b/aperisolve/analyzers/foremost.py index 2c71322..3feabcc 100644 --- a/aperisolve/analyzers/foremost.py +++ b/aperisolve/analyzers/foremost.py @@ -1,26 +1,28 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Foremost Analyzer for Image Submissions.""" from pathlib import Path from .base_analyzer import SubprocessAnalyzer +FOREMOST_ERROR_THRESHOLD = 60 + class ForemostAnalyzer(SubprocessAnalyzer): """Analyzer for foremost.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the foremost analyzer.""" 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: + 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 + _ = stdout, returncode, zip_exist + return len(stderr) > FOREMOST_ERROR_THRESHOLD def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]: """Process the stdout/stderr.""" + _ = stdout if "Processing" in stderr and "|*|" in stderr: return stderr.strip().replace("\n", "") return [] diff --git a/aperisolve/analyzers/identify.py b/aperisolve/analyzers/identify.py index 005e379..38accec 100644 --- a/aperisolve/analyzers/identify.py +++ b/aperisolve/analyzers/identify.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Identify Analyzer for Image Submissions.""" from pathlib import Path @@ -12,11 +9,13 @@ class IdentifyAnalyzer(SubprocessAnalyzer): """Analyzer for GraphicMagick identify command.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the identify analyzer.""" super().__init__("identify", input_img, output_dir) self.cmd = ["identify", "-verbose", self.img] def process_output(self, stdout: str, stderr: str) -> str | list[str]: """Process the stdout into a list of lines.""" + _ = stderr return [line for line in stdout.split("\n") if line.strip()] diff --git a/aperisolve/analyzers/jpseek.py b/aperisolve/analyzers/jpseek.py index fb4ec96..b5f30d3 100644 --- a/aperisolve/analyzers/jpseek.py +++ b/aperisolve/analyzers/jpseek.py @@ -1,10 +1,6 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """JPSeek Analyzer for Image Submissions.""" from pathlib import Path -from typing import Optional from .base_analyzer import SubprocessAnalyzer @@ -13,11 +9,12 @@ class JpseekAnalyzer(SubprocessAnalyzer): """Analyzer for jpseek.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the jpseek analyzer.""" super().__init__("jpseek", input_img, output_dir, has_archive=True) - def build_cmd(self, password: Optional[str] = None) -> list[str]: + def build_cmd(self, password: str | None = None) -> list[str]: + """Build the jpseek command wrapped with expect for password support.""" extracted_dir = self.get_extracted_dir() - assert extracted_dir is not None # since has_archive is True out = str(extracted_dir / "jpseek.out") jpseek_cmd = f"jpseek {self.img} {out}" if password is None: @@ -28,9 +25,7 @@ class JpseekAnalyzer(SubprocessAnalyzer): f'send "{password}\\r"; ' f"expect eof'" ) - command = ["bash", "-c", expect_cmd] - print("Built command:", command) - return command + return ["bash", "-c", expect_cmd] def _remove_output_artifacts(self, output: str) -> str: extracted_dir = self.get_extracted_dir() @@ -46,15 +41,15 @@ class JpseekAnalyzer(SubprocessAnalyzer): o = o.replace(f"spawn jpseek {self.img} {out_file}", "") return o.strip() - def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: """Check if the result is an error.""" - return bool(returncode) and "File not completely recovered" not in "".join([stdout, stderr]) + _ = zip_exist + return bool(returncode) and "File not completely recovered" not in f"{stdout}{stderr}" def process_error(self, stdout: str, stderr: str) -> str: """Process stderr.""" output = stdout + stderr - output = self._remove_output_artifacts(output) - return output + return self._remove_output_artifacts(output) def process_output(self, stdout: str, stderr: str) -> str: """Process stdout.""" @@ -65,7 +60,7 @@ class JpseekAnalyzer(SubprocessAnalyzer): return output -def analyze_jpseek(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: +def analyze_jpseek(input_img: Path, output_dir: Path, password: str | None = None) -> None: """Analyze an image submission using jpseek.""" analyzer = JpseekAnalyzer(input_img, output_dir) if password: diff --git a/aperisolve/analyzers/jsteg.py b/aperisolve/analyzers/jsteg.py index 98eb1ac..f8b48c9 100644 --- a/aperisolve/analyzers/jsteg.py +++ b/aperisolve/analyzers/jsteg.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Jsteg Analyzer for Image Submissions.""" from pathlib import Path @@ -12,11 +9,13 @@ class JstegAnalyzer(SubprocessAnalyzer): """Analyzer for jsteg.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the jsteg analyzer.""" super().__init__("jsteg", input_img, output_dir) self.cmd = ["jsteg", "reveal", self.img] def process_output(self, stdout: str, stderr: str) -> str | list[str]: """Process the stdout into lines.""" + _ = stderr if stdout.strip(): return [line for line in stdout.split("\n") if line.strip()] return [] diff --git a/aperisolve/analyzers/openstego.py b/aperisolve/analyzers/openstego.py index 559482a..f601541 100644 --- a/aperisolve/analyzers/openstego.py +++ b/aperisolve/analyzers/openstego.py @@ -1,10 +1,7 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """OpenStego Analyzer for Image Submissions.""" from pathlib import Path -from typing import Any, Optional +from typing import Any from .base_analyzer import SubprocessAnalyzer @@ -13,11 +10,12 @@ class OpenStegoAnalyzer(SubprocessAnalyzer): """Analyzer for openstego.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the OpenStego analyzer.""" 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""" + def build_cmd(self, password: str | None = None) -> list[str]: + """Return an OpenStego extraction command for the current algorithm.""" if password is None: password = "" @@ -28,13 +26,12 @@ class OpenStegoAnalyzer(SubprocessAnalyzer): 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: + 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 - ) + _ = returncode, stdout + return not zip_exist and "Extracted file: " not in stderr - def analyze(self, password: Optional[str] = None) -> None: + def analyze(self, password: str | None = None) -> None: """Run the subprocess command and handle results.""" result: dict[str, Any] try: @@ -42,16 +39,18 @@ class OpenStegoAnalyzer(SubprocessAnalyzer): if result["status"] == "error": # check second algorithm if failed result = self.get_results(password) self.update_result(result) - except Exception as e: + except (RuntimeError, ValueError, OSError, TimeoutError) as e: self.update_result({"status": "error", "error": str(e)}) raise def process_output(self, stdout: str, stderr: str) -> str | list[str] | dict[str, str]: """Process the stdout/stderr.""" + _ = stdout return stderr def process_error(self, stdout: str, stderr: str) -> str: """Process the stderr.""" + _ = stdout if "OpenStego is a steganography application" in stderr: return "OpenStego needs a password to work." return stderr diff --git a/aperisolve/analyzers/outguess.py b/aperisolve/analyzers/outguess.py index e99d996..4cc4fde 100644 --- a/aperisolve/analyzers/outguess.py +++ b/aperisolve/analyzers/outguess.py @@ -1,10 +1,6 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Outguess Analyzer for Image Submissions.""" from pathlib import Path -from typing import Optional from .base_analyzer import SubprocessAnalyzer @@ -13,18 +9,19 @@ class OutguessAnalyzer(SubprocessAnalyzer): """Analyzer for outguess.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the outguess analyzer.""" super().__init__("outguess", input_img, output_dir, has_archive=True) - def build_cmd(self, password: Optional[str] = None) -> list[str]: + def build_cmd(self, password: str | None = None) -> list[str]: + """Build the outguess extraction command.""" 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: return ["outguess", "-k", password, "-r", self.img, out] return ["outguess", "-r", self.img, out] -def analyze_outguess(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: +def analyze_outguess(input_img: Path, output_dir: Path, password: str | None = None) -> None: """Analyze an image submission using outguess.""" analyzer = OutguessAnalyzer(input_img, output_dir) if password: diff --git a/aperisolve/analyzers/pcrt.py b/aperisolve/analyzers/pcrt.py index 10a7f37..c1cdcd4 100644 --- a/aperisolve/analyzers/pcrt.py +++ b/aperisolve/analyzers/pcrt.py @@ -1,12 +1,10 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0902,R0801 -# mypy: disable-error-code=unused-awaitable """PCRT (PNG Check & Repair Tool) Analyzer for Image Submissions.""" from pathlib import Path -from typing import Any, Optional +from typing import Any + +from aperisolve.utils.png import PNG -from ..utils.png import PNG from .base_analyzer import SubprocessAnalyzer @@ -14,6 +12,7 @@ class PCRTAnalyzer(SubprocessAnalyzer): """Analyzer for PCRT (PNG Check & Repair Tool).""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the PCRT analyzer.""" super().__init__("pcrt", input_img, output_dir, has_archive=True) def _write_repaired_data(self, data: bytes) -> str: @@ -33,8 +32,9 @@ class PCRTAnalyzer(SubprocessAnalyzer): with extra_path.open("wb") as f: f.write(data) - def get_results(self, password: Optional[str] = None) -> dict[str, Any]: + def get_results(self, password: str | None = None) -> dict[str, Any]: """Analyze PNG and attempt repairs.""" + _ = password try: with self.input_img.open("rb") as f: data = f.read() @@ -52,11 +52,11 @@ class PCRTAnalyzer(SubprocessAnalyzer): return result result["status"] = "ok" - result["output"] = png.logs if png.logs else ["PNG appears valid, no repairs needed"] + result["output"] = png.logs or ["PNG appears valid, no repairs needed"] # Save repaired PNG if any fixes were made if fixed and png.repaired_data: - url = self._write_repaired_data(png.repaired_data) + url = self._write_repaired_data(bytes(png.repaired_data)) result["note"] = "PNG was repaired and saved" result["png_images"] = [url] @@ -67,11 +67,11 @@ class PCRTAnalyzer(SubprocessAnalyzer): result["note"] = result.get("note", "") + " | Extra data found after IEND" result["download"] = f"/download/{self.output_dir.name}/{self.name}" + except (ValueError, OSError, RuntimeError, TypeError) as e: + return {"status": "error", "error": f"Analysis failed: {e!s}"} + else: return result - except Exception as e: - return {"status": "error", "error": f"Analysis failed: {str(e)}"} - def analyze_pcrt(input_img: Path, output_dir: Path) -> None: """Analyze an image submission using PCRT.""" diff --git a/aperisolve/analyzers/pngcheck.py b/aperisolve/analyzers/pngcheck.py index 530c5d6..69b0165 100644 --- a/aperisolve/analyzers/pngcheck.py +++ b/aperisolve/analyzers/pngcheck.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Pngcheck Analyzer for Image Submissions.""" from pathlib import Path @@ -12,14 +9,18 @@ class PngcheckAnalyzer(SubprocessAnalyzer): """Analyzer for pngcheck.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the pngcheck analyzer.""" 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: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: + """Check whether the file is an unsupported format.""" + _ = returncode, stderr, zip_exist 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.""" + _ = 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 diff --git a/aperisolve/analyzers/steghide.py b/aperisolve/analyzers/steghide.py index c1785c2..536d54c 100644 --- a/aperisolve/analyzers/steghide.py +++ b/aperisolve/analyzers/steghide.py @@ -1,11 +1,7 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Steghide Analyzer for Image Submissions.""" import re from pathlib import Path -from typing import Optional from .base_analyzer import SubprocessAnalyzer @@ -14,30 +10,32 @@ class SteghideAnalyzer(SubprocessAnalyzer): """Analyzer for steghide.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the steghide analyzer.""" super().__init__("steghide", input_img, output_dir, has_archive=True) - def build_cmd(self, password: Optional[str] = None) -> list[str]: + def build_cmd(self, password: str | None = None) -> list[str]: + """Build the steghide command for info or extraction.""" 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) + err = self.is_error(data.returncode, data.stdout, data.stderr, zip_exist=False) if err: return cmd_test match = re.search(r'embedded file "([^"]+)"', data.stdout) - assert match is not None + if match is None: + return cmd_test 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 + return ["steghide", "extract", "-sf", self.img, "-xf", outfile, "-p", password] - def is_error(self, returncode: int, stdout: str, stderr: str, zip_exist: bool) -> bool: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: """Check if the result is an error.""" - + _ = returncode, zip_exist match = re.search(r'embedded file "([^"]+)".*', stdout + stderr) embedded_filename = match.group(1) if match else None @@ -49,20 +47,18 @@ class SteghideAnalyzer(SubprocessAnalyzer): 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 + _ = stdout + return [line for line in stderr.split("\n") if "wrote extracted data to" in line] def process_error(self, stdout: str, stderr: str) -> str: """Process stderr.""" + _ = stdout 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 -def analyze_steghide(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: +def analyze_steghide(input_img: Path, output_dir: Path, password: str | None = None) -> None: """Analyze an image submission using steghide.""" analyzer = SteghideAnalyzer(input_img, output_dir) if password: diff --git a/aperisolve/analyzers/strings.py b/aperisolve/analyzers/strings.py index 4d25a58..27765b1 100644 --- a/aperisolve/analyzers/strings.py +++ b/aperisolve/analyzers/strings.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Strings Analyzer for Image Submissions.""" from pathlib import Path @@ -12,6 +9,7 @@ class StringsAnalyzer(SubprocessAnalyzer): """Analyzer for strings.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the strings analyzer.""" super().__init__("strings", input_img, output_dir) self.cmd = ["strings", self.img] diff --git a/aperisolve/analyzers/template_analyzer.py b/aperisolve/analyzers/template_analyzer.py index cb5306a..fe23a18 100644 --- a/aperisolve/analyzers/template_analyzer.py +++ b/aperisolve/analyzers/template_analyzer.py @@ -1,10 +1,6 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """ Analyzer for Image Submissions.""" from pathlib import Path -from typing import Optional from .base_analyzer import SubprocessAnalyzer @@ -13,15 +9,17 @@ class TemplateAnalyzer(SubprocessAnalyzer): """Analyzer for .""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the template analyzer.""" super().__init__("", input_img, output_dir, has_archive=True) - def build_cmd(self, password: Optional[str] = None) -> list[str]: + def build_cmd(self, password: str | None = None) -> list[str]: + """Build command for a placeholder analyzer implementation.""" if password: return ["", "-p", password, "-i", self.img] return ["", "-i", self.img] -def analyze_template(input_img: Path, output_dir: Path, password: Optional[str] = None) -> None: +def analyze_template(input_img: Path, output_dir: Path, password: str | None = None) -> None: """Analyze an image submission using .""" analyzer = TemplateAnalyzer(input_img, output_dir) if password: diff --git a/aperisolve/analyzers/zsteg.py b/aperisolve/analyzers/zsteg.py index df67ef4..3be5ff5 100644 --- a/aperisolve/analyzers/zsteg.py +++ b/aperisolve/analyzers/zsteg.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Zsteg Analyzer for Image Submissions.""" from pathlib import Path @@ -12,10 +9,13 @@ class ZstegAnalyzer(SubprocessAnalyzer): """Analyzer for zsteg.""" def __init__(self, input_img: Path, output_dir: Path) -> None: + """Initialize the zsteg analyzer.""" 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: + def is_error(self, returncode: int, stdout: str, stderr: str, *, zip_exist: bool) -> bool: + """Check whether zsteg reported an error.""" + _ = returncode, zip_exist return bool(stderr) or "PNG::NotSupported" in stdout[:100] def process_error(self, stdout: str, stderr: str) -> str: diff --git a/aperisolve/app.py b/aperisolve/app.py index b66a161..d574b9e 100644 --- a/aperisolve/app.py +++ b/aperisolve/app.py @@ -1,24 +1,19 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801,R0915 -# mypy: disable-error-code=unused-awaitable """Aperi'Solve Flask application.""" import hashlib import json +import os import shutil import time -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path -from typing import Any, Optional +from typing import Any, cast import sentry_sdk from flask import Flask, Response, abort, jsonify, render_template, request, send_file from redis import Redis from rq import Queue - -from .utils.sentry import initialize_sentry - -initialize_sentry() +from sqlalchemy.exc import SQLAlchemyError from .config import ( CUSTOM_EXTERNAL_SCRIPT, @@ -34,12 +29,12 @@ from .config import ( WORKER_FILES, ) from .models import Image, Submission, UploadLog, cleanup_old_entries, db +from .utils.sentry import initialize_sentry -def create_app() -> Flask: - """Create flask application with routes.""" - app = Flask(__name__) - app.json.sort_keys = False # type: ignore +def _configure_app(app: Flask) -> None: + """Configure Flask and extension settings.""" + app.config["JSON_SORT_KEYS"] = False app.config["PROJECT_VERSION"] = PROJECT_VERSION app.config["CUSTOM_EXTERNAL_SCRIPT"] = CUSTOM_EXTERNAL_SCRIPT app.config["SQLALCHEMY_DATABASE_URI"] = DB_URI @@ -48,14 +43,59 @@ def create_app() -> Flask: app.config["REDIS_QUEUE"] = Queue(connection=Redis(host="redis", port=6379)) db.init_app(app) + +def _extract_unique_ips(upload_logs: list[UploadLog]) -> set[str]: + """Return unique non-empty uploader IP addresses from logs.""" + return {str(log.ip_address) for log in upload_logs if log.ip_address} + + +def _archive_original_image( + original_image_path: Path, + image_hash: str, + submission_hash: str, +) -> None: + """Archive original image file before deletion.""" + if not original_image_path.exists(): + return + REMOVED_IMAGES_FOLDER.mkdir(parents=True, exist_ok=True) + dt = datetime.now(UTC).isoformat() + archive_filename = f"{image_hash}_{submission_hash}_{dt}{original_image_path.suffix}" + archive_path = REMOVED_IMAGES_FOLDER / archive_filename + shutil.copy2(original_image_path, archive_path) + + +def _remove_result_folder(image_hash: str, submission_hash: str) -> None: + """Delete result folder for a submission if it exists.""" + result_path = RESULT_FOLDER / image_hash / submission_hash + if result_path.exists(): + shutil.rmtree(result_path) + + +def _delete_submission_entities( + submission: Submission, + image: Image, + original_image_path: Path, +) -> None: + """Delete submission and related image entity if applicable.""" + db.session.delete(submission) + if len(image.submissions) <= 1: + if original_image_path.exists(): + original_image_path.unlink() + db.session.delete(image) + db.session.commit() + + +def _register_error_handlers(app: Flask) -> None: + """Register application-wide error handlers.""" + @app.context_processor - def inject_datetime() -> dict[str, Any]: + def inject_datetime() -> dict[str, type[datetime]]: """Inject datetime into all templates.""" return {"datetime": datetime} @app.errorhandler(413) - def too_large(_: Any) -> tuple[Response, int]: - """Error Handler for Max File Size.""" + def too_large(_: object) -> tuple[Response, int]: + """Handle max upload size errors.""" sentry_sdk.set_context( "upload", { @@ -67,10 +107,14 @@ def create_app() -> Flask: return jsonify({"error": "Image size exceeded"}), 413 @app.errorhandler(404) - def not_found(_: Any) -> str: - """Error Handler for 404 not found.""" + def not_found(_: object) -> str: + """Handle not found errors.""" return render_template("error.html", message="Resource not found", statuscode=404) + +def _register_page_routes(app: Flask) -> None: + """Register static page routes.""" + @app.route("/") def index() -> str: """Render the main index page.""" @@ -94,139 +138,138 @@ def create_app() -> Flask: @app.route("/", methods=["GET"]) def result_page(hash_val: str) -> str: """Render the result page for a given submission hash.""" - submission = Submission.query.get_or_404(hash_val) # type: ignore - Image.query.get_or_404(submission.image_hash) # type: ignore - + submission = Submission.query.get_or_404(hash_val) + Image.query.get_or_404(submission.image_hash) return render_template("result.html", hash_val=hash_val) + @app.route("/ads.txt") + def google_ads() -> str: + """Serve the Google Ads required file.""" + return GOOGLE_ADS_TXT + + +def _upload_image(app: Flask) -> tuple[Response, int]: + """Handle image upload and initiate analysis.""" + cleanup_old_entries() + + if "image" not in request.files: + return jsonify({"error": "No image provided"}), 400 + + image = request.files["image"] + password = request.form.get("password") + deep_analysis = request.form.get("deep") == "true" + + 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: + return jsonify({"error": "Unsupported file type"}), 400 + + image_data = image.read() + img_hash = hashlib.md5(image_data, usedforsecurity=False).hexdigest() + image_name = img_hash + "." + image.filename.rsplit(".", maxsplit=1)[-1].lower() + + submission_data = ( + image_data + + image.filename.encode() + + (password.encode() if password else b"") + + (b"deep_analysis" if deep_analysis else b"") + ) + submission_hash = hashlib.md5(submission_data, usedforsecurity=False).hexdigest() + submission_path = RESULT_FOLDER / img_hash / submission_hash + + client_ip = request.headers.get("X-Forwarded-For", request.remote_addr) + if client_ip and "," in client_ip: + client_ip = client_ip.split(",")[0].strip() + user_agent = request.headers.get("User-Agent", "Unknown") + + upload_log = UploadLog( + ip_address=client_ip, + user_agent=user_agent, + image_hash=img_hash, + submission_hash=submission_hash, + filename=image.filename, + ) + db.session.add(upload_log) + db.session.commit() + + submission = Submission.query.filter_by(hash=submission_hash).first() + if submission_path.exists() and submission is not None: + return jsonify({"submission_hash": submission.hash}), 200 + + new_img_path = RESULT_FOLDER / img_hash / image_name + if not new_img_path.exists(): + new_img_path.parent.mkdir(parents=True, exist_ok=True) + with new_img_path.open("wb") as file_obj: + file_obj.write(image_data) + + sub_img = Image.query.filter_by(hash=img_hash).first() + if sub_img is None: + sub_img = Image( + file=str(new_img_path), + hash=img_hash, + size=len(image_data), + upload_count=0, + first_submission_date=datetime.now(UTC), + last_submission_date=datetime.now(UTC), + ) + db.session.add(sub_img) + db.session.commit() + + sub_img_any = cast("Any", sub_img) + sub_img_any.upload_count = int(sub_img_any.upload_count or 0) + 1 + db.session.commit() + + submission_path.mkdir(parents=True, exist_ok=True) + if not submission: + submission = Submission( + filename=image.filename, + password=password, + deep_analysis=deep_analysis, + hash=submission_hash, + status="pending", + date=time.time(), + image_hash=sub_img.hash, + ) + db.session.add(submission) + db.session.commit() + + submission_any = cast("Any", submission) + submission_any.status = "pending" + db.session.commit() + + app.config["REDIS_QUEUE"].enqueue( + "aperisolve.workers.analyze_image", + submission.hash, + job_timeout=300, + ) + return jsonify({"submission_hash": submission.hash}), 200 + + +def _register_submission_routes(app: Flask) -> None: + """Register upload and status routes.""" + @app.route("/upload", methods=["POST"]) def upload_image() -> tuple[Response, int]: - """Handle image upload and initiate analysis.""" - - cleanup_old_entries() - - if "image" not in request.files: - return jsonify({"error": "No image provided"}), 400 - - image = request.files["image"] - password = request.form.get("password") - deep_analysis = request.form.get("deep") == "true" - - 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: - return jsonify({"error": "Unsupported file type"}), 400 - - # Compute hashes and prepare submission - image_data = image.read() - img_hash = hashlib.md5(image_data).hexdigest() - image_name = img_hash + "." + image.filename.rsplit(".", maxsplit=1)[-1].lower() - - submission_data = ( - image_data - + image.filename.encode() - + (password.encode() if password else b"") - + (b"deep_analysis" if deep_analysis else b"") - ) - submission_hash = hashlib.md5(submission_data).hexdigest() - submission_path = RESULT_FOLDER / img_hash / submission_hash - - # Log upload to database - client_ip = request.headers.get("X-Forwarded-For", request.remote_addr) - if client_ip and "," in client_ip: - client_ip = client_ip.split(",")[0].strip() - user_agent = request.headers.get("User-Agent", "Unknown") - - upload_log = UploadLog( - ip_address=client_ip, - user_agent=user_agent, - image_hash=img_hash, - submission_hash=submission_hash, - filename=image.filename, - ) - db.session.add(upload_log) # pylint: disable=no-member - db.session.commit() # pylint: disable=no-member - - # Check if hash is present on the DB - submission: Submission = Submission.query.filter_by(hash=submission_hash).first() - - # Only return if BOTH the file exists AND the DB entry exists - if submission_path.exists() and submission is not None: - return jsonify({"submission_hash": submission.hash}), 200 - - # Self-Healing Logic - new_img_path = RESULT_FOLDER / img_hash / image_name - - # This fixes the "Ghost File" issue (DB exists, but file is missing) - if not new_img_path.exists(): - new_img_path.parent.mkdir(parents=True, exist_ok=True) - with open(new_img_path, "wb") as f: - f.write(image_data) - - # Try to find image in DB by hash - sub_img = Image.query.filter_by(hash=img_hash).first() - - # If DB entry is missing (even if file exists on disk), create it! - if sub_img is None: - sub_img = Image( - file=str(new_img_path), - hash=img_hash, - size=len(image_data), - upload_count=0, - first_submission_date=datetime.now(timezone.utc), - last_submission_date=datetime.now(timezone.utc), - ) - db.session.add(sub_img) # pylint: disable=no-member - db.session.commit() # pylint: disable=no-member - - # Increment upload count for the image - sub_img.upload_count += 1 - db.session.commit() # pylint: disable=no-member - - # Create new Submission entry - submission_path.mkdir(parents=True, exist_ok=True) - - # Only create a new submission if it does not already exist - # New submission for new image - if not submission: - submission = Submission( - filename=image.filename, - password=password, - deep_analysis=deep_analysis, - hash=submission_hash, - status="pending", - date=time.time(), - image_hash=sub_img.hash, - ) - db.session.add(submission) # pylint: disable=no-member - db.session.commit() # pylint: disable=no-member - - # Re-analysis case to prevent early return - submission.status = "pending" # type: ignore - db.session.commit() # pylint: disable=no-member - - # Start the analysis jobs - app.config["REDIS_QUEUE"].enqueue( - "aperisolve.workers.analyze_image", submission.hash, job_timeout=300 - ) - - # Return submission identifier - return jsonify({"submission_hash": submission.hash}), 200 + return _upload_image(app) @app.route("/status/", methods=["GET"]) def get_status(hash_val: str) -> Response: """Get the processing status of a submission.""" - submission = Submission.query.get_or_404(hash_val) # type: ignore - return jsonify({"status": submission.status}) # type: ignore + submission = Submission.query.get_or_404(hash_val) + return jsonify({"status": submission.status}) + + +def _register_data_routes(app: Flask) -> None: + """Register metadata, results, and download routes.""" @app.route("/infos/", methods=["GET"]) def get_infos(hash_val: str) -> Response: """Get the metadata and information of a submission.""" submission = Submission.query.get_or_404(hash_val) - image = Image.query.get_or_404(submission.image_hash) # type: ignore - names = [name for name in set(sub.filename for sub in image.submissions) if name] - passwords = [pwd for pwd in set(sub.password for sub in image.submissions) if pwd] + image = Image.query.get_or_404(submission.image_hash) + names = [name for name in {sub.filename for sub in image.submissions} if name] + passwords = [pwd for pwd in {sub.password for sub in image.submissions} if pwd] return jsonify( { "image_path": "image/" + str(Path(image.file).name), @@ -238,32 +281,29 @@ def create_app() -> Flask: "passwords": passwords, "removal_min_age_seconds": REMOVAL_MIN_AGE_SECONDS, "submission_date": submission.date, - } - ) # type: ignore + }, + ) @app.route("/result/", methods=["GET"]) def get_result(hash_val: str) -> tuple[Response, int]: """Get the analysis results of a submission.""" submission = Submission.query.get_or_404(hash_val) - image = Image.query.get_or_404(submission.image_hash) # type: ignore - + image = Image.query.get_or_404(submission.image_hash) 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 - with open(result_path, "r", encoding="utf-8") as f: - results = json.load(f) + with result_path.open(encoding="utf-8") as file_obj: + results = json.load(file_obj) return jsonify({"results": results}), 200 @app.route("/download//") def download_output(hash_val: str, tool: str) -> Response: - """Download the output of a specific analyzer for a given submission hash.""" - + """Download output of a specific analyzer for a submission hash.""" submission = Submission.query.filter_by(hash=hash_val).first_or_404() - image = Image.query.get_or_404(submission.image_hash) # type: ignore - + image = Image.query.get_or_404(submission.image_hash) output_dir = RESULT_FOLDER / str(image.hash) / str(submission.hash) output_file = output_dir / f"{tool}.7z" @@ -272,35 +312,55 @@ def create_app() -> Flask: return send_file(output_file, as_attachment=True) + @app.route("/image/") + @app.route("/image//") + def get_image(hash_val: str | None = None, img_name: str | None = None) -> Response: + """Download an image for a submission hash or by direct image filename. + + Usually this serves derived images generated by the decomposer analyzer. + """ + if img_name is None: + return abort(404, description="Image not found or unsupported format") + + if hash_val is not None: + submission = Submission.query.filter_by(hash=hash_val).first_or_404() + image = Image.query.get_or_404(submission.image_hash) + output_dir = RESULT_FOLDER / str(image.hash) / str(submission.hash) + output_file = output_dir / Path(img_name).name + else: + hash_val = img_name.split(".")[0] + image = Image.query.filter_by(hash=hash_val).first_or_404() + output_file = Path(image.file) + + 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) + + +def _register_management_routes(app: Flask) -> None: + """Register removal and moderation routes.""" + @app.route("/remove/", methods=["POST"]) def remove_image(hash_val: str) -> tuple[Response, int]: """Remove an image and associated results if criteria are met.""" + submission = Submission.query.get_or_404(hash_val) + image = Image.query.get_or_404(submission.image_hash) - submission = Submission.query.get_or_404(hash_val) # type: ignore - image = Image.query.get_or_404(submission.image_hash) # type: ignore - - # Check if image is at least 5 minutes old - current_time = time.time() - age_seconds = current_time - submission.date + age_seconds = time.time() - submission.date if age_seconds < REMOVAL_MIN_AGE_SECONDS: return ( jsonify( { "error": "Image must be at least 5 minutes old. " - f"Current age: {int(age_seconds)}s" - } + f"Current age: {int(age_seconds)}s", + }, ), 403, ) - # Get all IPs that uploaded this image upload_logs = UploadLog.query.filter_by(image_hash=image.hash).all() - unique_ips = set() - for log in upload_logs: - if log.ip_address: - unique_ips.add(log.ip_address) - - # Check if only one IP has uploaded this image + unique_ips = _extract_unique_ips(upload_logs) if len(unique_ips) > 1: return ( jsonify( @@ -308,87 +368,45 @@ def create_app() -> Flask: "error": "Image uploaded from multiple IP addresses. Removal is not " "allowed.", "ip_count": len(unique_ips), - } + }, ), 403, ) - # Create archive directory if it doesn't exist - REMOVED_IMAGES_FOLDER.mkdir(parents=True, exist_ok=True) - - # Archive the original image (copy to removed_images folder with metadata) original_image_path = Path(image.file) - if original_image_path.exists(): - dt = datetime.now(timezone.utc).isoformat() - archive_filename = f"{image.hash}_{submission.hash}_{dt}{original_image_path.suffix}" - archive_path = REMOVED_IMAGES_FOLDER / archive_filename - try: - shutil.copy2(original_image_path, archive_path) - except Exception as e: - sentry_sdk.capture_exception(e) - return jsonify({"error": "Failed to archive image"}), 500 - - # Delete the results folder - result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) try: - if result_path.exists(): - shutil.rmtree(result_path) - except Exception as e: - sentry_sdk.capture_exception(e) - return jsonify({"error": "Failed to delete results"}), 500 - - # Delete submission from database - try: - db.session.delete(submission) # pylint: disable=no-member - - # If this was the last submission for this image, delete the image entry too - if len(image.submissions) <= 1: - # Also delete the original image file - if original_image_path.exists(): - original_image_path.unlink() - db.session.delete(image) # pylint: disable=no-member - - db.session.commit() # pylint: disable=no-member - except Exception as e: - db.session.rollback() # pylint: disable=no-member - sentry_sdk.capture_exception(e) - return jsonify({"error": "Failed to update database"}), 500 - - # Note: UploadLog entries are intentionally kept for audit purposes + _archive_original_image(original_image_path, image.hash, submission.hash) + _remove_result_folder(str(image.hash), str(submission.hash)) + _delete_submission_entities(submission, image, original_image_path) + except (OSError, SQLAlchemyError) as exc: + db.session.rollback() + sentry_sdk.capture_exception(exc) + return jsonify({"error": "Failed to remove image"}), 500 return jsonify({"message": "Image successfully removed"}), 200 @app.route("/remove_password/", methods=["POST"]) def remove_password(hash_val: str) -> tuple[Response, int]: """Remove a password from a submission if criteria are met.""" - - submission = Submission.query.get_or_404(hash_val) # type: ignore + submission = Submission.query.get_or_404(hash_val) if not submission.password: return jsonify({"error": "No password to remove"}), 400 - # Check if submission is at least 5 minutes old - current_time = time.time() - age_seconds = current_time - submission.date + age_seconds = time.time() - submission.date if age_seconds < REMOVAL_MIN_AGE_SECONDS: return ( jsonify( { - "error": f"Submission must be at least 5 minutes old. " - f"Current age: {int(age_seconds)}s" - } + "error": "Submission must be at least 5 minutes old. " + f"Current age: {int(age_seconds)}s", + }, ), 403, ) - # Get all IPs that uploaded this specific submission (password combination) upload_logs = UploadLog.query.filter_by(submission_hash=submission.hash).all() - unique_ips = set() - for log in upload_logs: - if log.ip_address: - unique_ips.add(log.ip_address) - - # Check if only one IP has uploaded this submission + unique_ips = _extract_unique_ips(upload_logs) if len(unique_ips) > 1: return ( jsonify( @@ -396,57 +414,36 @@ def create_app() -> Flask: "error": "Submission uploaded from multiple IP addresses. " "Password removal is not allowed.", "ip_count": len(unique_ips), - } + }, ), 403, ) - # Remove the password try: submission.password = None - db.session.commit() # pylint: disable=no-member - except Exception as e: - db.session.rollback() # pylint: disable=no-member - sentry_sdk.capture_exception(e) + db.session.commit() + except SQLAlchemyError as exc: + db.session.rollback() + sentry_sdk.capture_exception(exc) return jsonify({"error": "Failed to remove password"}), 500 return jsonify({"message": "Password successfully removed"}), 200 - @app.route("/image/") - @app.route("/image//") - 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. - """ - - if img_name is None: - return abort(404, description="Image not found or unsupported format") - - if hash_val is not None: - submission = Submission.query.filter_by(hash=hash_val).first_or_404() - image = Image.query.get_or_404(submission.image_hash) # type: ignore - - output_dir = RESULT_FOLDER / str(image.hash) / str(submission.hash) - output_file = output_dir / Path(img_name).name - else: - hash_val = img_name.split(".")[0] # Extract hash from the image name - 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): - return abort(404, description="Image not found or unsupported format") - - return send_file(output_file, as_attachment=True) - - @app.route("/ads.txt") - def google_ads() -> str: - """Google Ads required file""" - return GOOGLE_ADS_TXT +def create_app() -> Flask: + """Create flask application with routes.""" + initialize_sentry() + app = Flask(__name__) + _configure_app(app) + _register_error_handlers(app) + _register_page_routes(app) + _register_submission_routes(app) + _register_data_routes(app) + _register_management_routes(app) return app if __name__ == "__main__": my_app = create_app() - my_app.run(host="0.0.0.0", port=5000, debug=FLASK_DEBUG) + flask_host = os.getenv("FLASK_HOST", "127.0.0.1") + my_app.run(host=flask_host, port=5000, debug=FLASK_DEBUG) diff --git a/aperisolve/config.py b/aperisolve/config.py index a813a2a..3ad8901 100644 --- a/aperisolve/config.py +++ b/aperisolve/config.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Aperi'Solve configuration variables.""" from os import getenv diff --git a/aperisolve/models.py b/aperisolve/models.py index dddaa8f..0dc0a80 100644 --- a/aperisolve/models.py +++ b/aperisolve/models.py @@ -1,14 +1,12 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=E1101,C0413,W0212,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable -"""This module defines the database models for the Aperi'Solve application.""" +"""Database models for the Aperi'Solve application.""" import itertools import shutil import struct import time import zlib -from datetime import datetime, timezone +from datetime import UTC, datetime +from typing import cast from flask_sqlalchemy import SQLAlchemy from sqlalchemy import ( @@ -21,6 +19,7 @@ from sqlalchemy import ( SmallInteger, String, ) +from sqlalchemy.exc import SQLAlchemyError from aperisolve.config import MAX_PENDING_TIME, MAX_STORE_TIME, RESULT_FOLDER from aperisolve.utils.utils import get_resolutions, get_valid_depth_color_pairs @@ -28,29 +27,31 @@ from aperisolve.utils.utils import get_resolutions, get_valid_depth_color_pairs db: SQLAlchemy = SQLAlchemy() -class Image(db.Model): # type: ignore +class Image(db.Model): """Model representing an image file in the database.""" hash = Column(String(64), primary_key=True, unique=True, nullable=False) file = Column(String(128), unique=True, nullable=False) size = Column(Integer, nullable=False) first_submission_date = Column( - DateTime, nullable=False, default=lambda: datetime.now(timezone.utc) + DateTime, + nullable=False, + default=lambda: datetime.now(UTC), ) last_submission_date = Column( DateTime, nullable=False, - default=lambda: datetime.now(timezone.utc), - onupdate=lambda: datetime.now(timezone.utc), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), ) upload_count = Column(Integer) submissions = db.relationship("Submission", backref="image", lazy=True) -class Submission(db.Model): # type: ignore +class Submission(db.Model): """Model representing a file submission for analysis. - Submissions are defined with a filename, password, image content, and analysis - option. + + Submissions are defined with a filename, password, image content, and analysis option. """ hash = Column(String(128), primary_key=True, unique=True, nullable=False) @@ -58,13 +59,12 @@ class Submission(db.Model): # type: ignore password = Column(String(128)) deep_analysis = Column(Boolean, default=False) status = Column(String(20), default="pending") - date = Column(Float, nullable=False, default=lambda: datetime.now(timezone.utc)) + date = Column(Float, nullable=False, default=lambda: datetime.now(UTC)) - # Foreign key to Image image_hash = Column(String, db.ForeignKey("image.hash"), nullable=False) -class IHDR(db.Model): # type: ignore +class IHDR(db.Model): """IHDR CRC lookup table with direct parameter storage.""" iid = Column(Integer, primary_key=True, autoincrement=True) @@ -77,23 +77,30 @@ class IHDR(db.Model): # type: ignore def to_ihdr_bytes(self) -> bytes: """Convert database record to IHDR chunk data (13 bytes).""" + bit_depth = cast("int", self.bit_depth) + color_type = cast("int", self.color_type) + interlace = cast("int", self.interlace) return ( struct.pack(">I", self.width) + struct.pack(">I", self.height) + bytes( [ - self.bit_depth, - self.color_type, - 0, # compression method (always 0) - 0, # filter method (always 0) - self.interlace, - ] + bit_depth, + color_type, + 0, + 0, + interlace, + ], ) ) @staticmethod def compute_crc( - width: int, height: int, bit_depth: int, color_type: int, interlace: int + width: int, + height: int, + bit_depth: int, + color_type: int, + interlace: int, ) -> int: """Compute CRC32 for IHDR chunk.""" ihdr_data = ( @@ -104,53 +111,34 @@ class IHDR(db.Model): # type: ignore return zlib.crc32(b"IHDR" + ihdr_data) & 0xFFFFFFFF -class UploadLog(db.Model): # type: ignore +class UploadLog(db.Model): """Model representing upload activity logs.""" id = Column(Integer, primary_key=True, autoincrement=True) - ip_address = Column(String(45), nullable=False) # IPv6 max length is 45 + ip_address = Column(String(45), nullable=False) user_agent = Column(String(512), nullable=True) - upload_time = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + upload_time = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC)) image_hash = Column(String(64), nullable=False) submission_hash = Column(String(128), nullable=True) filename = Column(String(128), nullable=True) def fill_ihdr_db() -> None: - """ - Populate IHDR table with common PNG configurations. - - Creates entries for all combinations of: - - Common resolutions (from get_resolutions()) - - Valid bit depth/color type pairs (from get_valid_depth_color_pairs()) - - Interlace methods (0, 1) - - Computes CRC for each combination and stores parameters directly. - """ + """Populate IHDR table with common PNG configurations.""" try: - # Skip if table already has entries (check before doing any work) count_query = db.session.query(IHDR.iid).limit(1) if db.session.execute(count_query).first() is not None: - print("IHDR table already populated, skipping fill.") return resolutions = get_resolutions() bit_color_pairs = list(get_valid_depth_color_pairs()) interlace_methods = [0, 1] + combinations = itertools.product(resolutions, bit_color_pairs, interlace_methods) - combinations = itertools.product( - resolutions, - bit_color_pairs, - interlace_methods, - ) count = 0 - with db.session.no_autoflush: for (width, height), (bit_depth, color_type), interlace in combinations: - # Compute CRC directly crc = IHDR.compute_crc(width, height, bit_depth, color_type, interlace) - - # Add new entry db.session.add( IHDR( crc=crc, @@ -159,78 +147,58 @@ def fill_ihdr_db() -> None: bit_depth=bit_depth, color_type=color_type, interlace=interlace, - ) + ), ) - count += 1 if count % 5000 == 0: - print(f"Inserted {count} IHDR entries...") db.session.commit() - print(f"Inserted {count} total IHDR entries.") - - except Exception as e: - print(f"Error filling IHDR table: {e}") + except SQLAlchemyError: db.session.rollback() -def cleanup_old_entries() -> None: - """ - Clean up old and incomplete entries from the database and file system. +def _cleanup_submissions(now: float) -> None: + """Delete stale or invalid submissions.""" + for submission in Submission.query.all(): + if submission.status in ("pending", "running") and now - submission.date > MAX_PENDING_TIME: + db.session.delete(submission) + continue - This function performs two main cleanup operations: - - 1. Submission cleanup: - - Deletes submissions with "pending" or "running" status that have exceeded - the maximum allowed processing time (MAX_PENDING_TIME). - - Deletes completed submissions ("done" status) that have missing or buggy - results (results.json file not found). - - 2. Image cleanup: - - Deletes images older than MAX_STORE_TIME along with all associated - submissions and their result folders. - - Deletes orphaned images (with no submissions) older than MAX_PENDING_TIME - and removes their associated result folders from the file system. - - All database changes are committed at the end of the operation. - - Returns: - None - """ - now = time.time() - for submission in Submission.query.all(): # type: ignore - if ( - submission.status in ("pending", "running") # type: ignore - and now - submission.date > MAX_PENDING_TIME # type: ignore - ): - # Processing took too long, delete - db.session.delete(submission) # pylint: disable=no-member - elif submission.status == "done": # type: ignore - # Search for buggy results, delete + if submission.status == "done": 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 + db.session.delete(submission) shutil.rmtree(result_path) - # Delete "old" images - for img in Image.query.all(): # type: ignore + +def _cleanup_images() -> None: + """Delete old and orphaned images with their result folders.""" + for img in Image.query.all(): if img.last_submission_date.tzinfo is None: - img_date = img.last_submission_date.replace(tzinfo=timezone.utc) + img_date = img.last_submission_date.replace(tzinfo=UTC) else: img_date = img.last_submission_date - delay = datetime.now(timezone.utc) - img_date - img_fold = RESULT_FOLDER / img.hash + + delay = datetime.now(UTC) - img_date + img_folder = RESULT_FOLDER / img.hash + if delay.total_seconds() > MAX_STORE_TIME: - for s in img.submissions: - db.session.delete(s) # pylint: disable=no-member - if img_fold.exists(): - shutil.rmtree(img_fold) # type: ignore - db.session.delete(img) # pylint: disable=no-member + for submission in img.submissions: + db.session.delete(submission) + if img_folder.exists(): + shutil.rmtree(img_folder) + db.session.delete(img) + continue - # Delete Images with missing submission if len(img.submissions) == 0 and delay.total_seconds() > MAX_PENDING_TIME: - if img_fold.exists(): - shutil.rmtree(img_fold) # type: ignore - db.session.delete(img) # pylint: disable=no-member + if img_folder.exists(): + shutil.rmtree(img_folder) + db.session.delete(img) - db.session.commit() # pylint: disable=no-member + +def cleanup_old_entries() -> None: + """Clean up old and incomplete entries from the database and file system.""" + now = time.time() + _cleanup_submissions(now) + _cleanup_images() + db.session.commit() diff --git a/aperisolve/requirements-dev.txt b/aperisolve/requirements-dev.txt deleted file mode 100644 index 182cc37..0000000 --- a/aperisolve/requirements-dev.txt +++ /dev/null @@ -1,7 +0,0 @@ -black==25.12.0 -isort==7.0.0 -flake8==7.3.0 -mypy==1.18.1 -pylint==4.0.4 -pre-commit==4.5.1 -mypy-extensions diff --git a/aperisolve/requirements.txt b/aperisolve/requirements.txt deleted file mode 100644 index f7ee61b..0000000 --- a/aperisolve/requirements.txt +++ /dev/null @@ -1,11 +0,0 @@ -Flask -Flask-SQLAlchemy -numpy -redis -rq -psycopg2-binary -types-Flask-SQLAlchemy -sqlalchemy-stubs -Pillow -gunicorn -sentry-sdk[flask,sqlalchemy,rq] \ No newline at end of file diff --git a/aperisolve/utils/__init__.py b/aperisolve/utils/__init__.py index e69de29..eeb1947 100644 --- a/aperisolve/utils/__init__.py +++ b/aperisolve/utils/__init__.py @@ -0,0 +1 @@ +"""Utility helpers used by AperiSolve modules.""" diff --git a/aperisolve/utils/init_db.py b/aperisolve/utils/init_db.py index abe89a3..b095fd1 100644 --- a/aperisolve/utils/init_db.py +++ b/aperisolve/utils/init_db.py @@ -1,5 +1,5 @@ -""" -Database initialization script. +"""Database initialization script. + This script is meant to be run ONCE at deploy time, never from Gunicorn or runtime code. If CLEAR_AT_RESTART is set, database will be reset at launch time. """ @@ -7,11 +7,12 @@ If CLEAR_AT_RESTART is set, database will be reset at launch time. import os from shutil import rmtree +import sentry_sdk from sqlalchemy import inspect -from ..app import create_app -from ..config import RESULT_FOLDER -from ..models import db, fill_ihdr_db +from aperisolve.app import create_app +from aperisolve.config import RESULT_FOLDER +from aperisolve.models import db, fill_ihdr_db def main() -> None: @@ -25,7 +26,6 @@ def main() -> None: tables = inspector.get_table_names() if reset: - # Delete result folder content if RESULT_FOLDER.exists(): for item in RESULT_FOLDER.iterdir(): @@ -34,28 +34,20 @@ def main() -> None: rmtree(item) else: item.unlink() - except Exception: - continue + except OSError as exc: + sentry_sdk.capture_exception(exc) - print("⚠️ CLEAR_AT_RESTART=1 detected") - print("πŸ’₯ Dropping all tables...") db.drop_all() - print("πŸ—„οΈ Recreating schema...") db.create_all() + elif not tables: + db.create_all() else: - if not tables: - print("πŸ—„οΈ Creating database schema...") - db.create_all() - else: - print("πŸ—„οΈ Database schema already exists") + pass - print("πŸ“¦ Filling IHDR static data...") fill_ihdr_db() - print("βœ… Database initialization complete") - - RESULT_FOLDER.mkdir(parents=True, exist_ok=True) + RESULT_FOLDER.mkdir(parents=True, exist_ok=True) if __name__ == "__main__": diff --git a/aperisolve/utils/png.py b/aperisolve/utils/png.py index 05fd1fb..8fc0fcd 100644 --- a/aperisolve/utils/png.py +++ b/aperisolve/utils/png.py @@ -1,16 +1,14 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801,R0902 -# mypy: disable-error-code=unused-awaitable -"""PNG Class for analyzer modules""" +"""PNG Class for analyzer modules.""" import itertools import re import struct import zlib -from typing import Any, Optional +from typing import Any + +from aperisolve.app import create_app +from aperisolve.models import IHDR -from ..app import create_app -from ..models import IHDR from .utils import int2hex, str2hex __author__ = [ @@ -20,11 +18,14 @@ __author__ = [ "sherlly (https://github.com/sherlly/PCRT)", ] +IDAT_NOT_FOUND_OFFSET = -5 + class PNG: """PNG file analyzer and repair tool.""" def __init__(self, data: bytes) -> None: + """Initialize PNG parser state for a binary payload.""" self.data = data self.width = self.height = self.bits = self.mode = 0 self.compression = self.filter = self.interlace = self.channel = 0 @@ -47,7 +48,7 @@ class PNG: """Check if data contains PNG signature chunks.""" return all(p in data for p in [b"IHDR", b"IDAT", b"IEND"]) - def _check_crc(self, chunk_type: bytes, chunk_data: bytes, checksum: bytes) -> Optional[bytes]: + def _check_crc(self, chunk_type: bytes, chunk_data: bytes, checksum: bytes) -> bytes | None: """Check CRC of chunk.""" calc_crc = struct.pack("!I", zlib.crc32(chunk_type + chunk_data)) return calc_crc if calc_crc != checksum else None @@ -101,7 +102,8 @@ class PNG: return True def _find_ancillary( - self, data: bytes + self, + data: bytes, ) -> tuple[dict[bytes, list[bytes]], dict[bytes, list[bytes]], dict[bytes, Any]]: """Find ancillary chunks in PNG data with an optimized single pass.""" ancillary = [ @@ -165,8 +167,10 @@ class PNG: return txt_content, image_content, crcs def check_chunks(self) -> bool: - """Copy ancillary chunks (like PLTE) from original data to repaired_data with crc - validation.""" + """Copy ancillary chunks (like PLTE) from original data to repaired_data with CRC. + + validation. + """ critical_ancillary = [ b"PLTE", b"tRNS", @@ -222,8 +226,51 @@ class PNG: return True + def _lookup_ihdr_from_db(self, chunk_type: bytes, crc: bytes) -> bytes | None: + """Try to recover IHDR bytes from known CRC values in the database.""" + crc_int = struct.unpack("!I", crc)[0] + app = create_app() + with app.app_context(): + matches = IHDR.query.filter_by(crc=crc_int).all() + + if not matches: + return None + + self._log(f"Found {len(matches)} matching IHDR configuration(s) in database") + match = matches[0] + test_ihdr = match.to_ihdr_bytes() + if self._check_crc(chunk_type, test_ihdr, crc): + self._log("Database match found but CRC verification failed") + return None + + self._log( + f"Recovered IHDR: {match.width}x{match.height}, " + f"bit_depth={match.bit_depth}, " + f"color_type={match.color_type}, " + f"interlace={match.interlace}", + ) + return test_ihdr + + def _bruteforce_ihdr_dimensions( + self, + chunk_type: bytes, + ihdr: bytes, + crc: bytes, + ) -> bytes | None: + """Fallback recovery for width and height by brute-force CRC matching.""" + self._log("No database match found, falling back to exhaustive search...") + for width in range(1, 5000): + for height in range(1, 5000): + test_ihdr = struct.pack(">I", width) + struct.pack(">I", height) + ihdr[16:21] + if not self._check_crc(chunk_type, test_ihdr, crc): + self._log( + f"Found correct dimensions via exhaustive search: {width}x{height}", + ) + return test_ihdr + return None + def check_ihdr(self) -> bool: - """Check and repair IHDR chunk using database lookup - WITH OPTIMIZED EXHAUSTIVE SEARCH.""" + """Check and repair IHDR chunk using database lookup and exhaustive fallback.""" pos, ihdr = self._find_ihdr(self.data) if pos == -1: self._error("Lost IHDR chunk") @@ -239,43 +286,11 @@ class PNG: self._log(f"Chunk crc: {str2hex(crc)}, Correct crc: {str2hex(calc_crc)}") self._log("Looking up CRC in database...") - crc_int = struct.unpack("!I", crc)[0] - - # Database lookup - app = create_app() - with app.app_context(): - matches = IHDR.query.filter_by(crc=crc_int).all() - - if matches: - self._log(f"Found {len(matches)} matching IHDR configuration(s) in database") - match = matches[0] - test_ihdr = match.to_ihdr_bytes() - - if not self._check_crc(chunk_type, test_ihdr, crc): - ihdr = ihdr[:8] + test_ihdr + crc - self._log( - f"Recovered IHDR: {match.width}x{match.height}, " - f"bit_depth={match.bit_depth}, " - f"color_type={match.color_type}, " - f"interlace={match.interlace}" - ) - fixed = True - else: - self._log("Database match found but CRC verification failed") - - # Fallback: exhaustive search if database lookup fails - if not fixed: - self._log("No database match found, falling back to exhaustive search...") - for w in range(1, 5000): - for h in range(1, 5000): - test_ihdr = struct.pack(">I", w) + struct.pack(">I", h) + ihdr[16:21] - if not self._check_crc(chunk_type, test_ihdr, crc): - ihdr = ihdr[:8] + test_ihdr + crc - self._log(f"Found correct dimensions via exhaustive search: {w}x{h}") - fixed = True - break - if fixed: - break + if (recovered_ihdr := self._lookup_ihdr_from_db(chunk_type, crc)) or ( + recovered_ihdr := self._bruteforce_ihdr_dimensions(chunk_type, ihdr, crc) + ): + ihdr = ihdr[:8] + recovered_ihdr + crc + fixed = True if not fixed: self._error("Could not recover IHDR dimensions") @@ -289,8 +304,12 @@ class PNG: return fixed def _fix_dos2unix( - self, chunk_type: bytes, chunk_data: bytes, crc: bytes, count: int - ) -> Optional[bytes]: + self, + chunk_type: bytes, + chunk_data: bytes, + crc: bytes, + count: int, + ) -> bytes | None: """Fix DOS to Unix line ending conversion.""" pos_list = [] pos = -1 @@ -307,7 +326,7 @@ class PNG: def check_idat(self) -> bool: """Check and repair IDAT chunks.""" idat_begin = self.data.find(b"IDAT") - 4 - if idat_begin == -5: + if idat_begin == IDAT_NOT_FOUND_OFFSET: self._error("Lost all IDAT chunks") return False @@ -322,43 +341,50 @@ class PNG: for i, pos1 in enumerate(pos_list): if i + 1 == len(pos_list): idat_table.append( - self.data[pos1 - 4 : pos_iend - 4 if pos_iend != -1 else len(self.data)] + self.data[pos1 - 4 : pos_iend - 4 if pos_iend != -1 else len(self.data)], ) else: idat_table.append(self.data[pos1 - 4 : pos_list[i + 1] - 4]) offset = idat_begin fixed = False - for idat_chunk in idat_table: - length = struct.unpack("!I", idat_chunk[:4])[0] - chunk_type, chunk_data, crc = idat_chunk[4:8], idat_chunk[8:-4], idat_chunk[-4:] + for chunk in idat_table: + current_chunk = chunk + length = struct.unpack("!I", current_chunk[:4])[0] + chunk_type, chunk_data, crc = ( + current_chunk[4:8], + current_chunk[8:-4], + current_chunk[-4:], + ) if length != len(chunk_data): self._log(f"Error IDAT chunk data length at offset {int2hex(offset)}") self._log(f"Length: {int2hex(length)}, Actual: {int2hex(len(chunk_data))}") if fixed_data := self._fix_dos2unix( - chunk_type, chunk_data, crc, abs(length - len(chunk_data)) + chunk_type, + chunk_data, + crc, + abs(length - len(chunk_data)), ): - idat_chunk = idat_chunk[:8] + fixed_data + idat_chunk[-4:] + current_chunk = current_chunk[:8] + fixed_data + current_chunk[-4:] self._log("Successfully recovered IDAT chunk data (DOS->Unix fix)") fixed = True else: self._log("Failed to fix IDAT chunk, using original") - else: - if calc_crc := self._check_crc(chunk_type, chunk_data, crc): - self._log(f"Error IDAT CRC at offset {int2hex(offset + 8 + length)}") - self._log(f"Chunk crc: {str2hex(crc)}, Correct: {str2hex(calc_crc)}") - idat_chunk = idat_chunk[:-4] + calc_crc - self._log("Successfully fixed CRC") - fixed = True + elif calc_crc := self._check_crc(chunk_type, chunk_data, crc): + self._log(f"Error IDAT CRC at offset {int2hex(offset + 8 + length)}") + self._log(f"Chunk crc: {str2hex(crc)}, Correct: {str2hex(calc_crc)}") + current_chunk = current_chunk[:-4] + calc_crc + self._log("Successfully fixed CRC") + fixed = True - self.repaired_data.extend(idat_chunk) + self.repaired_data.extend(current_chunk) offset += len(chunk_data) + 12 self._log(f"IDAT chunk check complete at offset {int2hex(idat_begin)}") return fixed - def check_iend(self) -> tuple[bool, Optional[bytes]]: + def check_iend(self) -> tuple[bool, bytes | None]: """Check and repair IEND chunk.""" standard_iend = b"\x00\x00\x00\x00IEND\xae\x42\x60\x82" pos = self.data.find(b"IEND") @@ -385,9 +411,8 @@ class PNG: self._log("IEND chunk check complete") return fixed, extra_data - def repair(self) -> tuple[bool, Optional[bytes]]: + def repair(self) -> tuple[bool, bytes | None]: """Run full PNG repair process.""" - if not self._check_format(self.data): self._error("File may not be a PNG image") return False, None diff --git a/aperisolve/utils/sentry.py b/aperisolve/utils/sentry.py index c0307a5..16af9ad 100644 --- a/aperisolve/utils/sentry.py +++ b/aperisolve/utils/sentry.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Sentry initialization for Aperisolve.""" import os @@ -10,6 +7,7 @@ from sentry_sdk.integrations.flask import FlaskIntegration from sentry_sdk.integrations.rq import RqIntegration from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration from sentry_sdk.integrations.threading import ThreadingIntegration +from sentry_sdk.utils import BadDsn SENTRY_DSN = os.environ.get("SENTRY_DSN") SENTRY_ENVIRONMENT = os.environ.get("SENTRY_ENVIRONMENT", "development") @@ -20,7 +18,6 @@ SENTRY_RELEASE = os.environ.get("SENTRY_RELEASE", "1.0.0") def initialize_sentry() -> None: """Initialize Sentry SDK once, safely.""" - # No DSN β†’ do nothing # Prevent double initialization (Flask + RQ, CLI, tests, etc.) if not SENTRY_DSN or sentry_sdk.Hub.current.client is not None: @@ -42,7 +39,7 @@ def initialize_sentry() -> None: send_default_pii=False, enable_tracing=True, ) - except sentry_sdk.utils.BadDsn as e: - print(f"Sentry DSN is invalid, skipping Sentry initialization: {e}") - except Exception as e: - print(f"Sentry initialization failed: {e}") + except BadDsn: + sentry_sdk.capture_message("Invalid Sentry DSN configuration", level="warning") + except RuntimeError as exc: + sentry_sdk.capture_exception(exc) diff --git a/aperisolve/utils/utils.py b/aperisolve/utils/utils.py index 0cf7845..9727cd7 100644 --- a/aperisolve/utils/utils.py +++ b/aperisolve/utils/utils.py @@ -1,11 +1,10 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Utility functions for analyzers modules.""" import binascii import math -from typing import Iterable +from collections.abc import Iterable + +MAX_RESOLUTION_SIZE = 10_000 def str2hex(s: bytes) -> str: @@ -15,12 +14,11 @@ def str2hex(s: bytes) -> str: def int2hex(i: int) -> str: """Convert int to hex string with 0x prefix.""" - return "0x" + hex(i)[2:].upper() + return f"0x{i:X}" def get_resolutions() -> list[tuple[int, int]]: - """ - Generate a sorted list of common display and image resolutions. + """Generate a sorted list of common display and image resolutions. Creates a comprehensive set of resolution tuples by combining various base widths with standard aspect ratios used in screens, digital displays, photography, and @@ -42,14 +40,15 @@ def get_resolutions() -> list[tuple[int, int]]: Returns: list[tuple[int, int]]: Sorted list of (width, height) resolution tuples with heights constrained between 1 and 10000 pixels. + """ base_widths = ( list(range(16, 257, 16)) # 16 β†’ 256 + list(range(320, 1025, 32)) # 320 β†’ 1024 + list(range(1280, 2561, 64)) # 1280 β†’ 2560 - + list(range(3000, 4097, 128)) # 3K–4K - + list(range(5120, 8193, 256)) # 5K–8K - + [10000] # upper bound + + list(range(3000, 4097, 128)) # 3K-4K + + list(range(5120, 8193, 256)) # 5K-8K + + [MAX_RESOLUTION_SIZE] # upper bound ) aspect_ratios = [ # Screens / digital @@ -74,15 +73,14 @@ def get_resolutions() -> list[tuple[int, int]]: for w in base_widths: for ar_w, ar_h in aspect_ratios: h = math.floor(round(w * ar_h / ar_w)) - if 1 <= h <= 10000: + if 1 <= h <= MAX_RESOLUTION_SIZE: resolutions.add((w, h)) resolutions.add((h, w)) # portrait / landscape return sorted(resolutions) def get_valid_depth_color_pairs() -> Iterable[tuple[int, int]]: - """ - Generate valid combinations of color and bit depth values. + """Generate valid combinations of color and bit depth values. Yields tuples of (depth, color) representing valid color mode and bit depth combinations. Supports grayscale (0), RGB (2, 3, 4, 6) color modes with @@ -91,6 +89,7 @@ def get_valid_depth_color_pairs() -> Iterable[tuple[int, int]]: Yields: tuple: A tuple of (depth, color) where depth is an integer representing bit depth and color is an integer representing the color mode. + """ valid_color_depth = { 0: [1, 2, 4, 8, 16], diff --git a/aperisolve/utils/wsgi.py b/aperisolve/utils/wsgi.py index 3a90300..eec6b7f 100644 --- a/aperisolve/utils/wsgi.py +++ b/aperisolve/utils/wsgi.py @@ -1,6 +1,3 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Aperi'Solve WSGI runner.""" from aperisolve.app import create_app diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 98c6b5a..2454e0f 100644 --- a/aperisolve/workers.py +++ b/aperisolve/workers.py @@ -1,19 +1,13 @@ -# flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 -# mypy: disable-error-code=unused-awaitable """Asynchronous worker for analyzing image submissions.""" import threading +from collections.abc import Callable from pathlib import Path -from typing import Any import sentry_sdk +from sqlalchemy.exc import SQLAlchemyError -from .utils.sentry import initialize_sentry - -initialize_sentry() - -from .analyzers.binwalk import analyze_binwalk # pylint: disable=C0413 +from .analyzers.binwalk import analyze_binwalk from .analyzers.color_remapping import analyze_color_remapping from .analyzers.decomposer import analyze_decomposer from .analyzers.exiftool import analyze_exiftool @@ -32,62 +26,40 @@ from .analyzers.zsteg import analyze_zsteg from .app import create_app from .config import RESULT_FOLDER from .models import Image, Submission, db +from .utils.sentry import initialize_sentry def analyze_image(submission_hash: str) -> None: - """ - Analyze an image submission by running multiple analysis tools concurrently. - - This function retrieves a submission and its associated image from the database, - then executes various image analysis tools in parallel threads. The analysis tools - include binwalk, decomposer, exiftool, foremost, strings, pngcheck, steghide, - zsteg, and image resize. If deep analysis is enabled, additional tools like - outguess are included. - - Args: - submission_hash (str): The unique hash identifier of the submission to analyze. - - Returns: - None - - Raises: - Implicitly catches and handles exceptions during analysis, setting submission - status to "error" if an exception occurs. - - Side Effects: - - Queries the database for the submission and associated image - - Updates submission status in the database ("running" -> "completed" or "error") - - Creates result directories and generates analysis output files - - Modifies the database session and commits changes - """ + """Analyze an image submission by running multiple analysis tools concurrently.""" + initialize_sentry() app = create_app() with app.app_context(): - submission: Submission = Submission.query.get(submission_hash) # type: ignore - image = Image.query.get_or_404(submission.image_hash) # type: ignore - - if not submission or not image: # No submission found - sentry_sdk.capture_message( - f"Submission or image not found: {submission_hash}", level="warning" - ) + submission = Submission.query.get(submission_hash) + if submission is None: + sentry_sdk.capture_message(f"Submission not found: {submission_hash}", level="warning") return - submission.status = "running" # type: ignore - db.session.commit() # pylint: disable=no-member + image = Image.query.get_or_404(submission.image_hash) + submission.status = "running" + db.session.commit() try: - img_path: Path = Path(image.file) - result_path: Path = RESULT_FOLDER / str(image.hash) / str(submission.hash) + img_path = Path(image.file) + result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) result_path.mkdir(parents=True, exist_ok=True) threads: list[threading.Thread] = [] - def run_analyzer(analyzer_func: Any, *args: Any) -> None: + def run_analyzer(analyzer_func: Callable[..., None], *args: object) -> None: """Run an analyzer function in a separate thread.""" - analyzer_name = analyzer_func.__name__.replace("analyze_", "") + analyzer_name = getattr( + analyzer_func, + "__name__", + analyzer_func.__class__.__name__, + ).replace("analyze_", "") try: analyzer_func(*args) - except Exception as e: - print(f"Error in {analyzer_name}: {e}") + except (RuntimeError, ValueError, OSError, TypeError) as exc: with sentry_sdk.push_scope() as scope: scope.set_tag("analyzer", analyzer_name) scope.set_tag("submission_hash", submission_hash) @@ -102,7 +74,7 @@ def analyze_image(submission_hash: str) -> None: }, ) scope.fingerprint = ["analyzer-error", analyzer_name] - sentry_sdk.capture_exception(e) + sentry_sdk.capture_exception(exc) analyzers = [ (analyze_binwalk, img_path, result_path), @@ -122,13 +94,8 @@ def analyze_image(submission_hash: str) -> None: (analyze_zsteg, img_path, result_path), ] - # Deep analysis only if submission.deep_analysis: - analyzers.extend( - [ - (analyze_outguess, img_path, result_path), - ] - ) + analyzers.append((analyze_outguess, img_path, result_path)) for analyzer in analyzers: thread = threading.Thread(target=run_analyzer, args=analyzer) @@ -138,11 +105,10 @@ def analyze_image(submission_hash: str) -> None: for thread in threads: thread.join() - submission.status = "completed" # type: ignore - except Exception as e: - sentry_sdk.capture_exception(e) - submission.status = "error" # type: ignore + submission.status = "completed" + except (RuntimeError, ValueError, OSError, TypeError, SQLAlchemyError) as exc: + sentry_sdk.capture_exception(exc) + submission.status = "error" finally: - db.session.commit() # pylint: disable=no-member - # Flush Sentry events - wait up to 5 seconds for events to be sent + db.session.commit() sentry_sdk.flush(timeout=5) diff --git a/lint.sh b/lint.sh index bda4320..931919f 100755 --- a/lint.sh +++ b/lint.sh @@ -1,21 +1,15 @@ #!/usr/bin/env bash -# lint.sh - Run all formatters, linters, and type checks according to pyproject.toml +# lint.sh - Run linting and type checks according to pyproject.toml set -euo pipefail -echo "πŸ–€ Running Black..." -black . +echo "🧹 Running Ruff lint checks..." +ruff check . -echo "πŸ“¦ Running Isort..." -isort . +echo "🎨 Running Ruff format checks..." +ruff format . --check -echo "🐍 Running Flake8..." -flake8 --max-line-length 100 --extend-ignore=E203,E501,W503 . - -echo "πŸ” Running Mypy..." -mypy . - -echo "⚑ Running Pylint..." -pylint aperisolve/ +echo "πŸ” Running ty type checks..." +ty check aperisolve echo "βœ… All checks passed!" diff --git a/pyproject.toml b/pyproject.toml index 1be4be7..88baeff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,9 +1,9 @@ # Minimal Python project configuration for linters and type checkers. -# This file centralizes config for black, isort, flake8, mypy, pylint. +# This file centralizes config for ruff and ty. [project] name = "aperisolve" -version = "3.1.3" +version = "3.2.0" description = "Open-source steganalysis web platform for automated image analysis and steganography detection" readme = "README.md" requires-python = ">=3.12" @@ -11,37 +11,34 @@ license = {text = "MIT"} authors = [ {name = "Zeecka"} ] +dependencies = [ + "Flask", + "Flask-SQLAlchemy", + "numpy", + "redis", + "rq", + "psycopg2-binary", + "Pillow", + "gunicorn", + "sentry-sdk[flask,sqlalchemy,rq]" +] -[tool.black] +[project.optional-dependencies] +dev = [ + "ruff", + "ty", + "pre-commit", + "types-Flask-SQLAlchemy", + "sqlalchemy-stubs" +] + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[tool.ruff] line-length = 100 -target-version = ["py312"] +target-version = "py312" -[tool.isort] -profile = "black" - -[tool.flake8] -extend-ignore = ["E203", "E501", "W503"] -max-line-length = 100 - -[tool.mypy] -disallow_any_generics = true -disallow_untyped_defs = true -follow_imports = "skip" -ignore_missing_imports = true -disable_error_code = ["unused-awaitable"] - -[tool.pylint] -disable = ["W0511", "W0718", "R0801", "R0903", "R0914"] - -[tool.pre-commit] -skip = true - -[tool.doc] -description = """ -Linting and type checking configuration for this project. -- Black: code formatting -- Isort: import sorting -- Flake8: code style -- Mypy: static typing -- Pylint: code quality -""" +[tool.ruff.lint] +select = ["ALL"]