Lab: Video Model Adversarial Attacks

import cv2
import numpy as np
from typing import List, Tuple
from pathlib import Path
 
 
def load_video_frames(
    video_path: str,
    max_frames: int = 0,
    resize: Tuple[int, int] = None
) -> Tuple[List[np.ndarray], float, Tuple[int, int]]:
    """
    Load all frames from a video file.
 
    Returns:
        frames: List of numpy arrays (H, W, 3) in BGR format
        fps: Frames per second
        original_size: (width, height) of original video
    """
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise IOError(f"Cannot open video: {video_path}")
 
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
    frames = []
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        if resize:
            frame = cv2.resize(frame, resize)
        frames.append(frame)
        if max_frames > 0 and len(frames) >= max_frames:
            break
 
    cap.release()
    return frames, fps, (width, height)
 
 
def save_video(
    frames: List[np.ndarray],
    output_path: str,
    fps: float = 30.0
) -> str:
    """Save list of frames as a video file."""
    if not frames:
        raise ValueError("No frames to save")
 
    height, width = frames[0].shape[:2]
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
 
    for frame in frames:
        out.write(frame)
 
    out.release()
    return output_path
 
 
def uniform_sample_indices(total: int, num_samples: int) -> List[int]:
    """Compute uniform sampling indices matching video model behavior."""
    return [int(i * total / num_samples) for i in range(num_samples)]
 
 
def create_test_video(
    num_frames: int = 90,
    size: Tuple[int, int] = (224, 224),
    pattern: str = "gradient"
) -> List[np.ndarray]:
    """Create a synthetic test video for experimentation."""
    frames = []
    for i in range(num_frames):
        if pattern == "gradient":
            # Smooth color gradient that changes over time
            frame = np.zeros((size[1], size[0], 3), dtype=np.uint8)
            progress = i / num_frames
            frame[:, :, 0] = int(255 * progress)  # Blue increases
            frame[:, :, 1] = 128  # Green constant
            frame[:, :, 2] = int(255 * (1 - progress))  # Red decreases
        elif pattern == "moving_circle":
            frame = np.zeros((size[1], size[0], 3), dtype=np.uint8)
            center_x = int(size[0] * i / num_frames)
            center_y = size[1] // 2
            cv2.circle(frame, (center_x, center_y), 20, (0, 255, 0), -1)
        else:
            frame = np.random.randint(0, 256, (size[1], size[0], 3), dtype=np.uint8)
 
        frames.append(frame)
    return frames

