Aanvallen op modellen voor videobegrip

import numpy as np
from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class SamplingStrategy(Enum):
    UNIFORM = "uniform"
    KEYFRAME = "keyframe"
    SCENE_CHANGE = "scene_change"
    ATTENTION_BASED = "attention_based"
    HIERARCHICAL = "hierarchical"
 
@dataclass
class VideoProcessingConfig:
    """Configuratie voor hoe een model video-invoer verwerkt."""
    model_name: str
    sampling_strategy: SamplingStrategy
    num_frames: int
    max_resolution: tuple[int, int]
    supports_audio: bool
    max_duration_seconds: int
    temporal_encoding: str
 
VIDEO_MODEL_CONFIGS = {
    "gemini_2_5_pro": VideoProcessingConfig(
        model_name="Gemini 2.5 Pro",
        sampling_strategy=SamplingStrategy.HIERARCHICAL,
        num_frames=32,
        max_resolution=(1280, 720),
        supports_audio=True,
        max_duration_seconds=3600,
        temporal_encoding="Absolute timestamp tokens",
    ),
    "gpt_4o": VideoProcessingConfig(
        model_name="GPT-4o",
        sampling_strategy=SamplingStrategy.UNIFORM,
        num_frames=16,
        max_resolution=(1024, 1024),
        supports_audio=True,
        max_duration_seconds=300,
        temporal_encoding="Frame index embeddings",
    ),
    "video_llava": VideoProcessingConfig(
        model_name="Video-LLaVA",
        sampling_strategy=SamplingStrategy.UNIFORM,
        num_frames=8,
        max_resolution=(336, 336),
        supports_audio=False,
        max_duration_seconds=600,
        temporal_encoding="Positional embeddings",
    ),
}
 
def compute_sampled_frame_indices(
    total_frames: int,
    strategy: SamplingStrategy,
    num_samples: int,
    fps: float = 30.0,
) -> list[int]:
    """Bereken welke frame-indices een model zal samplen.
 
    Dit is de sleutelfunctie voor frame injection-aanvallen:
    weten welke frames worden gesampled vertelt de aanvaller
    precies waar adversariële content geplaatst moet worden.
    """
    if strategy == SamplingStrategy.UNIFORM:
        # Gelijkmatig verdeelde frames over de video
        if num_samples >= total_frames:
            return list(range(total_frames))
        step = total_frames / num_samples
        return [int(i * step) for i in range(num_samples)]
 
    elif strategy == SamplingStrategy.KEYFRAME:
        # Sample I-frames uit de video-codec (vereenvoudigd)
        gop_size = int(fps)  # Doorgaans één I-frame per seconde
        keyframes = list(range(0, total_frames, gop_size))
        if len(keyframes) > num_samples:
            step = len(keyframes) / num_samples
            return [keyframes[int(i * step)] for i in range(num_samples)]
        return keyframes
 
    elif strategy == SamplingStrategy.HIERARCHICAL:
        # Sample eerst grof, verfijn daarna interessegebieden
        coarse_samples = num_samples // 2
        coarse_step = total_frames / coarse_samples
        coarse_indices = [int(i * coarse_step) for i in range(coarse_samples)]
 
        # Simuleer verfijning (in de praktijk beslist het model welke gebieden)
        fine_indices = []
        for idx in coarse_indices:
            offset = int(coarse_step / 4)
            fine_indices.append(min(idx + offset, total_frames - 1))
 
        return sorted(set(coarse_indices + fine_indices))[:num_samples]
 
    else:
        # Standaard naar uniform
        step = total_frames / num_samples
        return [int(i * step) for i in range(num_samples)]
 
# Voorbeeld: Bepaal de posities van aanvalsframes
config = VIDEO_MODEL_CONFIGS["gpt_4o"]
video_length_seconds = 60
fps = 30.0
total_frames = int(video_length_seconds * fps)
 
sampled_indices = compute_sampled_frame_indices(
    total_frames=total_frames,
    strategy=config.sampling_strategy,
    num_samples=config.num_frames,
    fps=fps,
)
print(f"Model: {config.model_name}")
print(f"Total frames: {total_frames}")
print(f"Sampled frames: {len(sampled_indices)}")
print(f"Sampled indices: {sampled_indices}")
print(f"Adversarial frames needed: {len(sampled_indices)} out of {total_frames}")
print(f"Attack coverage: {len(sampled_indices)/total_frames*100:.2f}% of frames modified")

import cv2
import numpy as np
from PIL import Image, ImageDraw, ImageFont
from pathlib import Path
 
