Beeldsteganografie voor AI-aanvallen

import numpy as np
from PIL import Image
from typing import Optional
 
class LSBSteganography:
    """Bed verborgen data in en extraheer deze met Least Significant Bit-encoding.
 
    Klassieke LSB-steganografie verbergt willekeurige binaire data in de
    minst significante bits van pixelwaarden. Voor AI-aanvallen is de
    verborgen data vijandige tekst die de visuele encoder van het model
    mogelijk detecteert als zwakke patronen in de afbeelding.
 
    Let op: Standaard-LSB-steganografie beïnvloedt het gedrag van een VLM
    niet direct, omdat het model visuele kenmerken op hoger niveau verwerkt,
    niet individuele pixel-LSB's. Maar in combinatie met
    extractie-dan-injectie-pipelines (bijv. een voorbewerkingsstap die
    verborgen tekst extraheert), kan LSB-encoding payloads afleveren
    die visuele inspectie ontwijken.
    """
 
    def encode(
        self,
        cover_image_path: str,
        payload: str,
        output_path: str,
        bits_per_channel: int = 1,
    ) -> dict:
        """Codeer een tekst-payload in de LSB's van een afbeelding.
 
        Args:
            cover_image_path: Pad naar de dekafbeelding.
            payload: Tekst om in de afbeelding te verbergen.
            output_path: Pad voor de uitvoer-steganografische afbeelding.
            bits_per_channel: Aantal te gebruiken LSB's per kleurkanaal (1-4).
        """
        img = Image.open(cover_image_path).convert("RGB")
        pixels = np.array(img)
 
        # Converteer payload naar binair
        payload_bytes = payload.encode("utf-8")
        # Voeg lengteheader vooraan toe (4 bytes, big-endian)
        length_header = len(payload_bytes).to_bytes(4, byteorder="big")
        data = length_header + payload_bytes
 
        # Converteer naar bit-array
        bits = []
        for byte in data:
            for bit_pos in range(8):
                bits.append((byte >> (7 - bit_pos)) & 1)
 
        # Controleer capaciteit
        capacity = pixels.size * bits_per_channel
        if len(bits) > capacity:
            raise ValueError(
                f"Payload ({len(bits)} bits) exceeds image capacity "
                f"({capacity} bits at {bits_per_channel} bits/channel)"
            )
 
        # Bed bits in in de pixel-LSB's
        flat_pixels = pixels.flatten()
        mask = 0xFF << bits_per_channel  # Wis doelbits
        bit_idx = 0
 
        for i in range(len(flat_pixels)):
            if bit_idx >= len(bits):
                break
            # Wis de doel-LSB's
            flat_pixels[i] = flat_pixels[i] & mask
            # Stel de payload-bits in
            value = 0
            for b in range(bits_per_channel):
                if bit_idx < len(bits):
                    value |= bits[bit_idx] << (bits_per_channel - 1 - b)
                    bit_idx += 1
            flat_pixels[i] |= value
 
        # Vorm opnieuw en sla op
        stego_pixels = flat_pixels.reshape(pixels.shape)
        stego_img = Image.fromarray(stego_pixels.astype(np.uint8))
        stego_img.save(output_path)
 
        return {
            "output_path": output_path,
            "payload_size_bytes": len(payload_bytes),
            "bits_embedded": len(bits),
            "capacity_used_pct": len(bits) / capacity * 100,
            "bits_per_channel": bits_per_channel,
            "psnr_db": self._compute_psnr(pixels, stego_pixels),
        }
 
    def decode(
        self,
        stego_image_path: str,
        bits_per_channel: int = 1,
    ) -> str:
        """Extraheer de verborgen payload uit een steganografische afbeelding."""
        img = Image.open(stego_image_path).convert("RGB")
        flat_pixels = np.array(img).flatten()
 
        # Extraheer bits
        bits = []
        for pixel_val in flat_pixels:
            for b in range(bits_per_channel):
                bits.append((pixel_val >> (bits_per_channel - 1 - b)) & 1)
 
        # Lees lengteheader (4 bytes = 32 bits)
        length_bits = bits[:32]
        payload_length = 0
        for bit in length_bits:
            payload_length = (payload_length << 1) | bit
 
        # Lees payload
        payload_bits = bits[32 : 32 + payload_length * 8]
        payload_bytes = bytearray()
        for i in range(0, len(payload_bits), 8):
            byte_val = 0
            for bit in payload_bits[i : i + 8]:
                byte_val = (byte_val << 1) | bit
            payload_bytes.append(byte_val)
 
        return payload_bytes.decode("utf-8", errors="replace")
 
    def _compute_psnr(
        self, original: np.ndarray, modified: np.ndarray
    ) -> float:
        """Bereken de Peak Signal-to-Noise Ratio tussen afbeeldingen."""
        mse = np.mean((original.astype(float) - modified.astype(float)) ** 2)
        if mse == 0:
            return float("inf")
        return 10 * np.log10(255.0 ** 2 / mse)

