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

PCRT, DB fix, Refactoring, Documentation

Browse Source 
Adds a PNG repair tool (PCRT) that can fix corrupted PNG files and extracts hidden data. Also simplifies the database structure and fixes initialization issues.
This commit is contained in:
Zeecka
2026-01-08 20:19:11 +01:00
committed by GitHub
parent 9b5f3caf99 1e4f12174c
commit e6cc8aa62e
28 changed files with 1335 additions and 1013 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1 -2
View File
@@ -1,5 +1,5 @@
# Application configuration
MAX_CONTENT_LENGTH=1048576
MAX_CONTENT_LENGTH=3145728
MAX_PENDING_TIME=600
MAX_STORE_TIME=259200
CLEAR_AT_RESTART=1
@@ -7,7 +7,6 @@ SKIP_IHDR_FILL=0
# Flask configuration
FLASK_DEBUG=0
FLASK_ENV=development
# Redis configuration
RQ_DASHBOARD_REDIS_URL=redis://redis:6379/0
.github/workflows/lint.yml
+2 -7
View File
@@ -2,11 +2,6 @@ name: Linters
on: [push, pull_request]
env:
FLAKE8_IGNORE: E203,E501,W503
PYLINT_IGNORE: W0718,R0903,R0801
MYPY_IGNORE: unused-awaitable
permissions:
checks: write
contents: write
@@ -23,11 +18,11 @@ jobs:
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: 3.11
python-version: 3.12
- name: Install Python dependencies
run: |
pip install isort black flake8 pylint mypy mypy-extensions
pip install -r aperisolve/requirements-dev.txt
pip install -r aperisolve/requirements.txt
- name: Run linters
.gitignore
+1
View File
@@ -1,5 +1,6 @@
# Python
.env
.venv/
__pycache__/
.mypy_cache/
results/
.pre-commit-config.yaml
+20 -14
View File
@@ -1,17 +1,11 @@
repos:
# - repo: https://github.com/pre-commit/pre-commit-hooks
# rev: v6.0.0
# hooks:
# - id: check-yaml
# - id: end-of-file-fixer
# - id: trailing-whitespace
- repo: https://github.com/psf/black
rev: 25.11.0
rev: 25.12.0
hooks:
- id: black
args: ['--line-length', '100']
- repo: https://github.com/pycqa/isort
rev: 5.11.2
rev: 7.0.0
hooks:
- id: isort
args: ['--profile', 'black']
@@ -21,14 +15,26 @@ repos:
- id: flake8
args: ['--extend-ignore','E203,E501,W503']
- repo: https://github.com/pylint-dev/pylint
rev: v4.0.2
rev: v4.0.4
hooks:
- id: pylint
args: ['--disable', 'W0511,W0718,R0801,R0903,R0914']
language: system
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.18.2
- repo: local
hooks:
- id: mypy
args: ["--disallow-any-generics", "--disallow-untyped-def", "--follow-imports", "skip", "--ignore-missing-imports", "--disable-error-code", "unused-awaitable"]
language: system
- id: mypy
name: mypy
entry: mypy
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",
]
CONTRIBUTING.md
+39 -23
View File
@@ -60,41 +60,57 @@ Contributions to the web interface are welcome:
## Code Style & Quality
> [!IMPORTANT]
> Please follow the code style and run linters before submitting.
> Follow the project’s code style and run linters before submitting any code.
The project adheres to:
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
- **Black** (line length 100)
- **Isort** (profile black)
- **Flake8** (ignoring E203, E501, W503)
- **Pylint** (ignoring W0718, R0903, R0801)
- **Mypy** (ignoring unused-awaitable)
See [.pre-commit-config.yaml](.pre-commit-config.yaml) for the exact configuration.
See [.pre-commit-config.yaml](.pre-commit-config.yaml) for exact configuration.
### Running Linters Locally
### Setup
> [!TIP]
> Install pre-commit hooks to automatically check your code before commits:
> ```bash
> pip install pre-commit
> pre-commit install
> ```
> Use a virtual environment and install development dependencies:
```bash
# Create & activate a virtual environment
python3 -m venv .venv
source .venv/bin/activate # Linux/macOS
# .venv\Scripts\activate # Windows
# Install dev dependencies
pip install -r aperisolve/requirements-dev.txt
pip install -r aperisolve/requirements.txt
```
### Running Tools Manually
Install pre-commit hooks to automatically check your code before commits:
```bash
pre-commit install
```
Check for issues:
```bash
pre-commit run --all-files
```
Format code:
**Manual checks:**
```bash
# Format code
black . --line-length 100
isort . --profile black
# Check for issues
flake8 .
pylint **/*.py
mypy .
```
> [!NOTE]
> CI will automatically run these checks on every PR. Make sure they pass!
> Continuous Integration (CI) runs all checks on every PR. Ensure your code passes before submitting.
## Pull Request Guidelines
Dockerfile
+1 -1
View File
@@ -49,4 +49,4 @@ COPY aperisolve/ /app/aperisolve/
RUN pip install --no-cache-dir -r /app/aperisolve/requirements.txt
# Commande de lancement
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "--access-logfile", "-", "--error-logfile", "-", "--log-level", "info", "--capture-output", "aperisolve.wsgi:application"]
CMD ["gunicorn", "-w", "8", "-b", "0.0.0.0:5000", "--access-logfile", "-", "--error-logfile", "-", "--log-level", "info", "--capture-output", "aperisolve.utils.wsgi:application"]
aperisolve/__init__.py
+5
View File
@@ -1,3 +1,8 @@
"""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/base_analyzer.py
+5 -2
View File
@@ -13,8 +13,9 @@ from pathlib import Path
from shutil import rmtree
from typing import Any, Optional, overload
from ..config import MAX_PENDING_TIME
_thread_lock = threading.Lock()
MAX_PENDING_TIME = int(os.getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default
class SubprocessAnalyzer(ABC):
@@ -50,8 +51,10 @@ class SubprocessAnalyzer(ABC):
timeout=MAX_PENDING_TIME,
)
def generate_archive(self, output_dir: Path, extracted_dir: Path) -> str:
def generate_archive(self, output_dir: Path, extracted_dir: Optional[Path] = None) -> str:
"""Zip the extracted files and remove the directory."""
if extracted_dir is None:
extracted_dir = self.get_extracted_dir()
zip_data = self.run_command(["7z", "a", f"../{self.name}.7z", "*"], cwd=extracted_dir)
rmtree(extracted_dir)
return zip_data.stderr
aperisolve/analyzers/image_resize.py
-164
View File
@@ -1,164 +0,0 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Image size bruteforcer for common PNG sizes."""
from pathlib import Path
from typing import Any, List, Optional
from ..app import app
from ..models import IHDR
from .base_analyzer import SubprocessAnalyzer
from .utils import PNG
class ResizeAnalyzer(SubprocessAnalyzer):
"""Analyzer for image resize."""
png: PNG
def __init__(self, input_img: Path, output_dir: Path) -> None:
super().__init__("image_resize", input_img, output_dir, has_archive=True)
def _lookup_crc(self) -> List[PNG]:
"""
Look up PNG images in the database by their CRC value.
Returns:
List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their
packed representations.
"""
with app.app_context():
imgs = IHDR.query.filter_by(crc=self.png.crc).all()
results = []
for img in imgs:
packed = img.packed
results.append(PNG.from_packed(packed, crc=self.png.crc))
return results
def _search_height_crc(self) -> List[PNG]:
"""
Search for PNG images with matching CRC by iterating through possible heights.
This function attempts to find alternative PNG configurations that produce the same
CRC checksum as the input PNG by systematically varying the height parameter while
keeping all other properties constant.
Returns:
List[PNG]: A list of PNG objects with matching CRC values. Returns an empty list
if no matches are found.
Note:
The function iterates through heights from 100 to 3500 (inclusive).
This may be computationally expensive for large ranges.
"""
results = []
for height in range(100, 3501):
candidate = PNG(
size=(self.png.width, height),
bit_depth=self.png.bit_depth,
color_type=self.png.color_type,
interlace=self.png.interlace,
)
if candidate.crc == self.png.crc:
results.append(candidate)
return results
def _write_recovered_png(self, png_bytes: bytearray, png_match: PNG) -> str:
"""
Rebuilds a PNG by patching IHDR width/height and writes it to disk.
"""
img_name = f"recovered_{png_match.width}x{png_match.height}.png"
output_path = self.output_dir / img_name
height_bytes = png_match.height.to_bytes(4, byteorder="big")
width_bytes = png_match.width.to_bytes(4, byteorder="big")
full_png_data = (
png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
+ width_bytes
+ height_bytes
+ png_bytes[24:]
)
with output_path.open("wb") as out_f:
out_f.write(full_png_data)
return img_name
def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
"""
Analyze a PNG image and recover resized versions based on CRC matching.
This function attempts to find matching image dimensions for a given PNG file
by looking up its CRC (Cyclic Redundancy Check) in a database. If matches are
found, it recovers and saves PNG images with the matched dimensions.
Returns:
dict[str, Any]: A dictionary containing the analysis status, any error message,
Raises:
Handles exceptions gracefully by updating output data with error status.
Side Effects:
- Creates output directory if it doesn't exist.
- Saves recovered PNG images to the output directory.
- Updates data in the output directory with analysis results via update_data().
Notes:
- If the PNG has invalid structure or IHDR chunk is not first, an error is logged.
- The function performs a two-step lookup: first by CRC, then by height/CRC if no
matches found.
- Recovered images are named as "recovered_{width}x{height}.png".
- Analysis results (logs, status, and image URLs) are persisted via update_data().
"""
logs = []
with self.input_img.open("rb") as image:
img_bytes = bytearray(image.read())
try:
self.png = PNG.from_bytearray(img_bytes)
except ValueError:
return {
"status": "error",
"error": "IHDR chunk is not the first chunk, or PNG has invalid structure.",
}
# Already correct
if self.png.crc == self.png.compute_crc():
return {
"status": "ok",
"note": f"PNG is already valid with dimensions "
f"{self.png.width}x{self.png.height} and crc 0x{self.png.crc:08x}.",
}
saved_img_urls = []
# No match found
if not (db_matches := self._lookup_crc()) and not (db_matches := self._search_height_crc()):
return {
"status": "error",
"error": "Failure: No matching dimensions found.",
}
# Match found, saving image(s)
logs.append(f"Target CRC found: 0x{self.png.crc:08x}")
self.output_dir.mkdir(parents=True, exist_ok=True)
for png_match in db_matches:
img_name = self._write_recovered_png(img_bytes, png_match)
logs.append(f"Image saved: {img_name}")
# Append the formatted URL to our list
saved_img_urls.append("/image/" + str(Path(self.output_dir.name) / img_name))
return {
"status": "ok",
"output": logs,
"png_images": saved_img_urls,
}
def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
"""Analyze an image submission using ResizeAnalyzer."""
analyzer = ResizeAnalyzer(input_img, output_dir)
analyzer.analyze()
aperisolve/analyzers/pcrt.py
+79
View File
@@ -0,0 +1,79 @@
# 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 ..utils.png import PNG
from .base_analyzer import SubprocessAnalyzer
class PCRTAnalyzer(SubprocessAnalyzer):
"""Analyzer for PCRT (PNG Check & Repair Tool)."""
def __init__(self, input_img: Path, output_dir: Path) -> None:
super().__init__("pcrt", input_img, output_dir, has_archive=True)
def _write_repaired_data(self, data: bytes) -> str:
"""Write recovered image."""
img_name = f"pcrt_recoverd_{self.input_img.name}.png"
output_path = self.output_dir / img_name
saved_img_url = "/image/" + str(Path(self.output_dir.name) / img_name)
with output_path.open("wb") as f:
f.write(data)
return saved_img_url
def _write_extra_data(self, data: bytes) -> None:
"""Write recovered extra data."""
extracted_dir = self.get_extracted_dir()
extracted_dir.mkdir(parents=True, exist_ok=True)
extra_path = extracted_dir / "extra_data.bin"
with extra_path.open("wb") as f:
f.write(data)
def get_results(self, password: Optional[str] = None) -> dict[str, Any]:
"""Analyze PNG and attempt repairs."""
try:
with self.input_img.open("rb") as f:
data = f.read()
png = PNG(data)
fixed, extra_data = png.repair()
result: dict[str, Any] = {}
if png.errors:
result["status"] = "error"
result["error"] = "\n".join(png.errors)
if png.logs:
result["output"] = png.logs
return result
result["status"] = "ok"
result["output"] = png.logs if png.logs else ["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)
result["note"] = "PNG was repaired and saved"
result["png_images"] = [url]
# Save extra data if found after IEND
if extra_data:
self._write_extra_data(extra_data)
self.generate_archive(self.output_dir)
result["note"] = result.get("note", "") + " | Extra data found after IEND"
result["download"] = f"/download/{self.output_dir.name}/{self.name}"
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."""
analyzer = PCRTAnalyzer(input_img, output_dir)
analyzer.analyze()
aperisolve/analyzers/utils.py
-264
View File
@@ -1,264 +0,0 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Utility functions for analyzers modules."""
import math
import struct
import zlib
from functools import cached_property
from typing import Iterable, Self
def get_resolutions() -> list[tuple[int, int]]:
"""
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
paper standards. Includes both portrait and landscape orientations.
Base widths range from 16px to 10000px, covering:
- Low resolution: 16-256px (16px increments)
- Standard: 320-1024px (32px increments)
- HD/Full HD: 1280-2560px (64px increments)
- 4K: 3000-4096px (128px increments)
- 5K-8K: 5120-8192px (256px increments)
- Upper bound: 10000px
Aspect ratios include common formats:
- Digital screens: 1:1, 4:3, 3:2, 16:10, 16:9, 21:9
- Photography/print: 5:4, 7:5, 2:3
- Paper standards: ISO A-series, US Letter, Legal, Tabloid
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
)
aspect_ratios = [
# Screens / digital
(1, 1), # square
(4, 3),
(3, 2),
(16, 10),
(16, 9),
(21, 9),
# Photography / print
(5, 4), # 8x10
(7, 5), # 5x7
(3, 2), # 4x6 (already included, kept for clarity)
(2, 3), # poster
# Paper standards
(1414, 1000), # ISO A-series (A4, A3, A2, ...)
(11, 85), # US Letter
(14, 85), # Legal
(17, 11), # Tabloid
]
resolutions: set[tuple[int, int]] = set()
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:
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.
Yields tuples of (depth, color) representing valid color mode and bit depth
combinations. Supports grayscale (0), RGB (2, 3, 4, 6) color modes with
appropriate bit depths (1, 2, 4, 8, 16 bits).
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],
2: [8, 16],
3: [1, 2, 4, 8],
4: [8, 16],
6: [8, 16],
}
for color, depths in valid_color_depth.items():
for depth in depths:
yield depth, color
class PNG:
"""
A class representing a PNG image header (IHDR chunk).
This class encapsulates the metadata from a PNG file's IHDR (Image Header) chunk,
including image dimensions, bit depth, color type, interlacing method, and CRC checksum.
It provides methods for parsing PNG files, packing/unpacking header data, and generating
PNG crc if omitted.
"""
width: int
height: int
bit_depth: int
color_type: int
crc: int
interlace: int = 0
def __init__( # pylint: disable=R0913
self,
size: tuple[int, int],
bit_depth: int,
color_type: int,
*,
crc: int | None = None,
interlace: int = 0,
):
"""Initialize a PNG IHDR (Image Header) chunk.
Args:
size (tuple[int, int]): The width and height of the image in pixels.
bit_depth (int): The number of bits per sample or per palette index (1, 2, 4, 8, or 16).
color_type (int): The color type of the image (0=Grayscale, 2=RGB, 3=Indexed,
4=Grayscale+Alpha, 6=RGBA).
crc (int | None, optional): CRC (Cyclic Redundancy Check) checksum for the chunk.
If None, it will be calculated automatically. Defaults
to None.
interlace (int, optional): The interlace method (0=no interlace, 1=Adam7 interlace).
Defaults to 0.
"""
self.width = size[0]
self.height = size[1]
self.bit_depth = bit_depth
self.color_type = color_type
self.interlace = interlace
self.crc = crc or self.compute_crc()
def compute_crc(self) -> int:
"""Compute the CRC checksum for the PNG IHDR chunk.
Returns:
int: The computed CRC checksum.
"""
return (
zlib.crc32(
b"IHDR"
+ self.width.to_bytes(4, "big")
+ self.height.to_bytes(4, "big")
+ bytes((self.bit_depth, self.color_type, 0, 0, self.interlace))
)
& 0xFFFFFFFF
)
@classmethod
def from_packed(cls, packed: int, crc: int | None = None) -> Self:
"""Create a PNG instance from a packed integer representation.
Parse a packed integer containing width, height, bit depth, color type and interlace
bits and return a PNG instance. The CRC may be supplied; if omitted it will be
computed by the constructor.
Args:
packed (int): Packed integer with fields laid out as described in the `packed` property.
crc (int | None): Optional CRC value for the IHDR chunk.
Returns:
Self: PNG instance with parsed fields.
"""
interlace = packed & 0b1
color = (packed >> 1) & 0b111
depth = (packed >> 4) & 0b111
height = (packed >> 7) & 0xFFFF
width = (packed >> 23) & 0xFFFF
return cls(
size=(width, height),
bit_depth=depth,
color_type=color,
crc=crc,
interlace=interlace,
)
@cached_property
def packed(self) -> int:
"""Pack IHDR fields into a single integer.
Returns:
int: Packed integer encoding the IHDR fields.
Layout:
width : 16 bits
height : 16 bits
depth : 3 bits
color : 3 bits
interlace : 1 bit
"""
return (
((self.width & 0xFFFF) << 23)
| ((self.height & 0xFFFF) << 7)
| ((self.bit_depth & 0b111) << 4)
| ((self.color_type & 0b111) << 1)
| (self.interlace & 0b1)
)
@classmethod
def from_bytearray(cls, data: bytearray) -> Self:
"""Create a PNG image instance from a bytearray containing PNG file data.
Parses the PNG signature and IHDR (Image Header) chunk to extract image metadata
including dimensions, bit depth, color type, and interlacing information.
Args:
data (bytearray): A bytearray containing the complete PNG file data, starting
with the PNG signature (8 bytes: \\x89PNG\\r\\n\\x1a\\n).
Returns:
Self | None: A new instance of the PNG image class with parsed IHDR metadata
(width, height, bit_depth, color_type, crc, interlace) if the data is
valid and contains a proper IHDR chunk. Returns None if the PNG signature
is invalid or the IHDR chunk type is not found.
Raises:
struct.error: If the data is too short or malformed for unpacking IHDR fields.
IndexError: If the bytearray is shorter than expected for the PNG structure.
"""
png_signature = b"\x89PNG\r\n\x1a\n"
if bytes(data[:8]) != png_signature:
raise ValueError("Invalid PNG signature")
offset = 8 # Read IHDR chunk
length = struct.unpack(">I", data[offset : offset + 4])[0]
offset += 4
chunk_type = data[offset : offset + 4]
offset += 4
if bytes(chunk_type) != b"IHDR":
raise ValueError("IHDR chunk not found at expected position")
ihdr_data = data[offset : offset + length]
offset += length
crc = struct.unpack(">I", data[offset : offset + 4])[0]
# Unpack IHDR fields
width, height, depth, color, _, _, interlace = struct.unpack(">IIBBBBB", ihdr_data)
return cls(
size=(width, height),
bit_depth=depth,
color_type=color,
crc=crc,
interlace=interlace,
)
def __repr__(self) -> str:
return (
f"PNG(width={self.width}, height={self.height}, "
f"bit_depth={self.bit_depth}, color_type={self.color_type}, "
f"interlace={self.interlace}, crc=0x{self.crc:08x})"
)
aperisolve/app.py
+205 -219
View File
@@ -1,5 +1,5 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# pylint: disable=C0413,W0718,R0903,R0801,R0915
# mypy: disable-error-code=unused-awaitable
"""Aperi'Solve Flask application."""
@@ -16,270 +16,256 @@ from flask import Flask, Response, abort, jsonify, render_template, request, sen
from redis import Redis
from rq import Queue
from .sentry import initialize_sentry
from .utils.sentry import initialize_sentry
initialize_sentry()
from .cleanup import cleanup_old_entries
from .config import IMAGE_EXTENSIONS, RESULT_FOLDER, WORKER_FILES
from .models import Image, Submission, db, init_db
app: Flask = Flask(__name__)
app.json.sort_keys = False # type: ignore
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URI")
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024))
RESULT_FOLDER.mkdir(parents=True, exist_ok=True)
db.init_app(app)
with app.app_context():
init_db(app)
redis_conn = Redis(host="redis", port=6379)
queue = Queue(connection=redis_conn)
from .models import Image, Submission, cleanup_old_entries, db
@app.errorhandler(413)
def too_large(_: Any) -> tuple[Response, int]:
"""Error Handler for Max File Size."""
sentry_sdk.set_context(
"upload",
{
"content_length": request.content_length,
"max_allowed": app.config["MAX_CONTENT_LENGTH"],
},
)
sentry_sdk.capture_message("Upload failed: 413 Payload Too Large", level="warning")
return jsonify({"error": "Image size exceeded"}), 413
def create_app() -> Flask:
"""Create flask application with routes."""
app = Flask(__name__)
app.json.sort_keys = False # type: ignore
app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("DB_URI")
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["MAX_CONTENT_LENGTH"] = int(os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024))
app.config["REDIS_QUEUE"] = Queue(connection=Redis(host="redis", port=6379))
db.init_app(app)
@app.errorhandler(413)
def too_large(_: Any) -> tuple[Response, int]:
"""Error Handler for Max File Size."""
sentry_sdk.set_context(
"upload",
{
"content_length": request.content_length,
"max_allowed": app.config["MAX_CONTENT_LENGTH"],
},
)
sentry_sdk.capture_message("Upload failed: 413 Payload Too Large", level="warning")
return jsonify({"error": "Image size exceeded"}), 413
@app.errorhandler(404)
def not_found(_: Any) -> str:
"""Error Handler for 404 not found."""
return render_template("error.html", message="Resource not found", statuscode=404)
@app.errorhandler(404)
def not_found(_: Any) -> str:
"""Error Handler for 404 not found."""
return render_template("error.html", message="Resource not found", statuscode=404)
@app.route("/")
def index() -> str:
"""Render the main index page."""
return render_template("index.html")
# Sentry automatically captures 500 errors
@app.route("/faq")
def faq() -> str:
"""Render the FAQ page."""
return render_template("faq.html")
@app.route("/show")
def show() -> str:
"""Show all active submissions."""
db_images = Image.query.all()
images = []
for img in db_images:
last_sub = img.submissions[-1]
images.append({"file": f"/image/{img.hash}", "link": f"/{last_sub.hash}"})
return render_template("show.html", images=images)
@app.route("/")
def index() -> str:
"""Render the main index page."""
return render_template("index.html")
@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
return render_template("result.html", hash_val=hash_val)
@app.route("/faq")
def faq() -> str:
"""Render the FAQ page."""
return render_template("faq.html")
@app.route("/upload", methods=["POST"])
def upload_image() -> tuple[Response, int]:
"""Handle image upload and initiate analysis."""
cleanup_old_entries()
@app.route("/show")
def show() -> str:
"""Show all active submissions."""
db_images = Image.query.all()
images = []
for img in db_images:
last_sub = img.submissions[-1]
images.append({"file": f"/image/{img.hash}", "link": f"/{last_sub.hash}"})
return render_template("show.html", images=images)
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"
@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
if image.filename is None or image.filename == "":
return jsonify({"error": "No image provided"}), 400
return render_template("result.html", hash_val=hash_val)
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()
@app.route("/upload", methods=["POST"])
def upload_image() -> tuple[Response, int]:
"""Handle image upload and initiate analysis."""
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
cleanup_old_entries()
# Check if hash is present on the DB
submission: Submission = Submission.query.filter_by(hash=submission_hash).first()
if "image" not in request.files:
return jsonify({"error": "No image provided"}), 400
# 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
image = request.files["image"]
password = request.form.get("password")
deep_analysis = request.form.get("deep") == "true"
# Self-Healing Logic
new_img_path = RESULT_FOLDER / img_hash / image_name
if image.filename is None or image.filename == "":
return jsonify({"error": "No image provided"}), 400
# 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)
if "." not in image.filename or Path(image.filename).suffix.lower() not in IMAGE_EXTENSIONS:
return jsonify({"error": "Unsupported file type"}), 400
# Try to find image in DB by hash
sub_img = Image.query.filter_by(hash=img_hash).first()
# 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()
# 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
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
# Increment upload count for the image
sub_img.upload_count += 1
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()
# Create new Submission entry
submission_path.mkdir(parents=True, exist_ok=True)
# Only return if BOTH the file exists AND the DB entry exists
if submission_path.exists() and submission is not None:
# 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
# Self-Healing Logic
new_img_path = RESULT_FOLDER / img_hash / image_name
@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
# 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)
@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]
return jsonify(
{
"image_path": "image/" + str(Path(image.file).name),
"names": names,
"size": image.size,
"first_submission_date": image.first_submission_date,
"last_submission_date": image.last_submission_date,
"upload_count": image.upload_count,
"passwords": passwords,
}
) # type: ignore
# Try to find image in DB by hash
sub_img = Image.query.filter_by(hash=img_hash).first()
@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
# 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
result_path = RESULT_FOLDER / str(image.hash) / str(submission.hash) / "results.json"
# Increment upload count for the image
sub_img.upload_count += 1
db.session.commit() # pylint: disable=no-member
if not result_path.exists():
return jsonify({"error": "Results not ready yet..."}), 425
# Create new Submission entry
submission_path.mkdir(parents=True, exist_ok=True)
with open(result_path, "r", encoding="utf-8") as f:
results = json.load(f)
# 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
return jsonify({"results": results}), 200
# Re-analysis case to prevent early return
submission.status = "pending" # type: ignore
db.session.commit() # pylint: disable=no-member
@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."""
# Start the analysis jobs
queue.enqueue("aperisolve.workers.analyze_image", submission.hash, job_timeout=300)
# Return submission identifier
return jsonify({"submission_hash": submission.hash}), 200
@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
@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]
return jsonify(
{
"image_path": "image/" + str(Path(image.file).name),
"names": names,
"size": image.size,
"first_submission_date": image.first_submission_date,
"last_submission_date": image.last_submission_date,
"upload_count": image.upload_count,
"passwords": passwords,
}
) # 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
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)
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."""
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 / f"{tool}.7z"
if tool not in WORKER_FILES or not output_file.exists():
abort(404, description="Tool output not found.")
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: 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
output_file = output_dir / f"{tool}.7z"
if (not output_file.exists()) or (output_file.suffix.lower() not in IMAGE_EXTENSIONS):
return abort(404, description="Image not found or unsupported format")
if tool not in WORKER_FILES or not output_file.exists():
abort(404, description="Tool output not found.")
return send_file(output_file, as_attachment=True)
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: 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)
return app
if __name__ == "__main__":
debug_mode = os.environ.get("FLASK_DEBUG", "false").lower() == "true"
app.run(host="0.0.0.0", port=5000, debug=debug_mode)
my_app = create_app()
debug_mode = os.environ.get("FLASK_DEBUG", "0") == "1"
my_app.run(host="0.0.0.0", port=5000, debug=debug_mode)
aperisolve/cleanup.py
-79
View File
@@ -1,79 +0,0 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Cleanup old entries in the database and file system."""
import shutil
import time
from datetime import datetime, timezone
from os import getenv
from pathlib import Path
from .models import Image, Submission, db
RESULT_FOLDER = Path("./results")
MAX_PENDING_TIME = int(getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default
MAX_STORE_TIME = int(getenv("MAX_STORE_TIME", "259200")) # 3 days by default
def cleanup_old_entries() -> None:
"""
Clean up old and incomplete entries from the database and file system.
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
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
shutil.rmtree(result_path)
# Delete "old" images
for img in Image.query.all(): # type: ignore
if img.last_submission_date.tzinfo is None:
img_date = img.last_submission_date.replace(tzinfo=timezone.utc)
else:
img_date = img.last_submission_date
delay = datetime.now(timezone.utc) - img_date
img_fold = 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
# 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
db.session.commit() # pylint: disable=no-member
aperisolve/config.py
+5 -1
View File
@@ -3,9 +3,13 @@
# mypy: disable-error-code=unused-awaitable
"""Aperi'Solve configuration variables."""
from os import getenv
from pathlib import Path
IMAGE_EXTENSIONS = [".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".tiff"]
WORKER_FILES = ["binwalk", "foremost", "steghide", "zsteg", "image_resize", "openstego"]
WORKER_FILES = ["binwalk", "foremost", "steghide", "zsteg", "image_resize", "openstego", "pcrt"]
RESULT_FOLDER = Path(__file__).parent.resolve() / "results"
MAX_PENDING_TIME = int(getenv("MAX_PENDING_TIME", "600")) # 10 minutes by default
MAX_STORE_TIME = int(getenv("MAX_STORE_TIME", "259200")) # 3 days by default
CLEAR_AT_RESTART = int(getenv("CLEAR_AT_RESTART", "0"))
aperisolve/models.py
+160 -93
View File
@@ -1,20 +1,29 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# 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."""
import itertools
import sys
import shutil
import struct
import time
import zlib
from datetime import datetime, timezone
from os import getenv
from pathlib import Path
from shutil import rmtree
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import BigInteger, Boolean, Column, DateTime, Float, Integer, String
from sqlalchemy import (
BigInteger,
Boolean,
Column,
DateTime,
Float,
Integer,
SmallInteger,
String,
)
from aperisolve.analyzers.utils import PNG, get_resolutions, get_valid_depth_color_pairs
from aperisolve.config import MAX_PENDING_TIME, MAX_STORE_TIME, RESULT_FOLDER
from aperisolve.utils.utils import get_resolutions, get_valid_depth_color_pairs
db: SQLAlchemy = SQLAlchemy()
@@ -56,102 +65,160 @@ class Submission(db.Model): # type: ignore
class IHDR(db.Model): # type: ignore
"""IHDR CRC lookup table"""
"""IHDR CRC lookup table with direct parameter storage."""
iid = Column(Integer, primary_key=True, autoincrement=True)
crc = Column(BigInteger, nullable=False)
packed = Column(BigInteger, nullable=False)
crc = Column(BigInteger, nullable=False, index=True)
width = Column(Integer, nullable=False)
height = Column(Integer, nullable=False)
bit_depth = Column(SmallInteger, nullable=False)
color_type = Column(SmallInteger, nullable=False)
interlace = Column(SmallInteger, nullable=False)
def to_ihdr_bytes(self) -> bytes:
"""Convert database record to IHDR chunk data (13 bytes)."""
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,
]
)
)
@staticmethod
def compute_crc(
width: int, height: int, bit_depth: int, color_type: int, interlace: int
) -> int:
"""Compute CRC32 for IHDR chunk."""
ihdr_data = (
struct.pack(">I", width)
+ struct.pack(">I", height)
+ bytes([bit_depth, color_type, 0, 0, interlace])
)
return zlib.crc32(b"IHDR" + ihdr_data) & 0xFFFFFFFF
def fill_ihdr_db() -> None:
"""
Creates a database of common PNG IHDR (Image Header) entries by generating
combinations of standard PNG parameters such as resolutions, bit depths,
color types, and interlace methods. The function checks for existing entries
in the database to avoid duplicates and inserts new entries in batches of
5000, committing the changes to the database periodically.
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.
"""
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,
)
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,
width=width,
height=height,
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}")
db.session.rollback()
def cleanup_old_entries() -> None:
"""
Clean up old and incomplete entries from the database and file system.
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
"""
# Skip if table already has entries
if db.session.query(IHDR.iid).first() is not None:
print("IHDR table already populated, skipping fill.")
return
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
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
shutil.rmtree(result_path)
# Standard PNG IHDR parameters
resolutions = get_resolutions()
bit_color_pairs = get_valid_depth_color_pairs()
interlace_methods = [0, 1]
# Delete "old" images
for img in Image.query.all(): # type: ignore
if img.last_submission_date.tzinfo is None:
img_date = img.last_submission_date.replace(tzinfo=timezone.utc)
else:
img_date = img.last_submission_date
delay = datetime.now(timezone.utc) - img_date
img_fold = 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
combinations = itertools.product(
resolutions,
bit_color_pairs,
interlace_methods,
)
count = 0
with db.session.no_autoflush: # pylint: disable=no-member
for size, (bd, ct), inter in combinations:
p = PNG(size=size, bit_depth=bd, color_type=ct, interlace=inter)
exists = db.session.execute( # pylint: disable=no-member
db.select(IHDR.iid).where(IHDR.packed == p.packed)
).first()
if exists: # Skip if already present
continue
db.session.add(IHDR(packed=p.packed, crc=p.crc)) # pylint: disable=no-member
count += 1
if count % 5000 == 0:
print(f"Inserted {count} common IHDR entries...")
db.session.commit()
# 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
db.session.commit() # pylint: disable=no-member
print(f"Precomputed {count} common IHDR entries. IHDR table filled successfully.")
def init_db(app: Flask) -> None:
"""
Initialize the database by creating tables and populating lookup data.
This function creates the database schema if it doesn't already exist,
and populates the CRC lookup table for PNG IHDR validation.
Args:
app (Flask): The Flask application instance used to establish the application context
for database operations.
Returns:
None
Side Effects:
- Creates all database tables if the "image" table doesn't exist
- Populates the IHDR CRC lookup table with initial data
- Prints status messages to console during initialization
"""
# Detect if running as RQ worker by checking command line
# Worker: /usr/local/bin/rq worker ...
# Web: /usr/local/bin/flask run ... or gunicorn/wsgi
is_worker = any("rq" in arg and "flask" not in arg for arg in sys.argv[:2])
with app.app_context():
# Workers should never clear the database
if not is_worker and getenv("CLEAR_AT_RESTART", "0") == "1":
print("Clearing database and file system at restart...")
db.session.remove() # pylint: disable=no-member
db.drop_all()
rmtree(Path("./results"), ignore_errors=True) # Clear results folder
# Always create tables (idempotent - safe to call multiple times)
db.create_all()
print("Database structure created successfully.")
# Workers should never fill IHDR table
if is_worker:
print("Running as worker, skipping IHDR table fill.")
else:
fill_ihdr_db()
aperisolve/requirements-dev.txt
+7
View File
@@ -0,0 +1,7 @@
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/static/css/aperisolve.css
+64 -21
View File
@@ -224,6 +224,7 @@ pre {
background-color: var(--dark-bg-hover);
box-shadow: 0 0 10px var(--theme-green);
}
#upload-form {
transition: height 0.4s ease, padding 0.4s ease, margin 0.4s ease;
overflow: hidden;
@@ -276,6 +277,15 @@ td {
min-width: 150px;
}
.a-pcrt .results_img {
width: 100%;
margin: auto;
text-align: center;
}
#result .a-pcrt .results_img img{
width: 75%;
}
/* ======== Modal ======== */
#image-modal-container {
position: fixed;
@@ -441,10 +451,12 @@ footer a i {
box-shadow: 0 0 10px var(--theme-green-shadow);
}
#httperror{
#httperror {
padding-top: 100px;
}
#httperror h1, #httperror h2{
#httperror h1,
#httperror h2 {
text-align: center;
}
@@ -480,7 +492,7 @@ footer a i {
flex-direction: column;
}
#result-infos > div {
#result-infos>div {
width: 100%;
flex: 1 1 100%;
}
@@ -489,7 +501,8 @@ footer a i {
font-size: 0.9rem;
}
th, td {
th,
td {
padding: 0.5rem;
}
}
@@ -523,7 +536,11 @@ footer a i {
width: 50px;
}
.card, .form-control, .table, .border, .card-hack {
.card,
.form-control,
.table,
.border,
.card-hack {
border-radius: 0.35rem;
}
@@ -549,7 +566,8 @@ footer a i {
padding: 5px;
}
textarea.form-control, pre {
textarea.form-control,
pre {
padding: 15px;
font-size: 0.85rem;
}
@@ -573,7 +591,8 @@ footer a i {
min-width: 400px;
}
th, td {
th,
td {
padding: 0.4rem;
word-break: break-word;
}
@@ -594,7 +613,7 @@ footer a i {
}
/* FAQ/Cheatsheet adjustments */
#cheatsheet > div {
#cheatsheet>div {
padding: 8px;
margin-bottom: 30px;
}
@@ -652,7 +671,9 @@ footer a i {
font-size: 0.85rem;
}
.btn, .btn-primary, .btn-hack {
.btn,
.btn-primary,
.btn-hack {
width: 100%;
padding: 12px;
font-size: 1rem;
@@ -668,11 +689,12 @@ footer a i {
padding: 5px;
}
#result-infos > div {
#result-infos>div {
padding: 5px;
}
textarea.form-control, pre {
textarea.form-control,
pre {
padding: 10px;
font-size: 0.8rem;
}
@@ -689,7 +711,8 @@ footer a i {
font-size: 0.75rem;
}
th, td {
th,
td {
padding: 0.3rem;
}
@@ -719,7 +742,7 @@ footer a i {
margin-top: 15px;
}
.analyzer h2 + div {
.analyzer h2+div {
margin-top: 10px;
}
@@ -757,7 +780,9 @@ footer a i {
width: 35px;
}
.btn, .btn-primary, .btn-hack {
.btn,
.btn-primary,
.btn-hack {
padding: 10px;
font-size: 0.9rem;
}
@@ -779,15 +804,21 @@ footer a i {
opacity: 0.7;
}
.btn, .btn-primary, .btn-hack {
.btn,
.btn-primary,
.btn-hack {
min-height: 44px;
}
input, textarea, select {
font-size: 16px; /* Prevents zoom on iOS */
input,
textarea,
select {
font-size: 16px;
/* Prevents zoom on iOS */
}
a, button {
a,
button {
min-height: 44px;
min-width: 44px;
}
@@ -817,18 +848,30 @@ footer a i {
color: black;
}
.card, .form-control, .table, .border, .card-hack {
.card,
.form-control,
.table,
.border,
.card-hack {
background-color: white;
border-color: #333 !important;
box-shadow: none;
}
h1, h2, h3, h4, h5, h6, strong, .highlight {
h1,
h2,
h3,
h4,
h5,
h6,
strong,
.highlight {
color: #333;
text-shadow: none;
}
a, .text-accent {
a,
.text-accent {
color: #333;
}
}
aperisolve/static/js/aperisolve.js
+24 -24
View File
@@ -5,11 +5,11 @@ const TOOL_ORDER = [
"foremost",
"outguess",
"pngcheck",
"pcrt",
"steghide",
"openstego",
"zsteg",
"strings",
"image_resize",
];
function escapeHtml(text) {
@@ -97,24 +97,24 @@ function showInfo(message, reset) {
*/
document.addEventListener('click', function (event) {
// Check if the clicked element is a copy icon
if (event.target.classList.contains('copy-icon')) {
const icon = event.target;
const textarea = icon.closest('.textarea-container').querySelector('textarea');
// Check if the clicked element is a copy icon
if (event.target.classList.contains('copy-icon')) {
const icon = event.target;
const textarea = icon.closest('.textarea-container').querySelector('textarea');
if (textarea) {
textarea.select();
navigator.clipboard.writeText(textarea.value).then(() => {
icon.classList.remove('fa-copy');
icon.classList.add('fa-check');
setTimeout(() => {
icon.classList.remove('fa-check');
icon.classList.add('fa-copy');
}, 1500);
});
}
if (textarea) {
textarea.select();
navigator.clipboard.writeText(textarea.value).then(() => {
icon.classList.remove('fa-copy');
icon.classList.add('fa-check');
setTimeout(() => {
icon.classList.remove('fa-check');
icon.classList.add('fa-copy');
}, 1500);
});
}
});
}
});
/**
* Image preview functions
@@ -250,7 +250,7 @@ function parseResult(result) {
analyzer.className = `analyzer a-${escapeHtml(tool)}`;
resultDiv.appendChild(analyzer);
analyzer.innerHTML += `<h2>${capitalize(tool.replace("_"," "))}</h2>`;
analyzer.innerHTML += `<h2>${capitalize(tool.replace("_", " "))}</h2>`;
// Parse text output
if (typeof result[tool]["output"] === "string") {
@@ -314,7 +314,7 @@ function parseResult(result) {
if ("png_images" in result[tool]) {
for (const image of result[tool]["png_images"]) {
analyzer.innerHTML += `<div class='results_img'><img src='${escapeHtml(image)}'/></div>`;
analyzer.innerHTML += `<div class='results_img'><img src='${escapeHtml(image)}'/></div>`;
}
}
@@ -347,9 +347,9 @@ async function pollStatus(submission_hash) {
const resultResp = await fetch(`/result/${submission_hash}`);
const resultData = await resultResp.json();
fetchImageInfo(submission_hash);
if ("error" in resultData){
if ("error" in resultData) {
showWarning(`❌ ${resultData.error}`, true);
}else if ("results" in resultData) {
} else if ("results" in resultData) {
parseResult(resultData.results);
}
} else if (statusData.status === "error") {
@@ -360,9 +360,9 @@ async function pollStatus(submission_hash) {
try {
const resultResp = await fetch(`/result/${submission_hash}`);
const resultData = await resultResp.json();
if ("error" in resultData){
if ("error" in resultData) {
// showWarning(`❌ ${resultData.error}`, true);
}else if ("results" in resultData) {
} else if ("results" in resultData) {
fetchImageInfo(submission_hash);
parseResult(resultData.results);
}
@@ -488,7 +488,7 @@ if (browseBtn) {
}
} else if (xhr.status == 413) {
showDanger("❌ File too large. Please upload a smaller file.", true);
}else {
} else {
showDanger(`❌ HTTP error ${xhr.status}`, true);
}
progressBar.style.width = "100%";
aperisolve/utils/__init__.py
View File
aperisolve/utils/init_db.py
+62
View File
@@ -0,0 +1,62 @@
"""
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.
"""
import os
from shutil import rmtree
from sqlalchemy import inspect
from ..app import create_app
from ..config import RESULT_FOLDER
from ..models import db, fill_ihdr_db
def main() -> None:
"""Database initialization main function."""
reset = os.getenv("CLEAR_AT_RESTART") == "1"
app = create_app()
with app.app_context():
inspector = inspect(db.engine)
tables = inspector.get_table_names()
if reset:
# Delete result folder content
if RESULT_FOLDER.exists():
for item in RESULT_FOLDER.iterdir():
try:
if item.is_dir():
rmtree(item)
else:
item.unlink()
except Exception:
continue
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...")
db.create_all()
else:
print("🗄️ Database schema already exists")
print("📦 Filling IHDR static data...")
fill_ihdr_db()
print("✅ Database initialization complete")
RESULT_FOLDER.mkdir(parents=True, exist_ok=True)
if __name__ == "__main__":
main()
aperisolve/utils/png.py
+402
View File
@@ -0,0 +1,402 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801,R0902
# mypy: disable-error-code=unused-awaitable
"""PNG Class for analyzer modules"""
import itertools
import re
import struct
import zlib
from typing import Any, Optional
from ..app import create_app
from ..models import IHDR
from .utils import int2hex, str2hex
__author__ = [
"Zeecka",
"indonumberone (https://github.com/indonumberone/PCRT3)",
"Etr1x (https://github.com/Etr1x/PCRT3)",
"sherlly (https://github.com/sherlly/PCRT)",
]
class PNG:
"""PNG file analyzer and repair tool."""
def __init__(self, data: bytes) -> None:
self.data = data
self.width = self.height = self.bits = self.mode = 0
self.compression = self.filter = self.interlace = self.channel = 0
self.txt_content: dict[bytes, list[bytes]] = {}
self.image_content: dict[bytes, list[bytes]] = {}
self.crcs: dict[bytes, Any] = {}
self.repaired_data = bytearray()
self.logs: list[str] = []
self.errors: list[str] = []
def _log(self, msg: str) -> None:
"""Add message to logs."""
self.logs.append(msg)
def _error(self, msg: str) -> None:
"""Add error message."""
self.errors.append(msg)
def _check_format(self, data: bytes) -> bool:
"""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]:
"""Check CRC of chunk."""
calc_crc = struct.pack("!I", zlib.crc32(chunk_type + chunk_data))
return calc_crc if calc_crc != checksum else None
def _find_ihdr(self, data: bytes) -> tuple[int, bytes]:
"""Find IHDR chunk in PNG data."""
pos = data.find(b"IHDR")
if pos == -1:
return -1, b""
idat_begin = data.find(b"IDAT")
return pos, data[pos - 4 : idat_begin - 4 if idat_begin != -1 else pos + 21]
def check_header(self) -> bool:
"""Check and fix PNG header."""
header = self.data[:8]
correct = b"\x89\x50\x4e\x47\x0d\x0a\x1a\x0a"
if header != correct:
self._log(f"Wrong PNG header: {str2hex(header)}")
self._log("Fixed header to: 89504E470D0A1A0A")
self.repaired_data.extend(correct)
return True
self._log("Correct PNG header")
self.repaired_data.extend(header)
return False
def get_pic_info(self, ihdr: bytes = b"") -> bool:
"""Extract picture information from IHDR chunk."""
if not ihdr:
pos, ihdr_chunk = self._find_ihdr(self.data)
if pos == -1:
self._error("Lost IHDR chunk")
return False
length = struct.unpack("!I", ihdr_chunk[:4])[0]
ihdr = ihdr_chunk[8 : 8 + length]
(
self.width,
self.height,
self.bits,
self.mode,
self.compression,
self.filter,
self.interlace,
) = struct.unpack("!iiBBBBB", ihdr)
self.interlace = int(ihdr[12])
self.channel = {0: 1, 3: 1, 2: 3, 4: 2, 6: 4}.get(self.mode, 0)
self.txt_content, self.image_content, self.crcs = self._find_ancillary(self.data)
return True
def _find_ancillary(
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 = [
b"cHRM",
b"pHYs",
b"gAMA",
b"sBIT",
b"PLTE",
b"bKGD",
b"sTER",
b"hIST",
b"iCCP",
b"sPLT",
b"sRGB",
b"dSIG",
b"tIME",
b"tRNS",
b"oFFs",
b"sCAL",
b"fRAc",
b"gIFg",
b"gIFt",
b"gIFx",
]
attach_txt = [b"eXIf", b"iTXt", b"tEXt", b"zTXt"]
# Initialize result dictionaries
image_content: dict[bytes, list[bytes]] = {chunk: [] for chunk in ancillary}
txt_content: dict[bytes, list[bytes]] = {chunk: [] for chunk in attach_txt}
crcs: dict[bytes, Any] = {chunk: [] for chunk in ancillary}
# Single pass through data
pos = 8 # Skip PNG signature
while pos < len(data) - 12: # Need at least 12 bytes for chunk header
# Try to read chunk length
try:
length = struct.unpack("!I", data[pos : pos + 4])[0]
chunk_type = data[pos + 4 : pos + 8]
# Check if this is a chunk we care about
if chunk_type in ancillary:
chunk_data = data[pos + 8 : pos + 8 + length]
image_content[chunk_type].append(chunk_data)
# Check CRC for first occurrence only
if len(image_content[chunk_type]) == 1:
crc_data = data[pos + 8 + length : pos + 12 + length]
calc_crc = self._check_crc(chunk_type, chunk_data, crc_data)
crcs[chunk_type] = (calc_crc, crc_data) if calc_crc else []
elif chunk_type in attach_txt:
chunk_data = data[pos + 8 : pos + 8 + length]
txt_content[chunk_type].append(chunk_data)
# Move to next chunk
pos += 12 + length
except (struct.error, IndexError):
# Malformed chunk, skip ahead
pos += 1
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."""
critical_ancillary = [
b"PLTE",
b"tRNS",
b"cHRM",
b"gAMA",
b"iCCP",
b"sBIT",
b"sRGB",
b"bKGD",
b"hIST",
b"pHYs",
b"sPLT",
]
ihdr_end = len(self.repaired_data)
idat_pos = self.data.find(b"IDAT")
if idat_pos == -1:
return False
search_start = self.data.find(b"IHDR")
if search_start == -1:
return False
# Skip past the IHDR chunk
ihdr_length = struct.unpack("!I", self.data[search_start - 4 : search_start])[0]
search_start = search_start + 4 + ihdr_length + 4
# Collect all ancillary chunks before IDAT
for chunk_type in critical_ancillary:
pos = search_start
while pos < idat_pos:
pos = self.data.find(chunk_type, pos, idat_pos)
if pos == -1:
break
chunk_start = pos - 4
length = struct.unpack("!I", self.data[chunk_start:pos])[0]
chunk_data = self.data[pos + 4 : pos + 4 + length]
chunk_crc = self.data[pos + 4 + length : pos + 8 + length]
# Validate CRC before copying
if calc_crc := self._check_crc(chunk_type, chunk_data, chunk_crc):
self._log(f"Warning: {chunk_type.decode()} has invalid CRC, fixing...")
chunk_crc = calc_crc
# Reconstruct complete chunk with validated CRC
complete_chunk = struct.pack("!I", length) + chunk_type + chunk_data + chunk_crc
self.repaired_data[ihdr_end:ihdr_end] = complete_chunk
ihdr_end += len(complete_chunk)
self._log(f"Copied {chunk_type.decode()} chunk ({length} bytes)")
pos = pos + 4 + length + 4
return True
def check_ihdr(self) -> bool:
"""Check and repair IHDR chunk using database lookup - WITH OPTIMIZED EXHAUSTIVE SEARCH."""
pos, ihdr = self._find_ihdr(self.data)
if pos == -1:
self._error("Lost IHDR chunk")
return False
length = struct.unpack("!I", ihdr[:4])[0]
chunk_type, chunk_ihdr = ihdr[4:8], ihdr[8 : 8 + length]
crc = ihdr[8 + length : 12 + length]
fixed = False
if calc_crc := self._check_crc(chunk_type, chunk_ihdr, crc):
self._log(f"Error IHDR CRC found at offset {int2hex(pos + 4 + length)}")
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 not fixed:
self._error("Could not recover IHDR dimensions")
else:
self._log(f"Correct IHDR CRC at offset {int2hex(pos + 4 + length)}")
self.repaired_data.extend(ihdr)
self._log(f"IHDR chunk check complete at offset {int2hex(pos - 4)}")
self.get_pic_info(ihdr=chunk_ihdr if not fixed else ihdr[8 : 8 + length])
return fixed
def _fix_dos2unix(
self, chunk_type: bytes, chunk_data: bytes, crc: bytes, count: int
) -> Optional[bytes]:
"""Fix DOS to Unix line ending conversion."""
pos_list = []
pos = -1
while (pos := chunk_data.find(b"\x0a", pos + 1)) != -1:
pos_list.append(pos)
for pos_combo in itertools.combinations(pos_list, count):
test_data = chunk_data
for i, pos in enumerate(pos_combo):
test_data = test_data[: pos + i] + b"\x0d" + test_data[pos + i :]
if not self._check_crc(chunk_type, test_data, crc):
return test_data
return None
def check_idat(self) -> bool:
"""Check and repair IDAT chunks."""
idat_begin = self.data.find(b"IDAT") - 4
if idat_begin == -5:
self._error("Lost all IDAT chunks")
return False
pos_iend = self.data.find(b"IEND")
pos_list = [
g.start()
for g in re.finditer(b"IDAT", self.data)
if pos_iend == -1 or g.start() < pos_iend
]
idat_table = []
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)]
)
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:]
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))
):
idat_chunk = idat_chunk[:8] + fixed_data + idat_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
self.repaired_data.extend(idat_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]]:
"""Check and repair IEND chunk."""
standard_iend = b"\x00\x00\x00\x00IEND\xae\x42\x60\x82"
pos = self.data.find(b"IEND")
fixed = False
extra_data = None
if pos == -1:
self._log("Lost IEND chunk, adding standard IEND")
iend = standard_iend
fixed = True
else:
iend = self.data[pos - 4 : pos + 8]
if iend != standard_iend:
self._log("Error IEND chunk, fixing...")
iend = standard_iend
fixed = True
else:
self._log("Correct IEND chunk")
if extra_data := self.data[pos + 8 :]:
self._log(f"Found {len(extra_data)} bytes after IEND: {extra_data[:20]!r}")
self.repaired_data.extend(iend)
self._log("IEND chunk check complete")
return fixed, extra_data
def repair(self) -> tuple[bool, Optional[bytes]]:
"""Run full PNG repair process."""
if not self._check_format(self.data):
self._error("File may not be a PNG image")
return False, None
fixed = False
fixed |= self.check_header()
fixed |= self.check_ihdr()
fixed |= self.check_chunks()
fixed |= self.check_idat()
iend_fixed, extra_data = self.check_iend()
fixed |= iend_fixed
return fixed, extra_data
aperisolve/sentry.py → aperisolve/utils/sentry.py
View File
aperisolve/utils/utils.py
+104
View File
@@ -0,0 +1,104 @@
# 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
def str2hex(s: bytes) -> str:
"""Convert bytes to hex string."""
return binascii.hexlify(s).decode().upper()
def int2hex(i: int) -> str:
"""Convert int to hex string with 0x prefix."""
return "0x" + hex(i)[2:].upper()
def get_resolutions() -> list[tuple[int, int]]:
"""
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
paper standards. Includes both portrait and landscape orientations.
Base widths range from 16px to 10000px, covering:
- Low resolution: 16-256px (16px increments)
- Standard: 320-1024px (32px increments)
- HD/Full HD: 1280-2560px (64px increments)
- 4K: 3000-4096px (128px increments)
- 5K-8K: 5120-8192px (256px increments)
- Upper bound: 10000px
Aspect ratios include common formats:
- Digital screens: 1:1, 4:3, 3:2, 16:10, 16:9, 21:9
- Photography/print: 5:4, 7:5, 2:3
- Paper standards: ISO A-series, US Letter, Legal, Tabloid
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
)
aspect_ratios = [
# Screens / digital
(1, 1), # square
(4, 3),
(3, 2),
(16, 10),
(16, 9),
(21, 9),
# Photography / print
(5, 4), # 8x10
(7, 5), # 5x7
(3, 2), # 4x6 (already included, kept for clarity)
(2, 3), # poster
# Paper standards
(1414, 1000), # ISO A-series (A4, A3, A2, ...)
(11, 85), # US Letter
(14, 85), # Legal
(17, 11), # Tabloid
]
resolutions: set[tuple[int, int]] = set()
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:
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.
Yields tuples of (depth, color) representing valid color mode and bit depth
combinations. Supports grayscale (0), RGB (2, 3, 4, 6) color modes with
appropriate bit depths (1, 2, 4, 8, 16 bits).
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],
2: [8, 16],
3: [1, 2, 4, 8],
4: [8, 16],
6: [8, 16],
}
for color, depths in valid_color_depth.items():
for depth in depths:
yield depth, color
aperisolve/wsgi.py → aperisolve/utils/wsgi.py
+2 -3
View File
@@ -3,7 +3,6 @@
# mypy: disable-error-code=unused-awaitable
"""Aperi'Solve WSGI runner."""
from .app import app as application
from aperisolve.app import create_app
if __name__ == "__main__":
application.run()
application = create_app()
aperisolve/workers.py
+6 -5
View File
@@ -9,7 +9,7 @@ from typing import Any
import sentry_sdk
from .sentry import initialize_sentry
from .utils.sentry import initialize_sentry
initialize_sentry()
@@ -17,16 +17,16 @@ from .analyzers.binwalk import analyze_binwalk # pylint: disable=C0413
from .analyzers.decomposer import analyze_decomposer
from .analyzers.exiftool import analyze_exiftool
from .analyzers.foremost import analyze_foremost
from .analyzers.image_resize import analyze_image_resize
from .analyzers.openstego import analyze_openstego
from .analyzers.outguess import analyze_outguess
from .analyzers.pcrt import analyze_pcrt
from .analyzers.pngcheck import analyze_pngcheck
from .analyzers.steghide import analyze_steghide
from .analyzers.strings import analyze_strings
from .analyzers.zsteg import analyze_zsteg
from .app import app, db
from .app import create_app
from .config import RESULT_FOLDER
from .models import Image, Submission
from .models import Image, Submission, db
def analyze_image(submission_hash: str) -> None:
@@ -55,6 +55,7 @@ def analyze_image(submission_hash: str) -> None:
- Creates result directories and generates analysis output files
- Modifies the database session and commits changes
"""
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
@@ -103,9 +104,9 @@ def analyze_image(submission_hash: str) -> None:
(analyze_decomposer, img_path, result_path),
(analyze_exiftool, img_path, result_path),
(analyze_foremost, img_path, result_path),
(analyze_image_resize, img_path, result_path),
(analyze_openstego, img_path, result_path, submission.password),
(analyze_pngcheck, img_path, result_path),
(analyze_pcrt, img_path, result_path),
(analyze_strings, img_path, result_path),
(analyze_steghide, img_path, result_path, submission.password),
(analyze_zsteg, img_path, result_path),
compose.dev.yml
+21 -10
View File
@@ -12,16 +12,14 @@ services:
depends_on:
postgres:
condition: service_healthy
initdb:
condition: service_completed_successfully
redis:
condition: service_started
worker:
condition: service_started
env_file:
- .env.example
environment:
- FLASK_DEBUG=true
- FLASK_ENV=development
- CLEAR_AT_RESTART=0 # Don't clear on Flask debug auto-reload
worker:
#image: ghcr.io/zeecka/aperisolve:latest
@@ -32,14 +30,27 @@ services:
volumes:
- ./aperisolve:/app/aperisolve
depends_on:
- redis
- postgres
postgres:
condition: service_healthy
initdb:
condition: service_completed_successfully
redis:
condition: service_started
env_file:
- .env.example
environment:
- FLASK_DEBUG=true
- FLASK_ENV=development
- CLEAR_AT_RESTART=0 # Always false for worker to prevent double clearing
initdb:
build: .
container_name: aperisolve-initdb
command: python -m aperisolve.utils.init_db
volumes:
- ./aperisolve:/app/aperisolve
depends_on:
postgres:
condition: service_healthy
env_file:
- .env.example
restart: "no"
redis:
image: redis:7
compose.yml
+27 -10
View File
@@ -1,9 +1,9 @@
services:
web:
image: ghcr.io/zeecka/aperisolve:latest
#build: .
#image: ghcr.io/zeecka/aperisolve:latest
build: .
container_name: aperisolve-web
restart: always
restart: no
ports:
- "5000:5000"
volumes:
@@ -11,6 +11,8 @@ services:
depends_on:
postgres:
condition: service_healthy
initdb:
condition: service_completed_successfully
redis:
condition: service_started
worker:
@@ -25,20 +27,22 @@ services:
start_period: 900s
worker:
image: ghcr.io/zeecka/aperisolve:latest
#build: .
#image: ghcr.io/zeecka/aperisolve:latest
build: .
container_name: aperisolve-worker
restart: always
restart: no
command: rq worker default --url redis://redis:6379/0
volumes:
- apresults:/app/aperisolve/results
depends_on:
- redis
- postgres
postgres:
condition: service_healthy
initdb:
condition: service_completed_successfully
redis:
condition: service_started
env_file:
- .env
environment:
- CLEAR_AT_RESTART=0 # Always false for worker to prevent double clearing
postgres:
image: postgres:18
@@ -54,6 +58,19 @@ services:
env_file:
- .env
initdb:
build: .
container_name: aperisolve-initdb
command: python -m aperisolve.utils.init_db
volumes:
- apresults:/app/aperisolve/results
depends_on:
postgres:
condition: service_healthy
env_file:
- .env
restart: "no"
redis:
image: redis:7
container_name: redis
docs/adding-analyzer.md
+23 -1
View File
@@ -5,7 +5,7 @@ This guide shows you how to add a new image analysis tool to the system.
## Quick Start
1. Create a new file in `analyzers/` from template file `template_analyzer.py`.
2. Register it in `workers.py`
2. Register it in `workers.py`, `config.py` and `aperisolve.js`.
## Basic Template
@@ -52,6 +52,8 @@ class ForemostAnalyzer(SubprocessAnalyzer):
Note the `has_archive=True` parameter.
**If a tool return an archive, it's important to add the worker name in `WORKER_FILES` variable of `config.py`.**
### Tool With Password Support
Like `steghide` or `openstego`:
@@ -139,6 +141,26 @@ if submission.deep_analysis:
])
```
And also in `aperisolve.js` in the `TOOL_ORDER` constant:
```js
const TOOL_ORDER = [
"decomposer",
"exiftool",
"binwalk",
"mytool" // My new tool
"foremost",
"outguess",
"pngcheck",
"pcrt",
"steghide",
"openstego",
"zsteg",
"strings",
"image_resize",
];
```
## Available Methods to Override
| Method | Purpose | Default Behavior |