File size: 4,011 Bytes
23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 23d337e 892fa81 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """
Image integrity forensics provider.
Delegates SHA-256 hashing to cores.vision.hashing and Pillow parsing to
cores.metadata.extractor — no duplicated hashing or Pillow-open logic.
"""
from __future__ import annotations
from typing import Any
import numpy as np
from config.settings import Settings, settings as _default_settings
from cores.metadata import extract_all
from cores.vision import sha256_bytes
from pipeline.feature_extraction import PipelineOutput
from providers.base import BaseProvider, ProviderCapability
class ImageIntegrityProvider(BaseProvider):
name = "image_integrity"
capability = ProviderCapability.FORENSICS
def __init__(self, settings: Settings | None = None) -> None:
super().__init__(settings=settings or _default_settings)
def is_available(self) -> bool:
return True
def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]:
original = pipeline_output.original_bytes
img = pipeline_output.image
h, w = img.shape[:2]
checks: dict[str, Any] = {}
integrity_issues: list[str] = []
# Check 1: Pillow decode (via cores.metadata to avoid double-open)
meta = extract_all(original) if original else {"error": "no bytes", "format": None}
pillow_ok = meta.get("error") is None
pillow_format = meta.get("format")
if not pillow_ok:
integrity_issues.append(f"Pillow decode: {meta.get('error')}")
checks["pillow_decode"] = pillow_ok
checks["pillow_format"] = pillow_format
# Check 2: dimensions sanity
checks["opencv_dimensions"] = {"width": w, "height": h}
# Check 3: file size sanity
size_bytes = len(original) if original else 0
pixels = w * h
if pixels > 0 and size_bytes > 0:
bpp = size_bytes / pixels
if bpp < 0.1:
integrity_issues.append(f"Unusually compressed: {bpp:.3f} bytes/pixel")
elif bpp > 50:
integrity_issues.append(f"Unusually large: {bpp:.2f} bytes/pixel (possible embedded payload)")
checks["size_bytes"] = size_bytes
checks["bytes_per_pixel"] = round(size_bytes / pixels, 4) if pixels else 0
# Check 4: SHA-256 (via cores.vision.hashing)
sha256 = sha256_bytes(original) if original else None
checks["sha256"] = sha256
# Check 5: JPEG APP markers count
app_markers = _count_jpeg_app_markers(original)
checks["jpeg_app_markers"] = app_markers
if app_markers > 5:
integrity_issues.append(f"High APP markers ({app_markers}) — possible embedded data")
integrity_score = max(0.0, 1.0 - 0.15 * len(integrity_issues))
raw = {
"checks": checks,
"integrity_issues": integrity_issues,
"integrity_score": integrity_score,
}
normalized = {
"integrity_score": round(integrity_score, 4),
"manipulation_indicators": integrity_issues,
"details": {
"sha256": sha256,
"format": pillow_format,
"size_bytes": size_bytes,
"bytes_per_pixel": checks["bytes_per_pixel"],
"app_markers": app_markers,
},
}
return raw, normalized
def _count_jpeg_app_markers(data: bytes | None) -> int:
"""Count JPEG APPn markers — quick steganography heuristic."""
if not data or data[:2] != b"\xff\xd8":
return 0
count = 0
i = 2
while i < len(data) - 1:
if data[i] != 0xFF:
break
marker = data[i + 1]
if marker in (0xD8, 0xD9):
i += 2
continue
if marker == 0x00 or 0xD0 <= marker <= 0xD7:
i += 2
continue
if i + 4 > len(data):
break
seg_len = (data[i + 2] << 8) | data[i + 3]
if 0xE0 <= marker <= 0xEF:
count += 1
i += 2 + seg_len
return count
|