class DCTSteganography:
    """Bed payloads in in het DCT-domein voor JPEG-robuuste steganografie.
 
    JPEG-compressie werkt in het DCT-domein, dus payloads
    die in DCT-coëfficiënten zijn ingebed, overleven JPEG-hercompressie.
    Dit is cruciaal voor AI-aanvallen omdat veel beeldverwerkings-
    pipelines afbeeldingen opnieuw coderen als JPEG vóór de modelverwerking.
    """
 
    def embed_in_dct(
        self,
        cover_image_path: str,
        payload: str,
        output_path: str,
        coefficient_selection: str = "mid_frequency",
    ) -> dict:
        """Bed payload in in de DCT-coëfficiënten van een JPEG-afbeelding.
 
        Args:
            cover_image_path: Pad naar de dek-JPEG-afbeelding.
            payload: Tekst-payload om in te bedden.
            output_path: Uitvoerpad voor de steganografische JPEG.
            coefficient_selection: Welke DCT-coëfficiënten te wijzigen.
                'low_frequency': Robuuster maar zichtbaarder
                'mid_frequency': Balans tussen robuustheid en onzichtbaarheid
                'high_frequency': Het meest onzichtbaar maar het minst robuust
        """
        # De implementatie wijzigt gekwantiseerde DCT-coëfficiënten
        # in de JPEG-bestandsstructuur. Dit vereist JPEG-manipulatie
        # op laag niveau (bijv. met jpegio of libjpeg).
 
        robustness_map = {
            "low_frequency": {
                "survives_recompression": True,
                "survives_resizing": True,
                "visual_impact": "Moderate",
                "psnr_typical": "35-40 dB",
            },
            "mid_frequency": {
                "survives_recompression": True,
                "survives_resizing": False,
                "visual_impact": "Low",
                "psnr_typical": "40-45 dB",
            },
            "high_frequency": {
                "survives_recompression": False,
                "survives_resizing": False,
                "visual_impact": "Very Low",
                "psnr_typical": "45-55 dB",
            },
        }
 
        return {
            "output_path": output_path,
            "payload_size": len(payload.encode()),
            "coefficient_selection": coefficient_selection,
            "robustness": robustness_map.get(coefficient_selection, {}),
            "note": (
                "DCT-domain embedding survives JPEG recompression, "
                "making it effective for payloads that pass through "
                "image processing pipelines"
            ),
        }

