Validate vintage evidence image files

tests/test_validate_vintage_submission.py

@@ -1,5 +1,7 @@
 # SPDX-License-Identifier: MIT
 import importlib.util
+import struct
+import zlib
 from pathlib import Path
 
 
@@ -11,6 +13,36 @@
 SubmissionValidator = validate_vintage_submission.SubmissionValidator
 
 
+def write_png(path: Path, width: int = 640, height: int = 480) -> None:
+    def chunk(kind: bytes, data: bytes) -> bytes:
+        return (
+            struct.pack(">I", len(data))
+            + kind
+            + data
+            + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
+        )
+
+    raw_rows = b"".join(b"\x00" + (b"\xff\xff\xff" * width) for _ in range(height))
+    path.write_bytes(
+        b"\x89PNG\r\n\x1a\n"
+        + chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
+        + chunk(b"IDAT", zlib.compress(raw_rows))
+        + chunk(b"IEND", b"")
+    )
+
+
+def write_bmp(path: Path, width: int = 640, height: int = 480) -> None:
+    row_stride = ((width * 3 + 3) // 4) * 4
+    pixel_data_size = row_stride * height
+    file_size = 14 + 40 + pixel_data_size
+    path.write_bytes(
+        b"BM"
+        + struct.pack("<IHHI", file_size, 0, 0, 54)
+        + struct.pack("<IiiHHIIiiII", 40, width, height, 1, 24, 0, pixel_data_size, 0, 0, 0, 0)
+        + (b"\xff\xff\xff" * width + b"\x00" * (row_stride - width * 3)) * height
+    )
+
+
 def test_wallet_validation_accepts_rtc1_alphanumeric_range():
     validator = SubmissionValidator()
 
@@ -49,32 +81,67 @@ def test_attestation_log_json_requires_core_fields(tmp_path):
     assert result["checks"]["json_valid"] is True
 
 
-def test_photo_validation_preserves_size_and_format_warnings(tmp_path):
+def test_photo_validation_rejects_non_image_content(tmp_path):
     validator = SubmissionValidator()
-    photo_path = tmp_path / "photo.txt"
-    photo_path.write_bytes(b"tiny")
+    photo_path = tmp_path / "photo.jpg"
+    photo_path.write_bytes(b"not an image" * 1000)
+
+    result = validator.validate_photo(str(photo_path))
+
+    assert result["status"] == "FAIL"
+    assert "not a valid image" in result["message"]
+    assert result["checks"]["file_size_bytes"] == 12000
+    assert result["checks"]["format"] == ".jpg"
+
+
+def test_photo_validation_accepts_real_png_with_dimensions(tmp_path):
+    validator = SubmissionValidator()
+    photo_path = tmp_path / "photo.png"
+    write_png(photo_path)
+
+    result = validator.validate_photo(str(photo_path))
+
+    assert result["status"] == "PASS"
+    assert result["checks"]["image_type"] == "png"
+    assert result["checks"]["width"] == 640
+    assert result["checks"]["height"] == 480
+
+
+def test_photo_validation_accepts_real_bmp_with_dimensions(tmp_path):
+    validator = SubmissionValidator()
+    photo_path = tmp_path / "photo.bmp"
+    write_bmp(photo_path)
+
+    result = validator.validate_photo(str(photo_path))
+
+    assert result["status"] == "PASS"
+    assert result["checks"]["image_type"] == "bmp"
+    assert result["checks"]["width"] == 640
+    assert result["checks"]["height"] == 480
+
+
+def test_low_resolution_image_warning_is_in_summary(tmp_path):
+    validator = SubmissionValidator()
+    photo_path = tmp_path / "photo.png"
+    write_png(photo_path, width=320, height=240)
 
     result = validator.validate_photo(str(photo_path))
 
     assert result["status"] == "WARN"
-    assert "too small" in result["message"]
-    assert "Unusual photo format" in result["message"]
-    assert result["checks"]["file_size_bytes"] == 4
-    assert result["checks"]["format"] == ".txt"
-    assert "Photo file is unusually small" in validator.warnings
+    assert "resolution is too small" in result["message"]
+    assert validator.warnings == [result["message"]]
 
 
-def test_screenshot_validation_preserves_small_file_warning(tmp_path):
+def test_screenshot_validation_rejects_non_image_content(tmp_path):
     validator = SubmissionValidator()
     screenshot_path = tmp_path / "screenshot.png"
-    screenshot_path.write_bytes(b"tiny")
+    screenshot_path.write_bytes(b"not an image" * 1000)
 
     result = validator.validate_screenshot(str(screenshot_path))
 
-    assert result["status"] == "WARN"
-    assert "too small" in result["message"]
-    assert result["checks"]["file_size_bytes"] == 4
-    assert "Screenshot file is unusually small" in validator.warnings
+    assert result["status"] == "FAIL"
+    assert "not a valid image" in result["message"]
+    assert result["checks"]["file_size_bytes"] == 12000
 
 
 def test_validate_submission_extracts_arch_and_bounty_from_valid_log(tmp_path):

tools/validate_vintage_submission.py

@@ -17,6 +17,7 @@
 import argparse
 import json
 import os
+import struct
 import sys
 from datetime import datetime
 from typing import Dict, Any, Optional
@@ -29,6 +30,91 @@ def __init__(self):
         self.checks: Dict[str, Dict[str, Any]] = {}
         self.errors: list = []
         self.warnings: list = []
+
+    def _read_image_metadata(self, image_path: str) -> Dict[str, Any]:
+        """Read basic image type and dimensions without optional dependencies."""
+        with open(image_path, "rb") as f:
+            header = f.read(32)
+
+            if header.startswith(b"\x89PNG\r\n\x1a\n") and header[12:16] == b"IHDR":
+                width, height = struct.unpack(">II", header[16:24])
+                return {"image_type": "png", "width": width, "height": height}
+
+            if header[:6] in (b"GIF87a", b"GIF89a"):
+                width, height = struct.unpack("<HH", header[6:10])
+                return {"image_type": "gif", "width": width, "height": height}
+
+            if header.startswith(b"BM") and len(header) >= 26:
+                dib_size = struct.unpack("<I", header[14:18])[0]
+                if dib_size >= 40:
+                    width, height = struct.unpack("<ii", header[18:26])
+                    return {"image_type": "bmp", "width": abs(width), "height": abs(height)}
+
+            if header.startswith(b"\xff\xd8"):
+                f.seek(2)
+                while True:
+                    marker_prefix = f.read(1)
+                    if not marker_prefix:
+                        break
+                    if marker_prefix != b"\xff":
+                        continue
+
+                    marker = f.read(1)
+                    while marker == b"\xff":
+                        marker = f.read(1)
+                    if not marker:
+                        break
+
+                    if marker in b"\xd8\xd9":
+                        continue
+
+                    length_bytes = f.read(2)
+                    if len(length_bytes) != 2:
+                        break
+                    segment_length = struct.unpack(">H", length_bytes)[0]
+                    if segment_length < 2:
+                        break
+
+                    if marker in b"\xc0\xc1\xc2\xc3\xc5\xc6\xc7\xc9\xca\xcb\xcd\xce\xcf":
+                        segment = f.read(5)
+                        if len(segment) != 5:
+                            break
+                        height, width = struct.unpack(">HH", segment[1:5])
+                        return {"image_type": "jpeg", "width": width, "height": height}
+
+                    f.seek(segment_length - 2, os.SEEK_CUR)
+
+        raise ValueError("not a valid image")
+
+    def _validate_image_file(self, image_path: str, label: str) -> Dict[str, Any]:
+        file_size = os.path.getsize(image_path)
+        ext = os.path.splitext(image_path)[1].lower()
+        checks = {
+            "file_exists": True,
+            "file_size_bytes": file_size,
+            "format": ext,
+        }
+
+        try:
+            metadata = self._read_image_metadata(image_path)
+        except Exception as exc:
+            return {
+                "status": "FAIL",
+                "message": f"{label} is not a valid image: {exc}",
+                "checks": checks,
+            }
+
+        checks.update(metadata)
+        warnings = []
+        if metadata["width"] < 640 or metadata["height"] < 480:
+            warnings.append(f"{label} resolution is too small: {metadata['width']}x{metadata['height']}")
+        self.warnings.extend(warnings)
+
+        return {
+            "status": "WARN" if warnings else "PASS",
+            "message": "; ".join(warnings) if warnings else f"{label} is a valid image",
+            "checks": checks,
+        }
 
     def validate_photo(self, photo_path: str) -> Dict[str, Any]:
         """Validate photo evidence"""
@@ -43,37 +129,20 @@ def validate_photo(self, photo_path: str) -> Dict[str, Any]:
             result["message"] = f"Photo file not found: {photo_path}"
             return result
 
-        warning_messages = []
-
-        # Check file size (should be reasonable)
-        file_size = os.path.getsize(photo_path)
-        if file_size < 10000:  # Less than 10KB
-            warning_messages.append(f"Photo file seems too small: {file_size} bytes")
-            self.warnings.append("Photo file is unusually small")
-
         # Check file extension
         ext = os.path.splitext(photo_path)[1].lower()
         if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
-            warning_messages.append(f"Unusual photo format: {ext}")
             self.warnings.append(f"Unusual photo format: {ext}")
-        
+
         # In production, would check:
         # - EXIF timestamp
         # - Image content (machine + monitor)
         # - Metadata consistency
-
-        if warning_messages:
+
+        result = self._validate_image_file(photo_path, "Photo file")
+        if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.bmp'] and result["status"] == "PASS":
             result["status"] = "WARN"
-            result["message"] = "; ".join(warning_messages)
-        else:
-            result["status"] = "PASS"
-            result["message"] = "Photo file exists and appears valid"
-        result["checks"] = {
-            "file_exists": True,
-            "file_size_bytes": file_size,
-            "format": ext
-        }
-
+            result["message"] = f"Unusual photo format: {ext}"
         return result
 
     def validate_screenshot(self, screenshot_path: str) -> Dict[str, Any]:
@@ -89,22 +158,7 @@ def validate_screenshot(self, screenshot_path: str) -> Dict[str, Any]:
             result["message"] = f"Screenshot file not found: {screenshot_path}"
             return result
 
-        # Check file size
-        file_size = os.path.getsize(screenshot_path)
-        if file_size < 1000:  # Less than 1KB
-            result["status"] = "WARN"
-            result["message"] = f"Screenshot file seems too small: {file_size} bytes"
-            self.warnings.append("Screenshot file is unusually small")
-        else:
-            result["status"] = "PASS"
-            result["message"] = "Screenshot file exists"
-
-        result["checks"] = {
-            "file_exists": True,
-            "file_size_bytes": file_size
-        }
-
-        return result
+        return self._validate_image_file(screenshot_path, "Screenshot file")
 
     def validate_attestation_log(self, log_path: str) -> Dict[str, Any]:
         """Validate server-side attestation log"""