import cv2
import numpy as np
from PIL import Image, ImageDraw, ImageFont
from typing import List
 
 
def text_injection_frame(
    original_frame: np.ndarray,
    text: str,
    position: str = "center",
    font_size: int = 24,
    opacity: float = 1.0
) -> np.ndarray:
    """
    Create a frame with injected text for video-LLM attacks.
    """
    # Convert BGR (OpenCV) to RGB (Pillow)
    frame_rgb = cv2.cvtColor(original_frame, cv2.COLOR_BGR2RGB)
    img = Image.fromarray(frame_rgb).convert("RGBA")
 
    overlay = Image.new("RGBA", img.size, (0, 0, 0, 0))
    draw = ImageDraw.Draw(overlay)
 
    try:
        font = ImageFont.truetype(
            "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", font_size
        )
    except OSError:
        font = ImageFont.load_default()
 
    # Calculate position
    bbox = draw.textbbox((0, 0), text, font=font)
    tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
    w, h = img.size
 
    positions = {
        "center": ((w - tw) // 2, (h - th) // 2),
        "top": ((w - tw) // 2, 10),
        "bottom": ((w - tw) // 2, h - th - 10),
        "corner": (10, 10),
    }
    pos = positions.get(position, positions["center"])
 
    alpha = int(255 * opacity)
    draw.text(pos, text, fill=(255, 255, 255, alpha), font=font)
 
    result = Image.alpha_composite(img, overlay).convert("RGB")
    return cv2.cvtColor(np.array(result), cv2.COLOR_RGB2BGR)
 
 
def adversarial_noise_frame(
    original_frame: np.ndarray,
    epsilon: int = 8,  # 0-255 range
    pattern: str = "uniform"
) -> np.ndarray:
    """
    Add adversarial noise pattern to a frame.
    """
    if pattern == "uniform":
        noise = np.random.randint(-epsilon, epsilon + 1, original_frame.shape)
    elif pattern == "structured":
        # Checkerboard pattern -- more effective against some models
        noise = np.zeros_like(original_frame)
        noise[::2, ::2] = epsilon
        noise[1::2, 1::2] = -epsilon
    elif pattern == "frequency":
        # High-frequency noise targeting specific spatial frequencies
        h, w = original_frame.shape[:2]
        x = np.linspace(0, 10 * np.pi, w)
        y = np.linspace(0, 10 * np.pi, h)
        xx, yy = np.meshgrid(x, y)
        noise = (epsilon * np.sin(xx + yy)).astype(int)
        noise = np.stack([noise] * 3, axis=-1)
    else:
        noise = np.zeros_like(original_frame)
 
    perturbed = np.clip(
        original_frame.astype(int) + noise, 0, 255
    ).astype(np.uint8)
    return perturbed
 
 
def inject_frames_at_sample_points(
    frames: List[np.ndarray],
    injection_fn,
    num_model_samples: int = 16,
    **kwargs
) -> List[np.ndarray]:
    """
    Apply an injection function only to frames that would be sampled.
    """
    total = len(frames)
    sample_indices = set(
        int(i * total / num_model_samples) for i in range(num_model_samples)
    )
 
    result = []
    injected = 0
    for i, frame in enumerate(frames):
        if i in sample_indices:
            result.append(injection_fn(frame, **kwargs))
            injected += 1
        else:
            result.append(frame.copy())
 
    print(f"Injected {injected}/{total} frames")
    return result

import numpy as np
from typing import List
 
 
def flicker_attack(
    frames: List[np.ndarray],
    perturbation_magnitude: int = 4,
    flicker_rate: int = 2
) -> List[np.ndarray]:
    """
    Apply alternating positive/negative perturbation across frames.
    Creates temporal signal invisible in individual frames.
    """
    h, w, c = frames[0].shape
    # Create a fixed spatial perturbation pattern
    pattern = np.random.randint(
        -perturbation_magnitude,
        perturbation_magnitude + 1,
        (h, w, c)
    )
 
    result = []
    for i, frame in enumerate(frames):
        sign = 1 if (i // flicker_rate) % 2 == 0 else -1
        perturbed = np.clip(
            frame.astype(int) + sign * pattern, 0, 255
        ).astype(np.uint8)
        result.append(perturbed)
 
    return result
 
 
def gradual_injection(
    frames: List[np.ndarray],
    target_overlay: np.ndarray,
    start_pct: float = 0.3,
    end_pct: float = 0.7,
    max_opacity: float = 0.15
) -> List[np.ndarray]:
    """
    Gradually blend adversarial content into video frames.
    Starts invisible and slowly becomes visible.
    """
    total = len(frames)
    start_idx = int(total * start_pct)
    end_idx = int(total * end_pct)
    duration = end_idx - start_idx
 
    result = []
    for i, frame in enumerate(frames):
        if start_idx <= i < end_idx:
            progress = (i - start_idx) / duration
            opacity = max_opacity * progress
            overlay_resized = cv2.resize(target_overlay, (frame.shape[1], frame.shape[0]))
            blended = cv2.addWeighted(
                frame, 1 - opacity, overlay_resized, opacity, 0
            )
            result.append(blended)
        else:
            result.append(frame.copy())
 
    return result
 
 
def compute_temporal_difference(frames: List[np.ndarray]) -> List[float]:
    """
    Compute frame-to-frame difference to detect injected frames.
    Useful for evaluating attack stealth.
    """
    diffs = []
    for i in range(1, len(frames)):
        diff = np.mean(np.abs(
            frames[i].astype(float) - frames[i-1].astype(float)
        ))
        diffs.append(diff)
    return diffs