class VisualPatternSteganography:
    """Creëer afbeeldingen met patronen die het gedrag van een VLM beïnvloeden
    zonder opvallend te zijn voor menselijke beoordelaars.
 
    Deze techniek misbruikt de kloof tussen menselijke visuele perceptie
    en de kenmerkextractie van de visuele encoder. Patronen op specifieke
    ruimtelijke frequenties, onder de menselijke contrastgevoeligheid, kunnen
    kenmerken in de visuele encoder activeren die overeenkomen met tekst of instructies.
 
    Referentie: Zou et al., "Universal and Transferable Adversarial
    Attacks on Aligned Language Models" (2023).
    """
 
    def __init__(self, target_resolution: tuple[int, int] = (224, 224)):
        self.target_resolution = target_resolution
 
    def create_frequency_pattern(
        self,
        base_image_path: str,
        target_text: str,
        frequency_band: str = "mid",
        amplitude: float = 0.02,
        output_path: Optional[str] = None,
    ) -> dict:
        """Creëer een patroon in het frequentiedomein dat doeltekst codeert.
 
        Het patroon is een gestructureerd ruissignaal in een specifieke
        frequentieband. Wanneer ze door visuele encoders uit de CLIP-familie
        worden verwerkt, produceren deze patronen kenmerken die overlappen met
        de embedding van de doeltekst, waardoor tekstachtige informatie effectief
        in de afbeelding wordt gecodeerd zonder zichtbare tekst.
 
        Args:
            base_image_path: Dekafbeelding.
            target_text: Tekst om als visuele patronen te "coderen".
            frequency_band: 'low' (2-8 cycli), 'mid' (8-32), 'high' (32-112).
            amplitude: Patroonamplitude (0.01-0.05 typisch).
            output_path: Waar het resultaat moet worden opgeslagen.
        """
        img = Image.open(base_image_path).convert("RGB")
        img_array = np.array(img).astype(float) / 255.0
 
        # Creëer patroon in het frequentiedomein
        h, w = img_array.shape[:2]
 
        # Definieer frequentieband
        bands = {
            "low": (2, 8),
            "mid": (8, 32),
            "high": (32, min(h, w) // 2),
        }
        freq_low, freq_high = bands.get(frequency_band, bands["mid"])
 
        # Genereer gestructureerd patroon in het frequentiedomein
        pattern = np.zeros((h, w), dtype=complex)
 
        # Creëer een pseudo-codering van de doeltekst als frequentiecomponenten
        # Elk teken beïnvloedt specifieke frequentiecomponenten
        np.random.seed(hash(target_text) % (2**31))
        for i, char in enumerate(target_text):
            freq_x = freq_low + (ord(char) * (i + 1)) % (freq_high - freq_low)
            freq_y = freq_low + (ord(char) * (i + 2)) % (freq_high - freq_low)
            if freq_x < h // 2 and freq_y < w // 2:
                phase = (ord(char) / 128.0) * 2 * np.pi
                pattern[freq_x, freq_y] = amplitude * np.exp(1j * phase)
                # Hermitische symmetrie voor reële uitvoer
                pattern[h - freq_x, w - freq_y] = np.conj(pattern[freq_x, freq_y])
 
        # Converteer naar ruimtelijk domein
        spatial_pattern = np.real(np.fft.ifft2(pattern))
 
        # Normaliseer en pas toe op alle kanalen
        spatial_pattern = spatial_pattern / (np.max(np.abs(spatial_pattern)) + 1e-10)
        spatial_pattern = spatial_pattern * amplitude
 
        for channel in range(3):
            img_array[:, :, channel] += spatial_pattern
 
        # Clip en converteer terug
        img_array = np.clip(img_array, 0, 1)
        result = Image.fromarray((img_array * 255).astype(np.uint8))
 
        if output_path:
            result.save(output_path)
 
        return {
            "output_path": output_path,
            "frequency_band": frequency_band,
            "amplitude": amplitude,
            "target_text": target_text,
            "psnr_db": self._compute_psnr_from_arrays(
                np.array(Image.open(base_image_path).convert("RGB")),
                np.array(result),
            ),
            "human_perceptible": amplitude > 0.04,
        }
 
    def create_texture_pattern(
        self,
        base_image_path: str,
        pattern_type: str = "noise",
        seed: int = 42,
        strength: float = 0.015,
        output_path: Optional[str] = None,
    ) -> dict:
        """Leg een subtiel textuurpatroon over dat de kenmerken van de visuele encoder beïnvloedt.
 
        Textuurpatronen op specifieke schalen activeren verschillende kenmerken
        in convolutionele en op transformers gebaseerde visuele encoders. Deze
        techniek codeert geen specifieke tekst, maar stuurt de visuele kenmerken
        van het model richting specifieke semantische regio's.
        """
        img = Image.open(base_image_path).convert("RGB")
        img_array = np.array(img).astype(float) / 255.0
 
        np.random.seed(seed)
        h, w = img_array.shape[:2]
 
        if pattern_type == "noise":
            texture = np.random.randn(h, w) * strength
        elif pattern_type == "grid":
            texture = np.zeros((h, w))
            spacing = 16
            texture[::spacing, :] = strength
            texture[:, ::spacing] = strength
        elif pattern_type == "wave":
            x = np.arange(w)
            y = np.arange(h)
            xx, yy = np.meshgrid(x, y)
            texture = strength * np.sin(2 * np.pi * xx / 32) * np.sin(2 * np.pi * yy / 32)
        else:
            texture = np.zeros((h, w))
 
        for channel in range(3):
            img_array[:, :, channel] += texture
 
        img_array = np.clip(img_array, 0, 1)
        result = Image.fromarray((img_array * 255).astype(np.uint8))
 
        if output_path:
            result.save(output_path)
 
        return {
            "output_path": output_path,
            "pattern_type": pattern_type,
            "strength": strength,
        }
 
    def _compute_psnr_from_arrays(
        self, original: np.ndarray, modified: np.ndarray
    ) -> float:
        mse = np.mean((original.astype(float) - modified.astype(float)) ** 2)
        if mse == 0:
            return float("inf")
        return float(10 * np.log10(255.0 ** 2 / mse))

class RobustnessTester:
    """Test of steganografische payloads beeldverwerking overleven.
 
    Productie-beeldverwerkingspipelines passen doorgaans toe:
    - JPEG-compressie (kwaliteit 60-90)
    - Verkleining (tot maximale afmetingen)
    - Conversie van kleurruimte
    - Verwijdering van metadata
    - Optionele ruisreductie
 
    De payload moet dit alles overleven om het model te bereiken.
    """
 
    def __init__(self):
        self.transformations = [
            ("jpeg_q85", self._apply_jpeg_compression, {"quality": 85}),
            ("jpeg_q60", self._apply_jpeg_compression, {"quality": 60}),
            ("resize_50pct", self._apply_resize, {"scale": 0.5}),
            ("resize_200pct", self._apply_resize, {"scale": 2.0}),
            ("gaussian_noise", self._apply_noise, {"std": 5.0}),
            ("gaussian_blur", self._apply_blur, {"radius": 1.0}),
        ]
 
    def test_robustness(
        self,
        stego_image_path: str,
        decoder_fn,
        expected_payload: str,
    ) -> dict:
        """Test de overleving van de payload onder verschillende transformaties."""
        results = {}
 
        for name, transform_fn, kwargs in self.transformations:
            img = Image.open(stego_image_path)
            transformed = transform_fn(img, **kwargs)
 
            # Sla tijdelijk op en probeer te decoderen
            temp_path = f"/tmp/robustness_test_{name}.png"
            transformed.save(temp_path)
 
            try:
                decoded = decoder_fn(temp_path)
                survived = decoded == expected_payload
                similarity = self._string_similarity(decoded, expected_payload)
            except Exception:
                survived = False
                similarity = 0.0
 
            results[name] = {
                "survived": survived,
                "similarity": similarity,
                "transformation": name,
            }
 
        return {
            "results": results,
            "survival_rate": sum(
                1 for r in results.values() if r["survived"]
            ) / len(results),
            "recommended_technique": (
                "dct_domain" if not results.get("jpeg_q85", {}).get("survived", True)
                else "lsb"
            ),
        }
 
    def _apply_jpeg_compression(self, img: Image.Image, quality: int) -> Image.Image:
        from io import BytesIO
        buffer = BytesIO()
        img.save(buffer, format="JPEG", quality=quality)
        buffer.seek(0)
        return Image.open(buffer).convert("RGB")
 
    def _apply_resize(self, img: Image.Image, scale: float) -> Image.Image:
        new_size = (int(img.width * scale), int(img.height * scale))
        return img.resize(new_size, Image.LANCZOS)
 
    def _apply_noise(self, img: Image.Image, std: float) -> Image.Image:
        arr = np.array(img).astype(float)
        noise = np.random.randn(*arr.shape) * std
        noisy = np.clip(arr + noise, 0, 255).astype(np.uint8)
        return Image.fromarray(noisy)
 
    def _apply_blur(self, img: Image.Image, radius: float) -> Image.Image:
        from PIL import ImageFilter
        return img.filter(ImageFilter.GaussianBlur(radius=radius))
 
    def _string_similarity(self, a: str, b: str) -> float:
        if not a or not b:
            return 0.0
        matches = sum(1 for ca, cb in zip(a, b) if ca == cb)
        return matches / max(len(a), len(b))

class SteganalysisDetector:
    """Detecteer steganografische inhoud in afbeeldingen.
 
    Gebruikt statistische analyse om afbeeldingen te identificeren die
    waarschijnlijk verborgen data bevatten. Meerdere detectiemethoden worden
    gecombineerd voor verbeterde nauwkeurigheid.
    """
 
    def chi_square_analysis(self, image_path: str) -> dict:
        """Detecteer LSB-steganografie met chi-kwadraat-analyse.
 
        LSB-inbedding creëert karakteristieke patronen in de
        verdeling van pixelparen (2k, 2k+1). De chi-kwadraat-
        test detecteert deze patronen met hoge nauwkeurigheid.
        """
        img = Image.open(image_path).convert("L")
        pixels = np.array(img).flatten()
 
        # Tel pixelwaardeparen
        pair_counts = np.zeros(128)
        for i in range(0, 256, 2):
            count_even = np.sum(pixels == i)
            count_odd = np.sum(pixels == i + 1)
            expected = (count_even + count_odd) / 2
            if expected > 0:
                chi_sq = (count_even - expected) ** 2 / expected
                pair_counts[i // 2] = chi_sq
 
        total_chi_sq = np.sum(pair_counts)
        # Vrijheidsgraden = aantal paren met waarnemingen - 1
        df = np.sum(pair_counts > 0) - 1
 
        # Hoge chi-kwadraat suggereert LSB-inbedding
        p_value = 1.0  # Vereenvoudigd; gebruik scipy.stats.chi2.sf voor echte berekening
        if df > 0:
            # Benaderde p-waarde
            normalized = total_chi_sq / max(df, 1)
            stego_likelihood = min(1.0, max(0.0, 1.0 - 1.0 / (1.0 + normalized)))
        else:
            stego_likelihood = 0.0
 
        return {
            "chi_square_statistic": float(total_chi_sq),
            "degrees_of_freedom": int(df),
            "stego_likelihood": float(stego_likelihood),
            "detection": "SUSPICIOUS" if stego_likelihood > 0.5 else "CLEAN",
        }
 
    def rs_analysis(self, image_path: str) -> dict:
        """Regular-Singular (RS)-analyse voor detectie van LSB-steganografie.
 
        RS-analyse meet het aandeel "reguliere" en "singuliere"
        pixelgroepen voor en na het omklappen van de LSB. Natuurlijke afbeeldingen
        vertonen specifieke R/S-verhoudingen die voorspelbaar veranderen met inbedding.
        """
        img = Image.open(image_path).convert("L")
        pixels = np.array(img).astype(float)
 
        # Vereenvoudigde RS-analyse
        h, w = pixels.shape
        block_size = 4
        regular_count = 0
        singular_count = 0
        total_blocks = 0
 
        for y in range(0, h - block_size, block_size):
            for x in range(0, w - block_size, block_size):
                block = pixels[y : y + block_size, x : x + block_size]
 
                # Bereken gladheid (som van verschillen tussen aangrenzende pixels)
                smoothness = np.sum(np.abs(np.diff(block, axis=0))) + \
                             np.sum(np.abs(np.diff(block, axis=1)))
 
                # Klap LSB's om en herbereken
                flipped = block.copy()
                flipped = np.where(flipped % 2 == 0, flipped + 1, flipped - 1)
                smoothness_flipped = np.sum(np.abs(np.diff(flipped, axis=0))) + \
                                     np.sum(np.abs(np.diff(flipped, axis=1)))
 
                if smoothness_flipped > smoothness:
                    regular_count += 1
                elif smoothness_flipped < smoothness:
                    singular_count += 1
                total_blocks += 1
 
        r_ratio = regular_count / max(total_blocks, 1)
        s_ratio = singular_count / max(total_blocks, 1)
 
        # In natuurlijke afbeeldingen is R > S. LSB-inbedding duwt R en S dichter bij elkaar
        rs_difference = abs(r_ratio - s_ratio)
        stego_likelihood = 1.0 - min(1.0, rs_difference / 0.1)
 
        return {
            "regular_ratio": float(r_ratio),
            "singular_ratio": float(s_ratio),
            "rs_difference": float(rs_difference),
            "stego_likelihood": float(stego_likelihood),
            "detection": "SUSPICIOUS" if stego_likelihood > 0.6 else "CLEAN",
        }