diff --git a/aperisolve/analyzers/image_resize.py b/aperisolve/analyzers/image_resize.py index 17db9b5..26a6c9b 100644 --- a/aperisolve/analyzers/image_resize.py +++ b/aperisolve/analyzers/image_resize.py @@ -8,191 +8,163 @@ from typing import Any, List from ..app import app from ..models import IHDR -from .utils import PNG, update_data +from .base_analyzer import SubprocessAnalyzer +from .utils import PNG -def write_recovered_png( - png_bytes: bytearray, - width: int, - height: int, - output_path: Path, -) -> None: - """ - Rebuilds a PNG by patching IHDR width/height and writes it to disk. - """ - height_bytes = height.to_bytes(4, byteorder="big") - width_bytes = width.to_bytes(4, byteorder="big") +class ResizeAnalyzer(SubprocessAnalyzer): + """Analyzer for image resize.""" - full_png_data = ( - png_bytes[:16] # Splicing: [Start...IHDR+4] + [W] + [H] + [IHDR+12...End] - + width_bytes - + height_bytes - + png_bytes[24:] - ) + png: PNG - with output_path.open("wb") as out_f: - out_f.write(full_png_data) + def __init__(self, *args): + super().__init__("image_resize", *args, has_archive=True) + def _lookup_crc(self) -> List[PNG]: + """ + Look up PNG images in the database by their CRC value. -def lookup_crc(png: PNG) -> 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 - Args: - png (PNG): A PNG object containing the CRC value to search for. + def _search_height_crc(self) -> List[PNG]: + """ + Search for PNG images with matching CRC by iterating through possible heights. - 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=png.crc).all() + 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 img in imgs: - packed = img.packed - results.append(PNG.from_packed(packed, crc=png.crc)) - return results - - -def search_height_crc(png: PNG) -> 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. - - Args: - png (PNG): The reference PNG object whose CRC checksum will be matched. - - 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=(png.width, height), - bit_depth=png.bit_depth, - color_type=png.color_type, - interlace=png.interlace, - ) - if candidate.crc == png.crc: - results.append(candidate) - return results - - -def analyze_image_resize(input_img: Path, output_dir: Path) -> None: - """ - 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. - - Args: - input_img (Path): Path to the input PNG image file to analyze. - output_dir (Path): Path to the directory where recovered PNG images will be saved. - - Returns: - None - - 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(). - """ - - # 1. Setup Paths - input_img = Path(input_img) - output_dir = Path(output_dir) - - # 3. Initialize Logs - logs = [] - - try: - with input_img.open("rb") as image: - img_bytes = bytearray(image.read()) - try: - png = PNG.from_bytearray(img_bytes) - except ValueError: - png = None - - if png is None: - logs.append( - "Failure: IHDR chunk is not the first chunk, or PNG has invalid structure." + 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, ) - update_data( - output_dir, - { - "image_resize": { + 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: + None + + 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 = [] + + try: + 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.", } - }, - ) - return - logs.append(f"Target CRC found: 0x{png.crc:08x}") + # 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}.", + } - if png.crc == png.compute_crc(): - note = ( - f"PNG is already valid with dimensions {png.width}x{png.height} " - f"and crc 0x{png.crc:08x}." - ) - output_no_compute: dict[str, Any] = { - "image_resize": {"status": "ok", "note": note} - } - update_data(output_dir, output_no_compute) - return + saved_img_urls = [] - saved_img_urls = [] - - db_matches = lookup_crc(png) - if not db_matches: - db_matches = search_height_crc(png) - if not db_matches: - logs.append("Failure: No matching dimensions found.") - else: - # Create parent directories if they don't exist - output_dir.mkdir(parents=True, exist_ok=True) - - # Iterate through ALL candidates found - for im_match in db_matches: - img_name = f"recovered_{im_match.width}x{im_match.height}.png" - output_path = output_dir / img_name - write_recovered_png( - img_bytes, im_match.width, im_match.height, output_path - ) + # 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(output_dir.name) / img_name)) + saved_img_urls.append("/image/" + str(Path(self.output_dir.name) / img_name)) - # 4. Final Data Update - output_data: dict[str, Any] = { - "image_resize": { + return { "status": "ok", "output": logs, "png_images": saved_img_urls, } - } - update_data(output_dir, output_data) + except Exception as e: + return {"status": "error", "error": str(e)} - except Exception as e: - update_data(output_dir, {"image_resize": {"status": "error", "error": str(e)}}) + +def analyze_image_resize(input_img: Path, output_dir: Path) -> None: + """Analyze an image submission using outguess.""" + analyzer = ResizeAnalyzer(input_img, output_dir) + analyzer.analyze() diff --git a/aperisolve/workers.py b/aperisolve/workers.py index 72b7861..8ab66ef 100644 --- a/aperisolve/workers.py +++ b/aperisolve/workers.py @@ -89,12 +89,12 @@ 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_strings, img_path, result_path), - (analyze_pngcheck, img_path, result_path), - (analyze_steghide, img_path, result_path, submission.password), - (analyze_openstego, img_path, result_path, submission.password), - (analyze_zsteg, 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_strings, img_path, result_path), + (analyze_steghide, img_path, result_path, submission.password), + (analyze_zsteg, img_path, result_path), ] # Deep analysis only