class FrameInjector:
    """Injecteer adversariële frames in videobestanden.
 
    Ondersteunt meerdere injectiestrategieën:
    - Single-frame: Eén adversariële frame op een gesamplede positie
    - Multi-frame: Meerdere adversariële frames op gesamplede posities
    - Subliminaal: Zeer korte (<50ms) adversariële frames
    - Geblend: Adversariële content geleidelijk in-/uitgefade
    """
 
    def __init__(self, video_path: str):
        self.video_path = video_path
        self.cap = cv2.VideoCapture(video_path)
        self.fps = self.cap.get(cv2.CAP_PROP_FPS)
        self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
        self.width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        self.height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        self.cap.release()
 
    def inject_single_frame(
        self,
        adversarial_frame: np.ndarray,
        target_frame_index: int,
        output_path: str,
    ) -> dict:
        """Vervang een enkele frame door een adversariële frame.
 
        De adversariële frame wordt geplaatst op target_frame_index,
        die zou moeten overeenkomen met een frame die het model zal samplen.
        Bij 30fps duurt een enkele frame 33ms -- doorgaans te kort
        voor een menselijke kijker om de content te lezen.
        """
        cap = cv2.VideoCapture(self.video_path)
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (self.width, self.height))
 
        frame_idx = 0
        injected = False
 
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
 
            if frame_idx == target_frame_index:
                # Pas de grootte van de adversariële frame aan op de videodimensies
                adv_resized = cv2.resize(adversarial_frame, (self.width, self.height))
                out.write(adv_resized)
                injected = True
            else:
                out.write(frame)
 
            frame_idx += 1
 
        cap.release()
        out.release()
 
        return {
            "output_path": output_path,
            "injected_at_frame": target_frame_index,
            "injected_at_time": target_frame_index / self.fps,
            "frame_duration_ms": 1000 / self.fps,
            "total_frames": self.total_frames,
            "injection_successful": injected,
        }
 
    def inject_at_all_sample_points(
        self,
        adversarial_frame: np.ndarray,
        sampling_strategy: SamplingStrategy,
        num_model_samples: int,
        output_path: str,
    ) -> dict:
        """Injecteer adversariële frames op alle punten die het model zal samplen.
 
        Dit waarborgt dat het model adversariële content verwerkt ongeacht
        kleine variaties in de sampling-implementatie, terwijl de overgrote
        meerderheid van de frames (die menselijke beoordelaars zien) schoon blijft.
        """
        sampled_indices = compute_sampled_frame_indices(
            total_frames=self.total_frames,
            strategy=sampling_strategy,
            num_samples=num_model_samples,
            fps=self.fps,
        )
 
        # Voeg bufferframes toe rond elk samplepunt voor robuustheid
        injection_indices = set()
        for idx in sampled_indices:
            for offset in range(-2, 3):  # +/- 2 frames
                clamped = max(0, min(self.total_frames - 1, idx + offset))
                injection_indices.add(clamped)
 
        cap = cv2.VideoCapture(self.video_path)
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (self.width, self.height))
 
        frame_idx = 0
        injected_count = 0
 
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
 
            if frame_idx in injection_indices:
                adv_resized = cv2.resize(adversarial_frame, (self.width, self.height))
                out.write(adv_resized)
                injected_count += 1
            else:
                out.write(frame)
 
            frame_idx += 1
 
        cap.release()
        out.release()
 
        return {
            "output_path": output_path,
            "total_frames": self.total_frames,
            "injected_frames": injected_count,
            "injection_percentage": injected_count / self.total_frames * 100,
            "model_sample_points": len(sampled_indices),
        }
 
    def inject_subliminal(
        self,
        adversarial_frame: np.ndarray,
        target_time_seconds: float,
        duration_frames: int = 1,
        blend_frames: int = 2,
        output_path: str = "subliminal_output.mp4",
    ) -> dict:
        """Injecteer een subliminale adversariële frame met blending.
 
        De adversariële frame wordt ingefade over blend_frames,
        vastgehouden gedurende duration_frames en uitgefade over blend_frames.
        Dit creëert een vloeiendere visuele overgang die moeilijker
        te detecteren is voor menselijke beoordelaars, zelfs bij het
        frame voor frame doorlopen van de video.
        """
        target_frame = int(target_time_seconds * self.fps)
        start_blend = target_frame - blend_frames
        end_blend = target_frame + duration_frames + blend_frames
 
        cap = cv2.VideoCapture(self.video_path)
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (self.width, self.height))
 
        frame_idx = 0
        adv_resized = cv2.resize(adversarial_frame, (self.width, self.height))
 
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
 
            if start_blend <= frame_idx < target_frame:
                # Infade
                alpha = (frame_idx - start_blend) / max(blend_frames, 1)
                blended = cv2.addWeighted(frame, 1 - alpha, adv_resized, alpha, 0)
                out.write(blended)
            elif target_frame <= frame_idx < target_frame + duration_frames:
                # Volledige adversariële frame
                out.write(adv_resized)
            elif target_frame + duration_frames <= frame_idx < end_blend:
                # Uitfade
                alpha = 1 - (frame_idx - target_frame - duration_frames) / max(blend_frames, 1)
                blended = cv2.addWeighted(frame, 1 - alpha, adv_resized, alpha, 0)
                out.write(blended)
            else:
                out.write(frame)
 
            frame_idx += 1
 
        cap.release()
        out.release()
 
        return {
            "output_path": output_path,
            "target_time": target_time_seconds,
            "visible_duration_ms": (duration_frames + 2 * blend_frames) * (1000 / self.fps),
            "full_intensity_duration_ms": duration_frames * (1000 / self.fps),
        }

