venus/aperisolve
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature/ruff (#186)

Browse Source 
* Ruff and Pyright OK

* lint

* uv

* add uv to readme

* Replace pyright with TY

* Lint is ok
This commit is contained in:
Zeecka
2026-02-21 00:00:48 +01:00
committed by GitHub
parent 6660cd2e5d
commit 730ed50f70
39 changed files with 767 additions and 868 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/lint.yml
+12 -15
View File
@@ -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
.github/workflows/release.yml
+2
View File
@@ -8,6 +8,8 @@ on:
jobs:
docker-releases:
runs-on: ubuntu-latest
env:
RELEASE_VERSION: ${{ github.ref_name }}
steps:
- name: Get release version
CONTRIBUTING.md
+33 -61
View File
@@ -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 && \
Dockerfile
+7 -3
View File
@@ -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
README.md
+3 -1
View File
@@ -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)
aperisolve/__init__.py
+1 -4
View File
@@ -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"]
aperisolve/analyzers/__init__.py
+1 -3
View File
@@ -1,3 +1 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Analyzer implementations for AperiSolve."""
aperisolve/analyzers/base_analyzer.py
+60 -34
View File
@@ -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,
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"),
)
def generate_archive(self, output_dir: Path, extracted_dir: Optional[Path] = None) -> str:
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,8 +89,7 @@ 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:
with _thread_lock, lock_file.open("w", encoding="utf-8") as lock:
fcntl.flock(lock, fcntl.LOCK_EX) # synchronizes across processes
try:
@@ -77,7 +97,7 @@ class SubprocessAnalyzer(ABC):
data: dict[Any, Any] = {}
if json_file.exists():
try:
with open(json_file, "r", encoding="utf-8") as f:
with json_file.open(encoding="utf-8") as f:
data = json.load(f)
except json.JSONDecodeError:
data = {}
@@ -86,10 +106,10 @@ class SubprocessAnalyzer(ABC):
data.update(new_data)
# Write safely to a temp file
with open(tmp_file, "w", encoding="utf-8") as f:
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
tmp_file.replace(json_file) # ensures file write is atomic
finally:
fcntl.flock(lock, fcntl.LOCK_UN)
@@ -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
aperisolve/analyzers/binwalk.py
+5 -4
View File
@@ -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
aperisolve/analyzers/color_remapping.py
+26 -17
View File
@@ -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
aperisolve/analyzers/decomposer.py
+10 -14
View File
@@ -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
aperisolve/analyzers/exiftool.py
+2 -3
View File
@@ -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:
aperisolve/analyzers/file.py
+3 -3
View File
@@ -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
aperisolve/analyzers/foremost.py
+7 -5
View File
@@ -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 []
aperisolve/analyzers/identify.py
+2 -3
View File
@@ -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()]
aperisolve/analyzers/jpseek.py
+9 -14
View File
@@ -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:
aperisolve/analyzers/jsteg.py
+2 -3
View File
@@ -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 []
aperisolve/analyzers/openstego.py
+11 -12
View File
@@ -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
aperisolve/analyzers/outguess.py
+4 -7
View File
@@ -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:
aperisolve/analyzers/pcrt.py
+11 -11
View File
@@ -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."""
aperisolve/analyzers/pngcheck.py
+5 -4
View File
@@ -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
aperisolve/analyzers/steghide.py
+13 -17
View File
@@ -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:
aperisolve/analyzers/strings.py
+1 -3
View File
@@ -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]
aperisolve/analyzers/template_analyzer.py
+4 -6
View File
@@ -1,10 +1,6 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""<toolname> 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 <toolname>."""
def __init__(self, input_img: Path, output_dir: Path) -> None:
"""Initialize the template analyzer."""
super().__init__("<toolname>", 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 ["<toolname>", "-p", password, "-i", self.img]
return ["<toolname>", "-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 <toolname>."""
analyzer = TemplateAnalyzer(input_img, output_dir)
if password:
aperisolve/analyzers/zsteg.py
+4 -4
View File
@@ -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:
aperisolve/app.py
+183 -186
View File
@@ -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,15 +138,18 @@ def create_app() -> Flask:
@app.route("/<hash_val>", 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("/upload", methods=["POST"])
def upload_image() -> tuple[Response, int]:
"""Handle image upload and initiate analysis."""
@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:
@@ -118,9 +165,8 @@ def create_app() -> Flask:
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()
img_hash = hashlib.md5(image_data, usedforsecurity=False).hexdigest()
image_name = img_hash + "." + image.filename.rsplit(".", maxsplit=1)[-1].lower()
submission_data = (
@@ -129,10 +175,9 @@ def create_app() -> Flask:
+ (password.encode() if password else b"")
+ (b"deep_analysis" if deep_analysis else b"")
)
submission_hash = hashlib.md5(submission_data).hexdigest()
submission_hash = hashlib.md5(submission_data, usedforsecurity=False).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()
@@ -145,50 +190,37 @@ def create_app() -> Flask:
submission_hash=submission_hash,
filename=image.filename,
)
db.session.add(upload_log) # pylint: disable=no-member
db.session.commit() # pylint: disable=no-member
db.session.add(upload_log)
db.session.commit()
# 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
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
# 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)
with new_img_path.open("wb") as file_obj:
file_obj.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),
first_submission_date=datetime.now(UTC),
last_submission_date=datetime.now(UTC),
)
db.session.add(sub_img) # pylint: disable=no-member
db.session.commit() # pylint: disable=no-member
db.session.add(sub_img)
db.session.commit()
# Increment upload count for the image
sub_img.upload_count += 1
db.session.commit() # pylint: disable=no-member
sub_img_any = cast("Any", sub_img)
sub_img_any.upload_count = int(sub_img_any.upload_count or 0) + 1
db.session.commit()
# 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,
@@ -199,34 +231,45 @@ def create_app() -> Flask:
date=time.time(),
image_hash=sub_img.hash,
)
db.session.add(submission) # pylint: disable=no-member
db.session.commit() # pylint: disable=no-member
db.session.add(submission)
db.session.commit()
# Re-analysis case to prevent early return
submission.status = "pending" # type: ignore
db.session.commit() # pylint: disable=no-member
submission_any = cast("Any", submission)
submission_any.status = "pending"
db.session.commit()
# Start the analysis jobs
app.config["REDIS_QUEUE"].enqueue(
"aperisolve.workers.analyze_image", submission.hash, job_timeout=300
"aperisolve.workers.analyze_image",
submission.hash,
job_timeout=300,
)
# Return submission identifier
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]:
return _upload_image(app)
@app.route("/status/<hash_val>", 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/<hash_val>", 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/<hash_val>", 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/<hash_val>/<tool>")
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/<img_name>")
@app.route("/image/<hash_val>/<img_name>")
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/<hash_val>", 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/<hash_val>", 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/<img_name>")
@app.route("/image/<hash_val>/<img_name>")
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)
aperisolve/config.py
-3
View File
@@ -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
aperisolve/models.py
+69 -101
View File
@@ -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()
aperisolve/requirements-dev.txt
-7
View File
@@ -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
aperisolve/requirements.txt
-11
View File
@@ -1,11 +0,0 @@
Flask
Flask-SQLAlchemy
numpy
redis
rq
psycopg2-binary
types-Flask-SQLAlchemy
sqlalchemy-stubs
Pillow
gunicorn
sentry-sdk[flask,sqlalchemy,rq]
aperisolve/utils/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Utility helpers used by AperiSolve modules."""
aperisolve/utils/init_db.py
+10 -18
View File
@@ -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,27 +34,19 @@ 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()
else:
if not tables:
print("🗄️ Creating database schema...")
elif not tables:
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)
aperisolve/utils/png.py
+89 -64
View File
@@ -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}"
)
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
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 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):
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)}")
idat_chunk = idat_chunk[:-4] + 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
aperisolve/utils/sentry.py
+5 -8
View File
@@ -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)
aperisolve/utils/utils.py
+12 -13
View File
@@ -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],
aperisolve/utils/wsgi.py
-3
View File
@@ -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
aperisolve/workers.py
+28 -62
View File
@@ -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)
lint.sh
+7 -13
View File
@@ -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!"
pyproject.toml
+30 -33
View File
@@ -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"]