From 2ed2b219defe145813d6dcfee50ef6647152362a Mon Sep 17 00:00:00 2001 From: Zeecka Date: Tue, 6 Jan 2026 04:37:21 +0100 Subject: [PATCH 01/13] Add PCRT --- aperisolve/analyzers/base_analyzer.py | 4 +- aperisolve/analyzers/pcrt.py | 385 ++++++++++++++++++++++++++ aperisolve/config.py | 2 +- aperisolve/static/js/aperisolve.js | 1 + aperisolve/workers.py | 2 + docs/adding-analyzer.md | 24 +- 6 files changed, 415 insertions(+), 3 deletions(-) create mode 100644 aperisolve/analyzers/pcrt.py diff --git a/aperisolve/analyzers/base_analyzer.py b/aperisolve/analyzers/base_analyzer.py index 2ead24a..42edcb7 100644 --- a/aperisolve/analyzers/base_analyzer.py +++ b/aperisolve/analyzers/base_analyzer.py @@ -50,8 +50,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 diff --git a/aperisolve/analyzers/pcrt.py b/aperisolve/analyzers/pcrt.py new file mode 100644 index 0000000..9216225 --- /dev/null +++ b/aperisolve/analyzers/pcrt.py @@ -0,0 +1,385 @@ +# flake8: noqa: E203,E501,W503 +# pylint: disable=C0413,W0718,R0903,R0801 +# mypy: disable-error-code=unused-awaitable +"""PCRT (PNG Check & Repair Tool) Analyzer for Image Submissions.""" + +__author__ = [ + "Zeecka", + "indonumberone (https://github.com/indonumberone/PCRT3)", + "Etr1x (https://github.com/Etr1x/PCRT3)", + "sherlly (https://github.com/sherlly/PCRT)", +] + +import binascii +import itertools +import re +import struct +import zlib +from pathlib import Path +from typing import Any, Optional + +from .base_analyzer import SubprocessAnalyzer + + +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 str2num(s: bytes, n: int = 0) -> int: + """Convert bytes to number.""" + return struct.unpack("!I", s)[0] if n == 4 else int(str2hex(s), 16) + + +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 = {} + self.image_content: dict = {} + self.crcs: dict = {} + 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"PNG", 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, dict, dict]: + """Find ancillary chunks in PNG data.""" + 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"] + image_content: dict = {} + crcs: dict = {} + + for data_i in ancillary: + image_content[data_i] = [] + crcs[data_i] = [] + pos = 0 + while (pos := data.find(data_i, pos)) != -1: + length = str2num(data[pos - 4 : pos]) + image_content[data_i].append(data[pos + 4 : pos + 4 + length]) + crc_data = data[pos + 4 + length : pos + 4 + length + 4] + calc_crc = self._check_crc(data_i, image_content[data_i][0], crc_data) + crcs[data_i] = (calc_crc, crc_data) if calc_crc else [] + pos += 4 + length + 1 + + txt_content: dict = {} + for text in attach_txt: + txt_content[text] = [] + pos = 0 + while (pos := data.find(text, pos)) != -1: + length = str2num(data[pos - 4 : pos]) + txt_content[text].append(data[pos + 4 : pos + 4 + length]) + pos += 1 + return txt_content, image_content, crcs + + def check_ihdr(self) -> bool: + """Check and repair IHDR chunk.""" + 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] + width, height = struct.unpack("!II", chunk_ihdr[:8]) + + 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("Bruteforcing dimensions...") + + for w in range(width + height + 1000): + for h in range(width + height + 1000): + 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: {w}x{h}") + fixed = True + break + if fixed: + break + + if not fixed: + self._error(f"Exhausted dimensions up to ({width + height + 1000})") + 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) + 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]}") + + 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_idat() + iend_fixed, extra_data = self.check_iend() + fixed |= iend_fixed + + return fixed, extra_data + + +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() diff --git a/aperisolve/config.py b/aperisolve/config.py index f4a6458..5bc34b0 100644 --- a/aperisolve/config.py +++ b/aperisolve/config.py @@ -6,6 +6,6 @@ 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" diff --git a/aperisolve/static/js/aperisolve.js b/aperisolve/static/js/aperisolve.js index 06ec2e8..7403807 100644 --- a/aperisolve/static/js/aperisolve.js +++ b/aperisolve/static/js/aperisolve.js @@ -5,6 +5,7 @@ const TOOL_ORDER = [ "foremost", "outguess", "pngcheck", + "pcrt", "steghide", "openstego", "zsteg", diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 87a499c..3d2fbfb 100644 --- a/aperisolve/workers.py +++ b/aperisolve/workers.py @@ -20,6 +20,7 @@ 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 @@ -106,6 +107,7 @@ def analyze_image(submission_hash: str) -> None: (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), diff --git a/docs/adding-analyzer.md b/docs/adding-analyzer.md index 224dd13..24069a9 100644 --- a/docs/adding-analyzer.md +++ b/docs/adding-analyzer.md @@ -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 | From 899e25cd7852cfc2c2669f21807a29622bff32a2 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Tue, 6 Jan 2026 04:49:34 +0100 Subject: [PATCH 02/13] typing --- aperisolve/analyzers/pcrt.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/aperisolve/analyzers/pcrt.py b/aperisolve/analyzers/pcrt.py index 9216225..1a4b6fe 100644 --- a/aperisolve/analyzers/pcrt.py +++ b/aperisolve/analyzers/pcrt.py @@ -1,5 +1,5 @@ # flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 +# pylint: disable=C0413,W0718,R0903,R0902,R0801 # mypy: disable-error-code=unused-awaitable """PCRT (PNG Check & Repair Tool) Analyzer for Image Submissions.""" @@ -43,9 +43,9 @@ class PNG: 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 = {} - self.image_content: dict = {} - self.crcs: dict = {} + 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] = [] @@ -114,7 +114,9 @@ class PNG: self.txt_content, self.image_content, self.crcs = self._find_ancillary(self.data) return True - def _find_ancillary(self, data: bytes) -> tuple[dict, dict, dict]: + def _find_ancillary( + self, data: bytes + ) -> tuple[dict[bytes, list[bytes]], dict[bytes, list[bytes]], dict[bytes, Any]]: """Find ancillary chunks in PNG data.""" ancillary = [ b"cHRM", @@ -139,8 +141,8 @@ class PNG: b"gIFx", ] attach_txt = [b"eXIf", b"iTXt", b"tEXt", b"zTXt"] - image_content: dict = {} - crcs: dict = {} + image_content: dict[bytes, list[bytes]] = {} + crcs: dict[bytes, Any] = {} for data_i in ancillary: image_content[data_i] = [] @@ -154,7 +156,7 @@ class PNG: crcs[data_i] = (calc_crc, crc_data) if calc_crc else [] pos += 4 + length + 1 - txt_content: dict = {} + txt_content: dict[bytes, list[bytes]] = {} for text in attach_txt: txt_content[text] = [] pos = 0 From e5a2d189c3c4d8473eedc31d0dadedcaec52f94d Mon Sep 17 00:00:00 2001 From: Zeecka Date: Tue, 6 Jan 2026 04:50:18 +0100 Subject: [PATCH 03/13] raw bytes format string --- aperisolve/analyzers/pcrt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aperisolve/analyzers/pcrt.py b/aperisolve/analyzers/pcrt.py index 1a4b6fe..858b6ee 100644 --- a/aperisolve/analyzers/pcrt.py +++ b/aperisolve/analyzers/pcrt.py @@ -296,7 +296,7 @@ class PNG: 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]}") + 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") From 6cd72d58998ca4979b791efd8d707ad744e54c4e Mon Sep 17 00:00:00 2001 From: Zeecka Date: Wed, 7 Jan 2026 04:08:09 +0100 Subject: [PATCH 04/13] Move image_resize into PCRT --- aperisolve/analyzers/image_resize.py | 164 ----------- aperisolve/analyzers/pcrt.py | 310 +-------------------- aperisolve/analyzers/utils.py | 264 ------------------ aperisolve/models.py | 117 ++++++-- aperisolve/static/js/aperisolve.js | 1 - aperisolve/utils/png.py | 393 +++++++++++++++++++++++++++ aperisolve/utils/utils.py | 110 ++++++++ aperisolve/workers.py | 2 - 8 files changed, 595 insertions(+), 766 deletions(-) delete mode 100644 aperisolve/analyzers/image_resize.py delete mode 100644 aperisolve/analyzers/utils.py create mode 100644 aperisolve/utils/png.py create mode 100644 aperisolve/utils/utils.py diff --git a/aperisolve/analyzers/image_resize.py b/aperisolve/analyzers/image_resize.py deleted file mode 100644 index b19447e..0000000 --- a/aperisolve/analyzers/image_resize.py +++ /dev/null @@ -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() diff --git a/aperisolve/analyzers/pcrt.py b/aperisolve/analyzers/pcrt.py index 858b6ee..a318bad 100644 --- a/aperisolve/analyzers/pcrt.py +++ b/aperisolve/analyzers/pcrt.py @@ -3,321 +3,13 @@ # mypy: disable-error-code=unused-awaitable """PCRT (PNG Check & Repair Tool) Analyzer for Image Submissions.""" -__author__ = [ - "Zeecka", - "indonumberone (https://github.com/indonumberone/PCRT3)", - "Etr1x (https://github.com/Etr1x/PCRT3)", - "sherlly (https://github.com/sherlly/PCRT)", -] - -import binascii -import itertools -import re -import struct -import zlib from pathlib import Path from typing import Any, Optional +from ..utils.png import PNG from .base_analyzer import SubprocessAnalyzer -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 str2num(s: bytes, n: int = 0) -> int: - """Convert bytes to number.""" - return struct.unpack("!I", s)[0] if n == 4 else int(str2hex(s), 16) - - -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"PNG", 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.""" - 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"] - image_content: dict[bytes, list[bytes]] = {} - crcs: dict[bytes, Any] = {} - - for data_i in ancillary: - image_content[data_i] = [] - crcs[data_i] = [] - pos = 0 - while (pos := data.find(data_i, pos)) != -1: - length = str2num(data[pos - 4 : pos]) - image_content[data_i].append(data[pos + 4 : pos + 4 + length]) - crc_data = data[pos + 4 + length : pos + 4 + length + 4] - calc_crc = self._check_crc(data_i, image_content[data_i][0], crc_data) - crcs[data_i] = (calc_crc, crc_data) if calc_crc else [] - pos += 4 + length + 1 - - txt_content: dict[bytes, list[bytes]] = {} - for text in attach_txt: - txt_content[text] = [] - pos = 0 - while (pos := data.find(text, pos)) != -1: - length = str2num(data[pos - 4 : pos]) - txt_content[text].append(data[pos + 4 : pos + 4 + length]) - pos += 1 - return txt_content, image_content, crcs - - def check_ihdr(self) -> bool: - """Check and repair IHDR chunk.""" - 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] - width, height = struct.unpack("!II", chunk_ihdr[:8]) - - 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("Bruteforcing dimensions...") - - for w in range(width + height + 1000): - for h in range(width + height + 1000): - 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: {w}x{h}") - fixed = True - break - if fixed: - break - - if not fixed: - self._error(f"Exhausted dimensions up to ({width + height + 1000})") - 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) - 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_idat() - iend_fixed, extra_data = self.check_iend() - fixed |= iend_fixed - - return fixed, extra_data - - class PCRTAnalyzer(SubprocessAnalyzer): """Analyzer for PCRT (PNG Check & Repair Tool).""" diff --git a/aperisolve/analyzers/utils.py b/aperisolve/analyzers/utils.py deleted file mode 100644 index a2ee0d0..0000000 --- a/aperisolve/analyzers/utils.py +++ /dev/null @@ -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})" - ) diff --git a/aperisolve/models.py b/aperisolve/models.py index 3f13c4d..307b605 100644 --- a/aperisolve/models.py +++ b/aperisolve/models.py @@ -4,7 +4,9 @@ """This module defines the database models for the Aperi'Solve application.""" import itertools +import struct import sys +import zlib from datetime import datetime, timezone from os import getenv from pathlib import Path @@ -12,9 +14,18 @@ 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.utils.utils import get_resolutions, get_valid_depth_color_pairs db: SQLAlchemy = SQLAlchemy() @@ -56,32 +67,64 @@ 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. - Returns: - None + 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. """ + # Skip if table already has entries if db.session.query(IHDR.iid).first() is not None: print("IHDR table already populated, skipping fill.") return - # Standard PNG IHDR parameters resolutions = get_resolutions() - bit_color_pairs = get_valid_depth_color_pairs() + bit_color_pairs = list(get_valid_depth_color_pairs()) interlace_methods = [0, 1] combinations = itertools.product( @@ -91,27 +134,49 @@ def fill_ihdr_db() -> None: ) 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) - 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) + # Check if entry already exists + exists = db.session.execute( + db.select(IHDR.iid).where( + db.and_( + IHDR.crc == crc, + IHDR.width == width, + IHDR.height == height, + IHDR.bit_depth == bit_depth, + IHDR.color_type == color_type, + IHDR.interlace == interlace, + ) + ) ).first() - if exists: # Skip if already present + if exists: continue - db.session.add(IHDR(packed=p.packed, crc=p.crc)) # pylint: disable=no-member - count += 1 + # 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} common IHDR entries...") + print(f"Inserted {count} IHDR entries...") db.session.commit() - db.session.commit() # pylint: disable=no-member - print(f"Precomputed {count} common IHDR entries. IHDR table filled successfully.") + # Final commit + if count % 5000 != 0: + db.session.commit() + print(f"Inserted {count} total IHDR entries.") def init_db(app: Flask) -> None: diff --git a/aperisolve/static/js/aperisolve.js b/aperisolve/static/js/aperisolve.js index 7403807..abaedd4 100644 --- a/aperisolve/static/js/aperisolve.js +++ b/aperisolve/static/js/aperisolve.js @@ -10,7 +10,6 @@ const TOOL_ORDER = [ "openstego", "zsteg", "strings", - "image_resize", ]; function escapeHtml(text) { diff --git a/aperisolve/utils/png.py b/aperisolve/utils/png.py new file mode 100644 index 0000000..9065419 --- /dev/null +++ b/aperisolve/utils/png.py @@ -0,0 +1,393 @@ +# flake8: noqa: E203,E501,W503 +# pylint: disable=C0413,W0718,R0903,R0801 +# 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 app +from ..models import IHDR +from .utils import int2hex, str2hex, str2num + +__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.""" + 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"] + image_content: dict[bytes, list[bytes]] = {} + crcs: dict[bytes, Any] = {} + + for data_i in ancillary: + image_content[data_i] = [] + crcs[data_i] = [] + pos = 0 + while (pos := data.find(data_i, pos)) != -1: + length = str2num(data[pos - 4 : pos]) + image_content[data_i].append(data[pos + 4 : pos + 4 + length]) + crc_data = data[pos + 4 + length : pos + 4 + length + 4] + calc_crc = self._check_crc(data_i, image_content[data_i][0], crc_data) + crcs[data_i] = (calc_crc, crc_data) if calc_crc else [] + pos += 4 + length + 1 + + txt_content: dict[bytes, list[bytes]] = {} + for text in attach_txt: + txt_content[text] = [] + pos = 0 + while (pos := data.find(text, pos)) != -1: + length = str2num(data[pos - 4 : pos]) + txt_content[text].append(data[pos + 4 : pos + 4 + length]) + pos += 1 + return txt_content, image_content, crcs + + def check_chunks(self) -> bool: + """Copy ancillary chunks (like PLTE) from original data to repaired_data.""" + # Critical ancillary chunks that should be preserved + critical_ancillary = [ + b"PLTE", # Palette (required for color_type 3) + b"tRNS", # Transparency + b"cHRM", # Chromaticity + b"gAMA", # Gamma + b"iCCP", # ICC Profile + b"sBIT", # Significant bits + b"sRGB", # sRGB color space + b"bKGD", # Background color + b"hIST", # Histogram + b"pHYs", # Physical pixel dimensions + b"sPLT", # Suggested palette + ] + + # Find position after IHDR in repaired data + ihdr_end = len(self.repaired_data) + + # Find IDAT position in original data to know where to stop looking + idat_pos = self.data.find(b"IDAT") + if idat_pos == -1: + return False + + # Search for ancillary chunks between IHDR and IDAT + 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 # +4 for length, +4 for CRC + + # 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 + + # Extract the complete chunk (length + type + data + CRC) + chunk_start = pos - 4 # Include length field + length = struct.unpack("!I", self.data[chunk_start:pos])[0] + chunk_end = pos + 4 + length + 4 # type + data + CRC + + chunk_data = self.data[chunk_start:chunk_end] + self.repaired_data[ihdr_end:ihdr_end] = chunk_data + ihdr_end += len(chunk_data) + + self._log(f"Copied {chunk_type.decode()} chunk ({length} bytes)") + + pos = chunk_end + return True + + def check_ihdr(self) -> bool: + """Check and repair IHDR chunk using database lookup.""" + 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] + # width, height = struct.unpack("!II", chunk_ihdr[:8]) + + 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...") + + # Convert CRC bytes to integer for database lookup + crc_int = struct.unpack("!I", crc)[0] + + # Database lookup + 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") + + # Use the first match + match = matches[0] + + # Build the corrected IHDR chunk data directly from database fields + test_ihdr = match.to_ihdr_bytes() + + # Verify the CRC matches + 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("Exhausted all recovery attempts") + 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 diff --git a/aperisolve/utils/utils.py b/aperisolve/utils/utils.py new file mode 100644 index 0000000..43e8b65 --- /dev/null +++ b/aperisolve/utils/utils.py @@ -0,0 +1,110 @@ +# 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 +import struct +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 str2num(s: bytes, n: int = 0) -> int: + """Convert bytes to number.""" + return struct.unpack("!I", s)[0] if n == 4 else int(str2hex(s), 16) + + +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 diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 3d2fbfb..71b06f5 100644 --- a/aperisolve/workers.py +++ b/aperisolve/workers.py @@ -17,7 +17,6 @@ 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 @@ -104,7 +103,6 @@ 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), From c3543cca73a410b10fd4c9b97c32c7d66cbb4c9f Mon Sep 17 00:00:00 2001 From: Zeecka Date: Wed, 7 Jan 2026 04:23:57 +0100 Subject: [PATCH 05/13] Fix multiples CRC and optimize --- aperisolve/models.py | 36 ++++++----- aperisolve/utils/__init__.py | 0 aperisolve/utils/png.py | 118 +++++++++++++++++++---------------- 3 files changed, 83 insertions(+), 71 deletions(-) create mode 100644 aperisolve/utils/__init__.py diff --git a/aperisolve/models.py b/aperisolve/models.py index 307b605..115a74c 100644 --- a/aperisolve/models.py +++ b/aperisolve/models.py @@ -132,7 +132,8 @@ def fill_ihdr_db() -> None: bit_color_pairs, interlace_methods, ) - + batch = [] + batch_size = 10000 # Increased from 5000 count = 0 with db.session.no_autoflush: for (width, height), (bit_depth, color_type), interlace in combinations: @@ -156,27 +157,30 @@ def fill_ihdr_db() -> None: if exists: continue - # Add new entry - db.session.add( - IHDR( - crc=crc, - width=width, - height=height, - bit_depth=bit_depth, - color_type=color_type, - interlace=interlace, - ) + batch.append( + { + "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...") + if len(batch) >= batch_size: + db.session.bulk_insert_mappings(IHDR, batch) db.session.commit() + print(f"Inserted {count} IHDR entries...") + batch = [] - # Final commit - if count % 5000 != 0: + # Insert remaining entries + if batch: + db.session.bulk_insert_mappings(IHDR, batch) db.session.commit() - print(f"Inserted {count} total IHDR entries.") + + print(f"Inserted {count} total IHDR entries.") def init_db(app: Flask) -> None: diff --git a/aperisolve/utils/__init__.py b/aperisolve/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/aperisolve/utils/png.py b/aperisolve/utils/png.py index 9065419..1163a73 100644 --- a/aperisolve/utils/png.py +++ b/aperisolve/utils/png.py @@ -20,6 +20,7 @@ __author__ = [ "sherlly (https://github.com/sherlly/PCRT)", ] + class PNG: """PNG file analyzer and repair tool.""" @@ -101,7 +102,7 @@ class PNG: def _find_ancillary( self, data: bytes ) -> tuple[dict[bytes, list[bytes]], dict[bytes, list[bytes]], dict[bytes, Any]]: - """Find ancillary chunks in PNG data.""" + """Find ancillary chunks in PNG data with an optimized single pass.""" ancillary = [ b"cHRM", b"pHYs", @@ -125,64 +126,72 @@ class PNG: b"gIFx", ] attach_txt = [b"eXIf", b"iTXt", b"tEXt", b"zTXt"] - image_content: dict[bytes, list[bytes]] = {} - crcs: dict[bytes, Any] = {} - for data_i in ancillary: - image_content[data_i] = [] - crcs[data_i] = [] - pos = 0 - while (pos := data.find(data_i, pos)) != -1: - length = str2num(data[pos - 4 : pos]) - image_content[data_i].append(data[pos + 4 : pos + 4 + length]) - crc_data = data[pos + 4 + length : pos + 4 + length + 4] - calc_crc = self._check_crc(data_i, image_content[data_i][0], crc_data) - crcs[data_i] = (calc_crc, crc_data) if calc_crc else [] - pos += 4 + length + 1 + # 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} - txt_content: dict[bytes, list[bytes]] = {} - for text in attach_txt: - txt_content[text] = [] - pos = 0 - while (pos := data.find(text, pos)) != -1: - length = str2num(data[pos - 4 : pos]) - txt_content[text].append(data[pos + 4 : pos + 4 + length]) + # 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.""" - # Critical ancillary chunks that should be preserved + """Copy ancillary chunks (like PLTE) from original data to repaired_data with crc + validation.""" critical_ancillary = [ - b"PLTE", # Palette (required for color_type 3) - b"tRNS", # Transparency - b"cHRM", # Chromaticity - b"gAMA", # Gamma - b"iCCP", # ICC Profile - b"sBIT", # Significant bits - b"sRGB", # sRGB color space - b"bKGD", # Background color - b"hIST", # Histogram - b"pHYs", # Physical pixel dimensions - b"sPLT", # Suggested palette + b"PLTE", + b"tRNS", + b"cHRM", + b"gAMA", + b"iCCP", + b"sBIT", + b"sRGB", + b"bKGD", + b"hIST", + b"pHYs", + b"sPLT", ] - # Find position after IHDR in repaired data ihdr_end = len(self.repaired_data) - - # Find IDAT position in original data to know where to stop looking idat_pos = self.data.find(b"IDAT") if idat_pos == -1: return False - # Search for ancillary chunks between IHDR and IDAT 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 # +4 for length, +4 for CRC + search_start = search_start + 4 + ihdr_length + 4 # Collect all ancillary chunks before IDAT for chunk_type in critical_ancillary: @@ -192,22 +201,28 @@ class PNG: if pos == -1: break - # Extract the complete chunk (length + type + data + CRC) - chunk_start = pos - 4 # Include length field + chunk_start = pos - 4 length = struct.unpack("!I", self.data[chunk_start:pos])[0] - chunk_end = pos + 4 + length + 4 # type + data + CRC + chunk_data = self.data[pos + 4 : pos + 4 + length] + chunk_crc = self.data[pos + 4 + length : pos + 8 + length] - chunk_data = self.data[chunk_start:chunk_end] - self.repaired_data[ihdr_end:ihdr_end] = chunk_data - ihdr_end += len(chunk_data) + # 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 - pos = chunk_end return True def check_ihdr(self) -> bool: - """Check and repair IHDR chunk using database lookup.""" + """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") @@ -216,7 +231,6 @@ class PNG: length = struct.unpack("!I", ihdr[:4])[0] chunk_type, chunk_ihdr = ihdr[4:8], ihdr[8 : 8 + length] crc = ihdr[8 + length : 12 + length] - # width, height = struct.unpack("!II", chunk_ihdr[:8]) fixed = False if calc_crc := self._check_crc(chunk_type, chunk_ihdr, crc): @@ -224,7 +238,6 @@ class PNG: self._log(f"Chunk crc: {str2hex(crc)}, Correct crc: {str2hex(calc_crc)}") self._log("Looking up CRC in database...") - # Convert CRC bytes to integer for database lookup crc_int = struct.unpack("!I", crc)[0] # Database lookup @@ -233,14 +246,9 @@ class PNG: if matches: self._log(f"Found {len(matches)} matching IHDR configuration(s) in database") - - # Use the first match match = matches[0] - - # Build the corrected IHDR chunk data directly from database fields test_ihdr = match.to_ihdr_bytes() - # Verify the CRC matches if not self._check_crc(chunk_type, test_ihdr, crc): ihdr = ihdr[:8] + test_ihdr + crc self._log( @@ -268,7 +276,7 @@ class PNG: break if not fixed: - self._error("Exhausted all recovery attempts") + self._error("Could not recover IHDR dimensions") else: self._log(f"Correct IHDR CRC at offset {int2hex(pos + 4 + length)}") From 5e39d88ad68ba545e6c43ab88eff09d5dcc56572 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Wed, 7 Jan 2026 04:25:33 +0100 Subject: [PATCH 06/13] pylint --- aperisolve/utils/png.py | 4 ++-- aperisolve/utils/utils.py | 5 ----- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/aperisolve/utils/png.py b/aperisolve/utils/png.py index 1163a73..2ae2c74 100644 --- a/aperisolve/utils/png.py +++ b/aperisolve/utils/png.py @@ -1,5 +1,5 @@ # flake8: noqa: E203,E501,W503 -# pylint: disable=C0413,W0718,R0903,R0801 +# pylint: disable=C0413,W0718,R0903,R0801,R0902 # mypy: disable-error-code=unused-awaitable """PNG Class for analyzer modules""" @@ -11,7 +11,7 @@ from typing import Any, Optional from ..app import app from ..models import IHDR -from .utils import int2hex, str2hex, str2num +from .utils import int2hex, str2hex __author__ = [ "Zeecka", diff --git a/aperisolve/utils/utils.py b/aperisolve/utils/utils.py index 43e8b65..3dbc6d5 100644 --- a/aperisolve/utils/utils.py +++ b/aperisolve/utils/utils.py @@ -19,11 +19,6 @@ def int2hex(i: int) -> str: return "0x" + hex(i)[2:].upper() -def str2num(s: bytes, n: int = 0) -> int: - """Convert bytes to number.""" - return struct.unpack("!I", s)[0] if n == 4 else int(str2hex(s), 16) - - def get_resolutions() -> list[tuple[int, int]]: """ Generate a sorted list of common display and image resolutions. From 4b6a944817afe7e46032459f67abb87b6544d832 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Wed, 7 Jan 2026 04:27:55 +0100 Subject: [PATCH 07/13] pylint --- aperisolve/utils/utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/aperisolve/utils/utils.py b/aperisolve/utils/utils.py index 3dbc6d5..0cf7845 100644 --- a/aperisolve/utils/utils.py +++ b/aperisolve/utils/utils.py @@ -5,7 +5,6 @@ import binascii import math -import struct from typing import Iterable From 5719ee7c3ebd2115eedfede752cb49e263150cc7 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 05:34:24 +0100 Subject: [PATCH 08/13] Huge refactoring --- .env.example | 3 +- Dockerfile | 2 +- aperisolve/__init__.py | 3 + aperisolve/analyzers/base_analyzer.py | 3 +- aperisolve/app.py | 421 +++++++++++++------------- aperisolve/cleanup.py | 79 ----- aperisolve/config.py | 4 + aperisolve/models.py | 187 ++++++------ aperisolve/utils/init_db.py | 62 ++++ aperisolve/utils/png.py | 3 +- aperisolve/{ => utils}/sentry.py | 0 aperisolve/{ => utils}/wsgi.py | 5 +- aperisolve/workers.py | 7 +- compose.dev.yml | 31 +- compose.yml | 37 ++- 15 files changed, 424 insertions(+), 423 deletions(-) delete mode 100644 aperisolve/cleanup.py create mode 100644 aperisolve/utils/init_db.py rename aperisolve/{ => utils}/sentry.py (100%) rename aperisolve/{ => utils}/wsgi.py (63%) diff --git a/.env.example b/.env.example index 150aee9..f12363f 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/Dockerfile b/Dockerfile index 4055aa6..522d13a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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"] diff --git a/aperisolve/__init__.py b/aperisolve/__init__.py index 538f764..6659e63 100644 --- a/aperisolve/__init__.py +++ b/aperisolve/__init__.py @@ -1,3 +1,6 @@ # flake8: noqa: E203,E501,W503 # pylint: disable=C0413,W0718,R0903,R0801 # mypy: disable-error-code=unused-awaitable +from . import app, config, models, workers + +__all__ = ["app", "config", "models", "workers"] diff --git a/aperisolve/analyzers/base_analyzer.py b/aperisolve/analyzers/base_analyzer.py index 42edcb7..6597f99 100644 --- a/aperisolve/analyzers/base_analyzer.py +++ b/aperisolve/analyzers/base_analyzer.py @@ -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): diff --git a/aperisolve/app.py b/aperisolve/app.py index ae869ea..0927ce0 100644 --- a/aperisolve/app.py +++ b/aperisolve/app.py @@ -16,270 +16,255 @@ 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("/", 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("/", 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/", 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/", 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/", 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//") + 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/", 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/", 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/", 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//") -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/") -@app.route("/image//") -def get_image(hash_val: Optional[str] = None, img_name: Optional[str] = None) -> Response: - """ - Download the image with a specific name (usually ones generated with - decomposer) for a given submission hash. - """ - - if img_name is None: - return abort(404, description="Image not found or unsupported format") - - if hash_val is not None: submission = Submission.query.filter_by(hash=hash_val).first_or_404() image = Image.query.get_or_404(submission.image_hash) # type: ignore output_dir = RESULT_FOLDER / str(image.hash) / str(submission.hash) - output_file = output_dir / Path(img_name).name - else: - hash_val = img_name.split(".")[0] # Extract hash from the image name - image = Image.query.filter_by(hash=hash_val).first_or_404() - output_file = Path(image.file) # type: ignore + 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/") + @app.route("/image//") + def get_image(hash_val: Optional[str] = None, img_name: Optional[str] = None) -> Response: + """ + Download the image with a specific name (usually ones generated with + decomposer) for a given submission hash. + """ + + if img_name is None: + return abort(404, description="Image not found or unsupported format") + + if hash_val is not None: + submission = Submission.query.filter_by(hash=hash_val).first_or_404() + image = Image.query.get_or_404(submission.image_hash) # type: ignore + + output_dir = RESULT_FOLDER / str(image.hash) / str(submission.hash) + output_file = output_dir / Path(img_name).name + else: + hash_val = img_name.split(".")[0] # Extract hash from the image name + image = Image.query.filter_by(hash=hash_val).first_or_404() + output_file = Path(image.file) # type: ignore + + if (not output_file.exists()) or (output_file.suffix.lower() not in IMAGE_EXTENSIONS): + return abort(404, description="Image not found or unsupported format") + + return send_file(output_file, as_attachment=True) + + 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) diff --git a/aperisolve/cleanup.py b/aperisolve/cleanup.py deleted file mode 100644 index 032a7b6..0000000 --- a/aperisolve/cleanup.py +++ /dev/null @@ -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 diff --git a/aperisolve/config.py b/aperisolve/config.py index 5bc34b0..40d4c94 100644 --- a/aperisolve/config.py +++ b/aperisolve/config.py @@ -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", "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")) diff --git a/aperisolve/models.py b/aperisolve/models.py index 115a74c..66b2330 100644 --- a/aperisolve/models.py +++ b/aperisolve/models.py @@ -1,18 +1,15 @@ # 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 shutil import struct -import sys +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, @@ -25,6 +22,7 @@ from sqlalchemy import ( String, ) +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() @@ -117,110 +115,109 @@ def fill_ihdr_db() -> None: 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 - # Skip if table already has entries - if db.session.query(IHDR.iid).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] - 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 - combinations = itertools.product( - resolutions, - bit_color_pairs, - interlace_methods, - ) - batch = [] - batch_size = 10000 # Increased from 5000 - 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) + 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) - # Check if entry already exists - exists = db.session.execute( - db.select(IHDR.iid).where( - db.and_( - IHDR.crc == crc, - IHDR.width == width, - IHDR.height == height, - IHDR.bit_depth == bit_depth, - IHDR.color_type == color_type, - IHDR.interlace == interlace, + # Add new entry + db.session.add( + IHDR( + crc=crc, + width=width, + height=height, + bit_depth=bit_depth, + color_type=color_type, + interlace=interlace, ) ) - ).first() - if exists: - continue + count += 1 + if count % 5000 == 0: + print(f"Inserted {count} IHDR entries...") + db.session.commit() + print(f"Inserted {count} total IHDR entries.") - batch.append( - { - "crc": crc, - "width": width, - "height": height, - "bit_depth": bit_depth, - "color_type": color_type, - "interlace": interlace, - } - ) + except Exception as e: + print(f"Error filling IHDR table: {e}") + db.session.rollback() - count += 1 - if len(batch) >= batch_size: - db.session.bulk_insert_mappings(IHDR, batch) - db.session.commit() - print(f"Inserted {count} IHDR entries...") - batch = [] - - # Insert remaining entries - if batch: - db.session.bulk_insert_mappings(IHDR, batch) - db.session.commit() - - print(f"Inserted {count} total IHDR entries.") - - -def init_db(app: Flask) -> None: +def cleanup_old_entries() -> None: """ - Initialize the database by creating tables and populating lookup data. + Clean up old and incomplete entries from the database and file system. - This function creates the database schema if it doesn't already exist, - and populates the CRC lookup table for PNG IHDR validation. + This function performs two main cleanup operations: - Args: - app (Flask): The Flask application instance used to establish the application context - for database 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 - - 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]) + 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) - 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.") + # 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: - fill_ihdr_db() + 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 diff --git a/aperisolve/utils/init_db.py b/aperisolve/utils/init_db.py new file mode 100644 index 0000000..abe89a3 --- /dev/null +++ b/aperisolve/utils/init_db.py @@ -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() diff --git a/aperisolve/utils/png.py b/aperisolve/utils/png.py index 2ae2c74..ff319c7 100644 --- a/aperisolve/utils/png.py +++ b/aperisolve/utils/png.py @@ -9,7 +9,7 @@ import struct import zlib from typing import Any, Optional -from ..app import app +from ..app import create_app from ..models import IHDR from .utils import int2hex, str2hex @@ -241,6 +241,7 @@ class PNG: 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() diff --git a/aperisolve/sentry.py b/aperisolve/utils/sentry.py similarity index 100% rename from aperisolve/sentry.py rename to aperisolve/utils/sentry.py diff --git a/aperisolve/wsgi.py b/aperisolve/utils/wsgi.py similarity index 63% rename from aperisolve/wsgi.py rename to aperisolve/utils/wsgi.py index a220be7..3a90300 100644 --- a/aperisolve/wsgi.py +++ b/aperisolve/utils/wsgi.py @@ -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() diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 71b06f5..cb6e467 100644 --- a/aperisolve/workers.py +++ b/aperisolve/workers.py @@ -9,7 +9,7 @@ from typing import Any import sentry_sdk -from .sentry import initialize_sentry +from .utils.sentry import initialize_sentry initialize_sentry() @@ -24,9 +24,9 @@ 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 diff --git a/compose.dev.yml b/compose.dev.yml index 4d251b5..f095370 100644 --- a/compose.dev.yml +++ b/compose.dev.yml @@ -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 diff --git a/compose.yml b/compose.yml index f4d1d1f..b388c51 100644 --- a/compose.yml +++ b/compose.yml @@ -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 From 3d03f50547da28a62b49f03b2922498c0d7f9348 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 18:01:40 +0100 Subject: [PATCH 09/13] Update pre-commit file --- .pre-commit-config.yaml | 34 ++++++++++++++++++++-------------- 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b8fa7af..eda0da9 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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", + ] From 9cbdf8c4431b6cd33fa3293eff72d9b5b7ddc401 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 18:02:25 +0100 Subject: [PATCH 10/13] Update dev requirements --- .gitignore | 1 + CONTRIBUTING.md | 62 +++++++++++++++++++++------------ aperisolve/requirements-dev.txt | 6 ++++ 3 files changed, 46 insertions(+), 23 deletions(-) create mode 100644 aperisolve/requirements-dev.txt diff --git a/.gitignore b/.gitignore index 7b0fdd9..61c8c3e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Python .env +.venv/ __pycache__/ .mypy_cache/ results/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c8346db..029a156 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -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 diff --git a/aperisolve/requirements-dev.txt b/aperisolve/requirements-dev.txt new file mode 100644 index 0000000..4cfd75c --- /dev/null +++ b/aperisolve/requirements-dev.txt @@ -0,0 +1,6 @@ +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 From 26dbab3fe671c20e6b6d4b115296fdb40bba9111 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 18:05:46 +0100 Subject: [PATCH 11/13] Linting --- aperisolve/__init__.py | 2 ++ aperisolve/app.py | 3 ++- aperisolve/models.py | 1 + 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/aperisolve/__init__.py b/aperisolve/__init__.py index 6659e63..2d22b7f 100644 --- a/aperisolve/__init__.py +++ b/aperisolve/__init__.py @@ -1,3 +1,5 @@ +"""Main Aperi'Solve module""" + # flake8: noqa: E203,E501,W503 # pylint: disable=C0413,W0718,R0903,R0801 # mypy: disable-error-code=unused-awaitable diff --git a/aperisolve/app.py b/aperisolve/app.py index 0927ce0..db90c1f 100644 --- a/aperisolve/app.py +++ b/aperisolve/app.py @@ -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.""" @@ -264,6 +264,7 @@ def create_app() -> Flask: return app + if __name__ == "__main__": my_app = create_app() debug_mode = os.environ.get("FLASK_DEBUG", "0") == "1" diff --git a/aperisolve/models.py b/aperisolve/models.py index 66b2330..4af9cb3 100644 --- a/aperisolve/models.py +++ b/aperisolve/models.py @@ -160,6 +160,7 @@ def fill_ihdr_db() -> None: 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. From c9e74424deea0565a2a1915ec9242a454f894cc0 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 18:31:47 +0100 Subject: [PATCH 12/13] Fix linter in ci --- .github/workflows/lint.yml | 9 ++------- aperisolve/requirements-dev.txt | 1 + 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index e0c4365..1156ce5 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -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 diff --git a/aperisolve/requirements-dev.txt b/aperisolve/requirements-dev.txt index 4cfd75c..182cc37 100644 --- a/aperisolve/requirements-dev.txt +++ b/aperisolve/requirements-dev.txt @@ -4,3 +4,4 @@ flake8==7.3.0 mypy==1.18.1 pylint==4.0.4 pre-commit==4.5.1 +mypy-extensions From 1e4f12174c649941cfe89492bb61c8934ae19961 Mon Sep 17 00:00:00 2001 From: Zeecka Date: Thu, 8 Jan 2026 18:42:34 +0100 Subject: [PATCH 13/13] Fix image preview for PCRT --- aperisolve/static/css/aperisolve.css | 225 ++++++++++++++++----------- aperisolve/static/js/aperisolve.js | 46 +++--- 2 files changed, 157 insertions(+), 114 deletions(-) diff --git a/aperisolve/static/css/aperisolve.css b/aperisolve/static/css/aperisolve.css index 732a323..a27cb42 100644 --- a/aperisolve/static/css/aperisolve.css +++ b/aperisolve/static/css/aperisolve.css @@ -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; } @@ -455,7 +467,7 @@ footer a i { .results_img { width: 33.333%; } - + .browse-images { width: 33.333%; } @@ -467,29 +479,30 @@ footer a i { padding-left: 15px; padding-right: 15px; } - + .results_img { width: 50%; } - + .browse-images { width: 50%; } - + #result-infos { flex-direction: column; } - - #result-infos > div { + + #result-infos>div { width: 100%; flex: 1 1 100%; } - + table { font-size: 0.9rem; } - - th, td { + + th, + td { padding: 0.5rem; } } @@ -500,110 +513,116 @@ footer a i { body { font-size: 14px; } - + .container { margin-top: 20px !important; padding-left: 10px; padding-right: 10px; } - + h1 { font-size: 1.5rem; } - + h2 { font-size: 1.25rem; } - + h3 { font-size: 1.1rem; } - + #water-logo { width: 50px; } - - .card, .form-control, .table, .border, .card-hack { + + .card, + .form-control, + .table, + .border, + .card-hack { border-radius: 0.35rem; } - + .card { padding: 15px !important; } - + #drop-zone { padding: 20px !important; } - + #drop-zone p { font-size: 0.9rem; } - + .results_img { width: 50%; padding: 5px; } - + .browse-images { width: 50%; padding: 5px; } - - textarea.form-control, pre { + + textarea.form-control, + pre { padding: 15px; font-size: 0.85rem; } - + .a-strings textarea { height: 250px; } - + .a-exiftool .table-container { max-height: 300px; } - + /* Make tables scrollable horizontally */ .table-container { overflow-x: auto; -webkit-overflow-scrolling: touch; } - + table { font-size: 0.8rem; min-width: 400px; } - - th, td { + + th, + td { padding: 0.4rem; word-break: break-word; } - + footer { justify-content: center !important; margin: 20px 0; flex-wrap: wrap; gap: 15px; } - + footer a i { font-size: 28px; } - + #coffee { height: 28px; } - + /* FAQ/Cheatsheet adjustments */ - #cheatsheet > div { + #cheatsheet>div { padding: 8px; margin-bottom: 30px; } - + #cheatsheet img { max-width: 100%; height: auto; } - + .command { padding: 8px; font-size: 0.85rem; @@ -617,112 +636,116 @@ footer a i { body { font-size: 13px; } - + .container { margin-top: 15px !important; padding-left: 8px; padding-right: 8px; } - + h1 { font-size: 1.3rem; } - + h2 { font-size: 1.1rem; } - + h3 { font-size: 1rem; } - + #water-logo { width: 40px; } - + .card { padding: 12px !important; } - + #drop-zone { padding: 15px !important; } - + #drop-zone p { font-size: 0.85rem; } - - .btn, .btn-primary, .btn-hack { + + .btn, + .btn-primary, + .btn-hack { width: 100%; padding: 12px; font-size: 1rem; } - + .results_img { width: 100%; padding: 5px; } - + .browse-images { width: 100%; padding: 5px; } - - #result-infos > div { + + #result-infos>div { padding: 5px; } - - textarea.form-control, pre { + + textarea.form-control, + pre { padding: 10px; font-size: 0.8rem; } - + .a-strings textarea { height: 200px; } - + .a-exiftool .table-container { max-height: 250px; } - + table { font-size: 0.75rem; } - - th, td { + + th, + td { padding: 0.3rem; } - + .copy-icon { display: block; right: 5px; margin-top: -35px; } - + footer { gap: 12px; } - + footer a i { font-size: 24px; } - + #coffee { height: 24px; } - + #image-modal-container img { width: 95%; } - + .analyzer { margin-top: 15px; } - - .analyzer h2 + div { + + .analyzer h2+div { margin-top: 10px; } - + .analyzer h3 { margin-top: 15px; } @@ -732,12 +755,12 @@ footer a i { padding-left: 20px; font-size: 0.9rem; } - + .command { font-size: 0.75rem; padding: 6px; } - + #httperror { padding-top: 50px; } @@ -748,24 +771,26 @@ footer a i { body { font-size: 12px; } - + h1 { font-size: 1.1rem; } - + #water-logo { width: 35px; } - - .btn, .btn-primary, .btn-hack { + + .btn, + .btn-primary, + .btn-hack { padding: 10px; font-size: 0.9rem; } - + footer a i { font-size: 22px; } - + #coffee { height: 22px; } @@ -778,16 +803,22 @@ footer a i { display: block; 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; } @@ -798,7 +829,7 @@ footer a i { .container { margin-top: 10px !important; } - + #image-modal-container img { max-height: 85vh; width: auto; @@ -811,24 +842,36 @@ footer a i { #three-bg { display: none; } - + body { background-color: white; 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; } } diff --git a/aperisolve/static/js/aperisolve.js b/aperisolve/static/js/aperisolve.js index abaedd4..26eb3a4 100644 --- a/aperisolve/static/js/aperisolve.js +++ b/aperisolve/static/js/aperisolve.js @@ -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 += `

${capitalize(tool.replace("_"," "))}

`; + analyzer.innerHTML += `

${capitalize(tool.replace("_", " "))}

`; // 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 += `
`; + analyzer.innerHTML += `
`; } } @@ -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%";