def create_adversarial_video_frame(
    payload: str,
    width: int = 1280,
    height: int = 720,
    style: str = "document",
) -> np.ndarray:
    """Maak een adversariële frame met geïnjecteerde instructies.
 
    De frame is ontworpen om eruit te zien als legitieme videocontent
    (een document, slide of schermopname), zodat als een menselijke
    beoordelaar erop pauzeert, deze niet onmiddellijk kwaadaardig lijkt.
 
    Args:
        payload: De adversariële instructietekst.
        width: Framebreedte in pixels.
        height: Framehoogte in pixels.
        style: Visuele stijl ('document', 'slide', 'terminal', 'whiteboard').
    """
    if style == "document":
        bg_color = (255, 255, 255)
        text_color = (40, 40, 40)
        header = "INTERNAL MEMO - CONFIDENTIAL"
    elif style == "slide":
        bg_color = (240, 240, 250)
        text_color = (30, 30, 80)
        header = "System Configuration Notes"
    elif style == "terminal":
        bg_color = (30, 30, 30)
        text_color = (0, 255, 0)
        header = "root@system:~#"
    elif style == "whiteboard":
        bg_color = (250, 248, 240)
        text_color = (40, 40, 40)
        header = "Meeting Notes"
    else:
        bg_color = (255, 255, 255)
        text_color = (0, 0, 0)
        header = ""
 
    img = Image.new("RGB", (width, height), color=bg_color)
    draw = ImageDraw.Draw(img)
 
    try:
        font = ImageFont.truetype(
            "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 18
        )
        header_font = ImageFont.truetype(
            "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 22
        )
    except OSError:
        font = ImageFont.load_default()
        header_font = font
 
    # Teken de header
    draw.text((40, 30), header, fill=text_color, font=header_font)
    draw.line([(40, 60), (width - 40, 60)], fill=text_color, width=1)
 
    # Teken de payload-tekst
    y = 80
    import textwrap
    lines = textwrap.wrap(payload, width=80)
    for line in lines:
        draw.text((40, y), line, fill=text_color, font=font)
        y += 28
 
    return np.array(img)[:, :, ::-1]  # Converteer RGB naar BGR voor OpenCV

