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

Move image_resize into PCRT

Browse Source
This commit is contained in:
Zeecka
2026-01-07 04:08:09 +01:00
parent e5a2d189c3
commit 6cd72d5899
8 changed files with 595 additions and 766 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aperisolve/analyzers/image_resize.py
-164
View File
@@ -1,164 +0,0 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Image size bruteforcer for common PNG sizes."""
from pathlib import Path
from typing import Any, List, Optional
from ..app import app
from ..models import IHDR
from .base_analyzer import SubprocessAnalyzer
from .utils import PNG
class ResizeAnalyzer(SubprocessAnalyzer):
"""Analyzer for image resize."""
png: PNG
def __init__(self, input_img: Path, output_dir: Path) -> None:
super().__init__("image_resize", input_img, output_dir, has_archive=True)
def _lookup_crc(self) -> List[PNG]:
"""
Look up PNG images in the database by their CRC value.
Returns:
List[PNG]: A list of PNG objects with matching CRC values, reconstructed from their
packed representations.
"""
with app.app_context():
imgs = IHDR.query.filter_by(crc=self.png.crc).all()
results = []
for img in imgs:
packed = img.packed
results.append(PNG.from_packed(packed, crc=self.png.crc))
return results
def _search_height_crc(self) -> List[PNG]:
"""
Search for PNG images with matching CRC by iterating through possible heights.
This function attempts to find alternative PNG configurations that produce the same
CRC checksum as the input PNG by systematically varying the height parameter while
keeping all other properties constant.
Returns:
List[PNG]: A list of PNG objects with matching CRC values. Returns an empty list
if no matches are found.
Note:
The function iterates through heights from 100 to 3500 (inclusive).
This may be computationally expensive for large ranges.
"""
results = []
for height in range(100, 3501):
candidate = PNG(
size=(self.png.width, height),
bit_depth=self.png.bit_depth,
color_type=self.png.color_type,
interlace=self.png.interlace,
)
if candidate.crc == self.png.crc:
results.append(candidate)
return results
def _write_recovered_png(self, png_bytes: bytearray, png_match: PNG) -> str:
"""
Rebuilds a PNG by patching IHDR width/height and writes it to disk.
"""
img_name = f"recovered_{png_match.width}x{png_match.height}.png"
output_path = self.output_dir / img_name
height_bytes = png_match.height.to_bytes(4, byteorder="big")
width_bytes = png_match.width.to_bytes(4, byteorder="big")
full_png_data = (
png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End]
+ width_bytes
+ height_bytes
+ png_bytes[24:]
)
with output_path.open("wb") as out_f:
out_f.write(full_png_data)
return img_name
def get_results(self, _: Optional[str] = None) -> dict[str, Any]:
"""
Analyze a PNG image and recover resized versions based on CRC matching.
This function attempts to find matching image dimensions for a given PNG file
by looking up its CRC (Cyclic Redundancy Check) in a database. If matches are
found, it recovers and saves PNG images with the matched dimensions.
Returns:
dict[str, Any]: A dictionary containing the analysis status, any error message,
Raises:
Handles exceptions gracefully by updating output data with error status.
Side Effects:
- Creates output directory if it doesn't exist.
- Saves recovered PNG images to the output directory.
- Updates data in the output directory with analysis results via update_data().
Notes:
- If the PNG has invalid structure or IHDR chunk is not first, an error is logged.
- The function performs a two-step lookup: first by CRC, then by height/CRC if no
matches found.
- Recovered images are named as "recovered_{width}x{height}.png".
- Analysis results (logs, status, and image URLs) are persisted via update_data().
"""
logs = []
with self.input_img.open("rb") as image:
img_bytes = bytearray(image.read())
try:
self.png = PNG.from_bytearray(img_bytes)
except ValueError:
return {
"status": "error",
"error": "IHDR chunk is not the first chunk, or PNG has invalid structure.",
}
# Already correct
if self.png.crc == self.png.compute_crc():
return {
"status": "ok",
"note": f"PNG is already valid with dimensions "
f"{self.png.width}x{self.png.height} and crc 0x{self.png.crc:08x}.",
}
saved_img_urls = []
# No match found
if not (db_matches := self._lookup_crc()) and not (db_matches := self._search_height_crc()):
return {
"status": "error",
"error": "Failure: No matching dimensions found.",
}
# Match found, saving image(s)
logs.append(f"Target CRC found: 0x{self.png.crc:08x}")
self.output_dir.mkdir(parents=True, exist_ok=True)
for png_match in db_matches:
img_name = self._write_recovered_png(img_bytes, png_match)
logs.append(f"Image saved: {img_name}")
# Append the formatted URL to our list
saved_img_urls.append("/image/" + str(Path(self.output_dir.name) / img_name))
return {
"status": "ok",
"output": logs,
"png_images": saved_img_urls,
}
def analyze_image_resize(input_img: Path, output_dir: Path) -> None:
"""Analyze an image submission using ResizeAnalyzer."""
analyzer = ResizeAnalyzer(input_img, output_dir)
analyzer.analyze()
aperisolve/analyzers/pcrt.py
+1 -309
View File
@@ -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)."""
aperisolve/analyzers/utils.py
-264
View File
@@ -1,264 +0,0 @@
# flake8: noqa: E203,E501,W503
# pylint: disable=C0413,W0718,R0903,R0801
# mypy: disable-error-code=unused-awaitable
"""Utility functions for analyzers modules."""
import math
import struct
import zlib
from functools import cached_property
from typing import Iterable, Self
def get_resolutions() -> list[tuple[int, int]]:
"""
Generate a sorted list of common display and image resolutions.
Creates a comprehensive set of resolution tuples by combining various base widths
with standard aspect ratios used in screens, digital displays, photography, and
paper standards. Includes both portrait and landscape orientations.
Base widths range from 16px to 10000px, covering:
- Low resolution: 16-256px (16px increments)
- Standard: 320-1024px (32px increments)
- HD/Full HD: 1280-2560px (64px increments)
- 4K: 3000-4096px (128px increments)
- 5K-8K: 5120-8192px (256px increments)
- Upper bound: 10000px
Aspect ratios include common formats:
- Digital screens: 1:1, 4:3, 3:2, 16:10, 16:9, 21:9
- Photography/print: 5:4, 7:5, 2:3
- Paper standards: ISO A-series, US Letter, Legal, Tabloid
Returns:
list[tuple[int, int]]: Sorted list of (width, height) resolution tuples
with heights constrained between 1 and 10000 pixels.
"""
base_widths = (
list(range(16, 257, 16)) # 16 → 256
+ list(range(320, 1025, 32)) # 320 → 1024
+ list(range(1280, 2561, 64)) # 1280 → 2560
+ list(range(3000, 4097, 128)) # 3K–4K
+ list(range(5120, 8193, 256)) # 5K–8K
+ [10000] # upper bound
)
aspect_ratios = [
# Screens / digital
(1, 1), # square
(4, 3),
(3, 2),
(16, 10),
(16, 9),
(21, 9),
# Photography / print
(5, 4), # 8x10
(7, 5), # 5x7
(3, 2), # 4x6 (already included, kept for clarity)
(2, 3), # poster
# Paper standards
(1414, 1000), # ISO A-series (A4, A3, A2, ...)
(11, 85), # US Letter
(14, 85), # Legal
(17, 11), # Tabloid
]
resolutions: set[tuple[int, int]] = set()
for w in base_widths:
for ar_w, ar_h in aspect_ratios:
h = math.floor(round(w * ar_h / ar_w))
if 1 <= h <= 10000:
resolutions.add((w, h))
resolutions.add((h, w)) # portrait / landscape
return sorted(resolutions)
def get_valid_depth_color_pairs() -> Iterable[tuple[int, int]]:
"""
Generate valid combinations of color and bit depth values.
Yields tuples of (depth, color) representing valid color mode and bit depth
combinations. Supports grayscale (0), RGB (2, 3, 4, 6) color modes with
appropriate bit depths (1, 2, 4, 8, 16 bits).
Yields:
tuple: A tuple of (depth, color) where depth is an integer representing
bit depth and color is an integer representing the color mode.
"""
valid_color_depth = {
0: [1, 2, 4, 8, 16],
2: [8, 16],
3: [1, 2, 4, 8],
4: [8, 16],
6: [8, 16],
}
for color, depths in valid_color_depth.items():
for depth in depths:
yield depth, color
class PNG:
"""
A class representing a PNG image header (IHDR chunk).
This class encapsulates the metadata from a PNG file's IHDR (Image Header) chunk,
including image dimensions, bit depth, color type, interlacing method, and CRC checksum.
It provides methods for parsing PNG files, packing/unpacking header data, and generating
PNG crc if omitted.
"""
width: int
height: int
bit_depth: int
color_type: int
crc: int
interlace: int = 0
def __init__( # pylint: disable=R0913
self,
size: tuple[int, int],
bit_depth: int,
color_type: int,
*,
crc: int | None = None,
interlace: int = 0,
):
"""Initialize a PNG IHDR (Image Header) chunk.
Args:
size (tuple[int, int]): The width and height of the image in pixels.
bit_depth (int): The number of bits per sample or per palette index (1, 2, 4, 8, or 16).
color_type (int): The color type of the image (0=Grayscale, 2=RGB, 3=Indexed,
4=Grayscale+Alpha, 6=RGBA).
crc (int | None, optional): CRC (Cyclic Redundancy Check) checksum for the chunk.
If None, it will be calculated automatically. Defaults
to None.
interlace (int, optional): The interlace method (0=no interlace, 1=Adam7 interlace).
Defaults to 0.
"""
self.width = size[0]
self.height = size[1]
self.bit_depth = bit_depth
self.color_type = color_type
self.interlace = interlace
self.crc = crc or self.compute_crc()
def compute_crc(self) -> int:
"""Compute the CRC checksum for the PNG IHDR chunk.
Returns:
int: The computed CRC checksum.
"""
return (
zlib.crc32(
b"IHDR"
+ self.width.to_bytes(4, "big")
+ self.height.to_bytes(4, "big")
+ bytes((self.bit_depth, self.color_type, 0, 0, self.interlace))
)
& 0xFFFFFFFF
)
@classmethod
def from_packed(cls, packed: int, crc: int | None = None) -> Self:
"""Create a PNG instance from a packed integer representation.
Parse a packed integer containing width, height, bit depth, color type and interlace
bits and return a PNG instance. The CRC may be supplied; if omitted it will be
computed by the constructor.
Args:
packed (int): Packed integer with fields laid out as described in the `packed` property.
crc (int | None): Optional CRC value for the IHDR chunk.
Returns:
Self: PNG instance with parsed fields.
"""
interlace = packed & 0b1
color = (packed >> 1) & 0b111
depth = (packed >> 4) & 0b111
height = (packed >> 7) & 0xFFFF
width = (packed >> 23) & 0xFFFF
return cls(
size=(width, height),
bit_depth=depth,
color_type=color,
crc=crc,
interlace=interlace,
)
@cached_property
def packed(self) -> int:
"""Pack IHDR fields into a single integer.
Returns:
int: Packed integer encoding the IHDR fields.
Layout:
width : 16 bits
height : 16 bits
depth : 3 bits
color : 3 bits
interlace : 1 bit
"""
return (
((self.width & 0xFFFF) << 23)
| ((self.height & 0xFFFF) << 7)
| ((self.bit_depth & 0b111) << 4)
| ((self.color_type & 0b111) << 1)
| (self.interlace & 0b1)
)
@classmethod
def from_bytearray(cls, data: bytearray) -> Self:
"""Create a PNG image instance from a bytearray containing PNG file data.
Parses the PNG signature and IHDR (Image Header) chunk to extract image metadata
including dimensions, bit depth, color type, and interlacing information.
Args:
data (bytearray): A bytearray containing the complete PNG file data, starting
with the PNG signature (8 bytes: \\x89PNG\\r\\n\\x1a\\n).
Returns:
Self | None: A new instance of the PNG image class with parsed IHDR metadata
(width, height, bit_depth, color_type, crc, interlace) if the data is
valid and contains a proper IHDR chunk. Returns None if the PNG signature
is invalid or the IHDR chunk type is not found.
Raises:
struct.error: If the data is too short or malformed for unpacking IHDR fields.
IndexError: If the bytearray is shorter than expected for the PNG structure.
"""
png_signature = b"\x89PNG\r\n\x1a\n"
if bytes(data[:8]) != png_signature:
raise ValueError("Invalid PNG signature")
offset = 8 # Read IHDR chunk
length = struct.unpack(">I", data[offset : offset + 4])[0]
offset += 4
chunk_type = data[offset : offset + 4]
offset += 4
if bytes(chunk_type) != b"IHDR":
raise ValueError("IHDR chunk not found at expected position")
ihdr_data = data[offset : offset + length]
offset += length
crc = struct.unpack(">I", data[offset : offset + 4])[0]
# Unpack IHDR fields
width, height, depth, color, _, _, interlace = struct.unpack(">IIBBBBB", ihdr_data)
return cls(
size=(width, height),
bit_depth=depth,
color_type=color,
crc=crc,
interlace=interlace,
)
def __repr__(self) -> str:
return (
f"PNG(width={self.width}, height={self.height}, "
f"bit_depth={self.bit_depth}, color_type={self.color_type}, "
f"interlace={self.interlace}, crc=0x{self.crc:08x})"
)
aperisolve/models.py
+91 -26
View File
@@ -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:
aperisolve/static/js/aperisolve.js
-1
View File
@@ -10,7 +10,6 @@ const TOOL_ORDER = [
"openstego",
"zsteg",
"strings",
"image_resize",
];
function escapeHtml(text) {
aperisolve/utils/png.py
+393
View File
@@ -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
aperisolve/utils/utils.py
+110
View File
@@ -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
aperisolve/workers.py
-2
View File
@@ -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),