class TemporalManipulator:
    """Manipuleer de temporele structuur van video's om
    modellen voor videobegrip te misleiden over de sequentie van gebeurtenissen.
 
    Deze aanvallen buiten de afhankelijkheid van het model van gesamplede frames uit
    om temporele narratieven te reconstrueren. Door het herordenen, dupliceren
    of verwijderen van frames op strategische posities kan het model
    tot onjuiste conclusies over oorzaak en gevolg worden gebracht.
    """
 
    def __init__(self, video_path: str):
        self.video_path = video_path
        cap = cv2.VideoCapture(video_path)
        self.fps = cap.get(cv2.CAP_PROP_FPS)
        self.total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        cap.release()
 
    def reverse_segment(
        self,
        start_time: float,
        end_time: float,
        output_path: str,
    ) -> dict:
        """Keer een segment van de video om om de waargenomen causaliteit te wijzigen.
 
        Als het model frames uit het omgekeerde segment sampelt,
        kan het concluderen dat gebeurtenissen in de tegenovergestelde volgorde plaatsvonden.
        """
        start_frame = int(start_time * self.fps)
        end_frame = int(end_time * self.fps)
 
        cap = cv2.VideoCapture(self.video_path)
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
        # Lees alle frames in het segment
        segment_frames = []
        frame_idx = 0
        all_frames = []
 
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            all_frames.append(frame)
            if start_frame <= frame_idx <= end_frame:
                segment_frames.append(frame)
            frame_idx += 1
 
        cap.release()
 
        # Keer het segment om
        segment_frames.reverse()
 
        # Schrijf de uitvoer
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (width, height))
 
        seg_idx = 0
        for i, frame in enumerate(all_frames):
            if start_frame <= i <= end_frame:
                out.write(segment_frames[seg_idx])
                seg_idx += 1
            else:
                out.write(frame)
 
        out.release()
 
        return {
            "output_path": output_path,
            "reversed_segment": f"{start_time:.1f}s - {end_time:.1f}s",
            "frames_affected": len(segment_frames),
        }
 
    def duplicate_frame_at_sample_points(
        self,
        source_frame_index: int,
        sampling_strategy: SamplingStrategy,
        num_model_samples: int,
        output_path: str,
    ) -> dict:
        """Vervang frames op de samplepunten van het model door een duplicaat van
        een specifieke frame, waardoor het model dat moment overmatig zwaar weegt.
 
        Dit kan het model laten geloven dat een specifieke gebeurtenis in de video
        de dominante of enige gebeurtenis is, waardoor het begrip van
        andere gebeurtenissen wordt onderdrukt.
        """
        sampled_indices = compute_sampled_frame_indices(
            total_frames=self.total_frames,
            strategy=sampling_strategy,
            num_samples=num_model_samples,
            fps=self.fps,
        )
 
        cap = cv2.VideoCapture(self.video_path)
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
        # Lees de bronframe
        cap.set(cv2.CAP_PROP_POS_FRAMES, source_frame_index)
        ret, source_frame = cap.read()
        if not ret:
            cap.release()
            raise ValueError(f"Could not read frame {source_frame_index}")
 
        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        out = cv2.VideoWriter(output_path, fourcc, self.fps, (width, height))
 
        frame_idx = 0
        replaced = 0
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            if frame_idx in sampled_indices and frame_idx != source_frame_index:
                out.write(source_frame)
                replaced += 1
            else:
                out.write(frame)
            frame_idx += 1
 
        cap.release()
        out.release()
 
        return {
            "output_path": output_path,
            "source_frame": source_frame_index,
            "replaced_frames": replaced,
            "model_will_see": "Same frame repeated at most sample points",
        }

class VideoConsistencyChecker:
    """Controleer videoframes op injection- en manipulatie-aanvallen.
 
    Vergelijkt aangrenzende frames om abrupte contentveranderingen
    te detecteren die wijzen op frame injection. Analyseert de distributie
    van visuele kenmerken over gesamplede frames om
    temporele manipulatie te detecteren.
    """
 
    def __init__(self, anomaly_threshold: float = 0.3):
        self.anomaly_threshold = anomaly_threshold
 
    def check_frame_consistency(
        self, frames: list[np.ndarray]
    ) -> dict:
        """Controleer op abrupte visuele veranderingen tussen opeenvolgende frames.
 
        Legitieme video heeft vloeiende overgangen tussen frames.
        Geïnjecteerde frames creëren scherpe discontinuïteiten in pixel-
        statistieken, kleurhistogrammen en structurele kenmerken.
        """
        anomalies = []
 
        for i in range(1, len(frames)):
            prev_frame = frames[i - 1].astype(float)
            curr_frame = frames[i].astype(float)
 
            # Verschil op pixelniveau
            pixel_diff = np.mean(np.abs(prev_frame - curr_frame)) / 255.0
 
            # Histogramverschil
            hist_prev = cv2.calcHist([frames[i-1]], [0], None, [256], [0, 256])
            hist_curr = cv2.calcHist([frames[i]], [0], None, [256], [0, 256])
            hist_diff = cv2.compareHist(
                hist_prev.astype(np.float32),
                hist_curr.astype(np.float32),
                cv2.HISTCMP_BHATTACHARYYA,
            )
 
            # Gecombineerde anomaliescore
            anomaly_score = 0.6 * pixel_diff + 0.4 * hist_diff
 
            if anomaly_score > self.anomaly_threshold:
                anomalies.append({
                    "frame_index": i,
                    "anomaly_score": float(anomaly_score),
                    "pixel_diff": float(pixel_diff),
                    "histogram_diff": float(hist_diff),
                })
 
        return {
            "total_frames_checked": len(frames) - 1,
            "anomalies_detected": len(anomalies),
            "anomaly_details": anomalies,
            "recommendation": (
                "BLOCK" if len(anomalies) > 2
                else "REVIEW" if len(anomalies) > 0
                else "PASS"
            ),
        }