多模态AI模型安全取证深度分析

import torch
import torch.nn.functional as F
from transformers import CLIPModel, CLIPProcessor

model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")

def analyze_attention_maps(image, text_inputs):
    inputs = processor(text=text_inputs, images=image, return_tensors="pt")
    outputs = model(**inputs, output_attentions=True)
    attentions = outputs.vision_model_output.attentions
    attention_statistics = []
    for layer_attn in attentions:
        attn_map = layer_attn[0].detach().numpy()
        attention_statistics.append({
            "mean": float(attn_map.mean()),
            "std": float(attn_map.std()),
            "max": float(attn_map.max()),
            "entropy": float(-1 * (attn_map * torch.log(torch.tensor(attn_map + 1e-10))).sum())
        })
    return attention_statistics

def detect_adversarial_perturbation(image, epsilon=0.01):
    pixel_values = processor(images=image, return_tensors="pt").pixel_values
    gradient = torch.randn_like(pixel_values) * epsilon
    perturbed = torch.clamp(pixel_values + gradient, -1, 1)
    orig_logits = model.get_image_features(pixel_values)
    pert_logits = model.get_image_features(perturbed)
    cosine_sim = F.cosine_similarity(orig_logits, pert_logits)
    return cosine_sim.item()

import torch
import numpy as np
from PIL import Image

def generate_adversarial_patch(model, target_image, target_class,
                                patch_size=64, num_steps=500,
                                learning_rate=0.05, epsilon=0.03):
    patch = torch.randn(3, patch_size, patch_size, requires_grad=True)
    optimizer = torch.optim.Adam([patch], lr=learning_rate)
    
    for step in range(num_steps):
        image_tensor = preprocess_image(target_image)
        x_offset = np.random.randint(0, image_tensor.shape[2] - patch_size)
        y_offset = np.random.randint(0, image_tensor.shape[3] - patch_size)
        
        patched_image = image_tensor.clone()
        patched_image[:, :, x_offset:x_offset+patch_size,
                      y_offset:y_offset+patch_size] = torch.clamp(
            patched_image[:, :, x_offset:x_offset+patch_size,
                          y_offset:y_offset+patch_size] + patch, -1, 1
        )
        
        output = model(patched_image)
        loss = -F.cross_entropy(output, torch.tensor([target_class]))
        
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        with torch.no_grad():
            patch.clamp_(-epsilon * 255, epsilon * 255)
    
    return patch.detach().numpy()

def detect_adversarial_patch(image_path, patch_size_range=(32, 128)):
    image = np.array(Image.open(image_path))
    gradient_maps = []
    for ps in range(patch_size_range[0], patch_size_range[1], 16):
        for x in range(0, image.shape[0] - ps, ps // 2):
            for y in range(0, image.shape[1] - ps, ps // 2):
                region = image[x:x+ps, y:y+ps]
                sobel_x = np.abs(np.diff(region, axis=1))
                sobel_y = np.abs(np.diff(region, axis=0))
                edge_energy = sobel_x.mean() + sobel_y.mean()
                gradient_maps.append((x, y, ps, edge_energy))
    
    if gradient_maps:
        energies = [g[3] for g in gradient_maps]
        threshold = np.mean(energies) + 3 * np.std(energies)
        suspicious_regions = [g for g in gradient_maps if g[3] > threshold]
        return suspicious_regions
    return []

import numpy as np
from scipy import fft as sp_fft
from PIL import Image

def frequency_domain_analysis(image_path):
    image = np.array(Image.open(image_path).convert('L'), dtype=np.float64)
    f_transform = sp_fft.fft2(image)
    f_shift = sp_fft.fftshift(f_transform)
    magnitude_spectrum = np.log(np.abs(f_shift) + 1e-10)
    
    h, w = magnitude_spectrum.shape
    center_h, center_w = h // 2, w // 2
    radius = min(h, w) // 4
    
    low_freq_mask = np.zeros_like(magnitude_spectrum)
    cv2_mask = np.ogrid[:h, :w]
    low_freq_mask[(cv2_mask[0] - center_h)**2 + (cv2_mask[1] - center_w)**2 <= radius**2] = 1
    
    high_freq_energy = magnitude_spectrum[low_freq_mask == 0].mean()
    low_freq_energy = magnitude_spectrum[low_freq_mask == 1].mean()
    hf_lf_ratio = high_freq_energy / (low_freq_energy + 1e-10)
    
    return {
        "high_freq_energy": float(high_freq_energy),
        "low_freq_energy": float(low_freq_energy),
        "hf_lf_ratio": float(hf_lf_ratio),
        "is_suspicious": hf_lf_ratio > 0.85
    }

def pixel_level_anomaly_detection(image_path):
    image = np.array(Image.open(image_path), dtype=np.float64)
    channel_anomalies = []
    
    for c in range(image.shape[2]):
        channel = image[:, :, c]
        diff_h = np.abs(np.diff(channel, axis=1))
        diff_v = np.abs(np.diff(channel, axis=0))
        
        channel_anomalies.append({
            "channel": ['R', 'G', 'B'][c],
            "h_gradient_mean": float(diff_h.mean()),
            "v_gradient_mean": float(diff_v.mean()),
            "h_gradient_std": float(diff_h.std()),
            "v_gradient_std": float(diff_v.std()),
            "lsb_entropy": float(calculate_lsb_entropy(channel))
        })
    
    return channel_anomalies

def calculate_lsb_entropy(channel):
    lsb_bits = channel.astype(np.uint8) & 1
    total_pixels = lsb_bits.size
    ones_count = lsb_bits.sum()
    p1 = ones_count / total_pixels
    p0 = 1 - p1
    if p0 > 0 and p1 > 0:
        return -(p0 * np.log2(p0) + p1 * np.log2(p1))
    return 0.0

import pytesseract
from PIL import Image, ImageDraw, ImageFont
import re

def detect_embedded_instructions(image_path):
    image = Image.open(image_path)
    ocr_result = pytesseract.image_to_data(image, output_type=pytesseract.Output.DICT)
    
    suspicious_patterns = [
        r'(?i)ignore\s+(all\s+)?previous',
        r'(?i)disregard\s+(all\s+)?instructions',
        r'(?i)you\s+are\s+now',
        r'(?i)system\s*prompt',
        r'(?i)override\s+safety',
        r'(?i)jailbreak',
        r'(?i)daniel\b',
        r'(?i)do\s+anything\s+now',
        r'[\u200b\u200c\u200d\ufeff]',
        r'[\u202e\u202d\u202c\u202b\u202a]',
    ]
    
    findings = []
    for i, text in enumerate(ocr_result['text']):
        if text.strip():
            for pattern in suspicious_patterns:
                if re.search(pattern, text):
                    findings.append({
                        "text": text,
                        "confidence": ocr_result['conf'][i],
                        "position": (ocr_result['left'][i], ocr_result['top'][i]),
                        "pattern": pattern,
                        "severity": "HIGH"
                    })
    
    return findings

def detect_zero_width_characters(image_path):
    image = Image.open(image_path)
    raw_data = list(image.tobytes())
    zero_width_chars = [b for b in raw_data if b in [0xe2, 0x80, 0x8b, 0x8c, 0x8d]]
    
    if len(zero_width_chars) > len(raw_data) * 0.01:
        return {"status": "SUSPICIOUS", "count": len(zero_width_chars)}
    return {"status": "CLEAN", "count": len(zero_width_chars)}

#!/bin/bash
LOG_DIR="${1:-/var/log/mllm}"
REPORT_FILE="/tmp/mllm_jailbreak_hunt_$(date +%Y%m%d).txt"

echo "=== Multi-Modal Jailbreak Hunt Report ===" > "$REPORT_FILE"
echo "Generated: $(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$REPORT_FILE"
echo "=========================================" >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "--- API Call Frequency Analysis ---" >> "$REPORT_FILE"
if [ -f "$LOG_DIR/api_access.log" ]; then
    awk '{print $1, $2, $4}' "$LOG_DIR/api_access.log" | \
        grep -i "multimodal\|vision\|image" | \
        awk '{count[$1" "$2]++} END {for (k in count) if (count[k]>50) print count[k], k}' | \
        sort -rn | head -20 >> "$REPORT_FILE"
fi

echo "" >> "$REPORT_FILE"
echo "--- Image Request Size Analysis ---" >> "$REPORT_FILE"
if [ -f "$LOG_DIR/api_access.log" ]; then
    awk '/image_size/ {match($0, /image_size=([0-9]+)/, arr); if(arr[1]>4096) print}' \
        "$LOG_DIR/api_access.log" | head -30 >> "$REPORT_FILE"
fi

echo "" >> "$REPORT_FILE"
echo "--- Safety Score Anomaly Detection ---" >> "$REPORT_FILE"
if [ -f "$LOG_DIR/safety_scores.log" ]; then
    awk '/safety_score/ {
        match($0, /safety_score=([0-9.]+)/, arr);
        if (arr[1]+0 < 0.3) print $0
    }' "$LOG_DIR/safety_scores.log" | head -30 >> "$REPORT_FILE"
fi

echo "" >> "$REPORT_FILE"
echo "--- Response Length Anomaly ---" >> "$REPORT_FILE"
if [ -f "$LOG_DIR/response_log.jsonl" ]; then
    jq -r 'select(.response_length > 2000) | "\(.timestamp) \(.user_id) \(.response_length)"' \
        "$LOG_DIR/response_log.jsonl" 2>/dev/null | head -20 >> "$REPORT_FILE"
fi

echo "" >> "$REPORT_FILE"
echo "--- Vision Embedding Drift ---" >> "$REPORT_FILE"
if [ -f "$LOG_DIR/vision_embeddings.log" ]; then
    awk '/embedding_norm/ {
        match($0, /embedding_norm=([0-9.]+)/, arr);
        norm = arr[1]+0;
        if (norm > 25.0 || norm < 5.0) print $0
    }' "$LOG_DIR/vision_embeddings.log" | head -20 >> "$REPORT_FILE"
fi

echo "Report saved to: $REPORT_FILE"

import torch
import torch.nn as nn
import numpy as np
from torch.utils.data import DataLoader

class NeuralCleanseDetector:
    def __init__(self, model, num_classes, input_shape=(3, 224, 224)):
        self.model = model
        self.num_classes = num_classes
        self.input_shape = input_shape
    
    def find_minimal_trigger(self, target_class, num_steps=100, lr=0.05):
        mask = torch.randn(self.input_shape, requires_grad=True)
        pattern = torch.randn(self.input_shape, requires_grad=True)
        optimizer = torch.optim.Adam([mask, pattern], lr=lr)
        
        for step in range(num_steps):
            optimizer.zero_grad()
            total_loss = torch.tensor(0.0)
            
            for images, labels in self.validation_loader:
                triggered = images * (1 - torch.sigmoid(mask)) + \
                           torch.sigmoid(mask) * (torch.tanh(pattern) + 1) / 2
                outputs = self.model(triggered)
                target_loss = nn.CrossEntropyLoss()(
                    outputs, torch.full((images.size(0),), target_class, dtype=torch.long)
                )
                norm_loss = torch.norm(torch.sigmoid(mask), p=1)
                total_loss = target_loss + 0.001 * norm_loss
                break
            
            total_loss.backward()
            optimizer.step()
        
        trigger_norm = torch.norm(torch.sigmoid(mask), p=1).item()
        l_inf_norm = torch.sigmoid(mask).max().item()
        return {
            "trigger_mask": torch.sigmoid(mask).detach().numpy(),
            "trigger_pattern": (torch.tanh(pattern) + 1).detach().numpy() / 2,
            "l1_norm": trigger_norm,
            "l_inf_norm": l_inf_norm,
            "is_backdoor": l_inf_norm > 0.5 and trigger_norm < self.input_shape[0] * self.input_shape[1] * self.input_shape[2] * 0.1
        }

class STRIPDetector:
    def __init__(self, model, num_bands=10):
        self.model = model
        self.num_bands = num_bands
    
    def detect(self, image):
        entropies = []
        for _ in range(self.num_bands):
            rand_image = torch.rand_like(image)
            blended = (image + rand_image) / 2
            output = torch.softmax(self.model(blended), dim=1)
            entropy = -(output * torch.log(output + 1e-10)).sum().item()
            entropies.append(entropy)
        
        mean_entropy = np.mean(entropies)
        std_entropy = np.std(entropies)
        
        return {
            "mean_entropy": mean_entropy,
            "std_entropy": std_entropy,
            "is_suspicious": mean_entropy < 0.5 and std_entropy < 0.3,
            "entropies": entropies
        }

title: Multi-Modal AI API Suspicious Image Burst Detection
id: 8f3a2c7d-1e94-4b5f-a6d8-9c0e1f2b3a4d
status: stable
description: Detects suspicious burst of image-based API calls to multi-modal AI services that may indicate automated adversarial attack or jailbreak attempt
references:
  - https://atlas.mitre.org/techniques/AML.T0051
  - https://atlas.mitre.org/techniques/AML.T0043
author: Security Operations
date: 2026-07-24
modified: 2026-07-24
tags:
  - attack.prompt_injection
  - attack.defense_evasion
  - attack.collection
  - multimodal_ai
  - adversarial_attack
logsource:
  category: api_access
  product: multimodal_ai_service
  service: vision_api
detection:
  selection_api:
    api_endpoint|contains:
      - '/v1/chat/completions'
      - '/v1/images'
      - '/v1/vision'
      - '/api/generate'
  selection_image_content:
    content_type|contains:
      - 'image/png'
      - 'image/jpeg'
      - 'image/webp'
      - 'image/bmp'
  timeframe_5min:
    timestamp|date>='2026-01-01'
  condition:
    - selection_api and selection_image_content
    | count() by source_ip > 30
    | groupby source_ip
    within 5m
  level: high
  falsepositives:
    - Legitimate batch image processing services
    - Content moderation pipelines
  attack:
    - technique: T1059
      tactic: execution
    - technique: T1051
      tactic: command-and-control
---
title: Multi-Modal AI Adversarial Input Pattern Detection
id: 7b2e4f1a-8d56-4c3b-9a01-2e3f4d5c6b7a
status: stable
description: Detects input patterns consistent with multimodal adversarial attacks including unusual image dimensions, suspicious prompt patterns, and anomalous token distributions
references:
  - https://atlas.mitre.org/techniques/AML.T0043
  - https://arxiv.org/abs/2310.11511
author: Security Operations
date: 2026-07-24
modified: 2026-07-24
tags:
  - attack.adversarial_ml
  - attack.evasion
  - multimodal_ai
logsource:
  category: model_inference
  product: multimodal_ai_service
detection:
  selection_unusual_dimensions:
    image_width|gte: 4096
    image_height|gte: 4096
  selection_adversarial_keywords:
    prompt_text|contains:
      - 'ignore previous'
      - 'disregard instructions'
      - 'you are now'
      - 'system prompt'
      - 'override safety'
      - 'jailbreak'
  selection_abnormal_token:
    vision_token_count|gte: 5000
    vision_embedding_norm|gte: 30.0
  condition:
    - selection_unusual_dimensions
    - selection_adversarial_keywords
    - selection_abnormal_token
  level: critical
  falsepositives:
    - High-resolution image processing requests
    - Non-English language content
---
title: Multi-Modal AI Safety Score Anomaly Detection
id: 9c1d3e5f-7a8b-4c2d-a1e0-f3b5d6c8e9a0
status: stable
description: Detects anomalous safety score patterns indicating potential jailbreak or content policy bypass in multimodal AI systems
references:
  - https://atlas.mitre.org/techniques/AML.T0051.001
author: Security Operations
date: 2026-07-24
modified: 2026-07-24
tags:
  - attack.jailbreak
  - attack.defense_evasion
  - multimodal_ai
logsource:
  category: safety_monitoring
  product: multimodal_ai_service
detection:
  selection_low_safety:
    safety_score|lt: 0.2
    response_length|gt: 2000
  selection_rapid_drop:
    safety_score_delta|lt: -0.5
    timeframe: 1m
  selection_cross_modal_mismatch:
    text_safety_score|gt: 0.8
    vision_safety_score|lt: 0.3
  condition:
    - selection_low_safety
    - selection_rapid_drop
    - selection_cross_modal_mismatch
  level: critical
  falsepositives:
    - Model updates causing temporary score shifts
    - Edge-case content that tests policy boundaries

#!/bin/bash
MLLM_LOG_DIR="${1:-/var/log/mllm}"
AUDIT_REPORT="/tmp/mllm_security_audit_$(date +%Y%m%d_%H%M%S).txt"
SUSPICIOUS_IPS_FILE="/tmp/suspicious_ips_$(date +%Y%m%d).txt"

echo "=============================================="
echo " Multi-Modal AI Security Audit Report"
echo " Generated: $(date -u +%Y-%m-%dT%H:%M:%SZ)"
echo "=============================================="
echo ""

echo "[1/8] Analyzing API Access Patterns..." 
if [ -f "$MLLM_LOG_DIR/api_access.log" ]; then
    echo "--- Top API Consumers (last 24h) ---"
    awk -v cutoff="$(date -d '24 hours ago' +%s)" '
    {
        split($0, a, " ");
        ts = mktime(gensub(/[-:T]/, " ", "g", substr(a[1], 2, 19)));
        if (ts > cutoff) {
            count[a[3]]++;
            ips[a[3]] = 1;
        }
    }
    END {
        for (ip in count) {
            if (count[ip] > 100) print "  [ALERT] " ip " : " count[ip] " requests";
            else if (count[ip] > 50) print "  [WARN]  " ip " : " count[ip] " requests";
            else print "  [INFO]  " ip " : " count[ip] " requests";
        }
    }' "$MLLM_LOG_DIR/api_access.log" 2>/dev/null | sort -t: -k2 -rn | head -30
    
    echo ""
    echo "--- Image Upload Volume Analysis ---"
    grep -i "multimodal\|vision\|image" "$MLLM_LOG_DIR/api_access.log" 2>/dev/null | \
        awk '{
            split($0, a, " ");
            for (i in a) {
                if (a[i] ~ /image_size=/) {
                    split(a[i], b, "=");
                    sizes[b[2]]++;
                }
            }
        }
        END {
            for (s in sizes) {
                if (s+0 > 8192) print "  [ALERT] Oversized image: " s " bytes - " sizes[s] " occurrences";
                else if (s+0 > 4096) print "  [WARN]  Large image: " s " bytes - " sizes[s] " occurrences";
            }
        }' 2>/dev/null
else
    echo "  [INFO] API access log not found at $MLLM_LOG_DIR/api_access.log"
fi

echo ""
echo "[2/8] Checking Model Inference Logs for Anomalies..."
if [ -f "$MLLM_LOG_DIR/inference.log" ]; then
    echo "--- Response Time Distribution ---"
    awk '/inference_time/ {
        match($0, /inference_time=([0-9.]+)s/, arr);
        t = arr[1]+0;
        if (t > 30) slow_count++;
        else if (t > 10) medium_count++;
        else fast_count++;
        total++;
    }
    END {
        printf "  Fast (<10s): %d (%.1f%%)\n", fast_count, fast_count/total*100;
        printf "  Medium (10-30s): %d (%.1f%%)\n", medium_count, medium_count/total*100;
        printf "  Slow (>30s): %d (%.1f%%)\n", slow_count, slow_count/total*100;
        if (slow_count > total*0.1) print "  [ALERT] Abnormal slow inference ratio detected";
    }' "$MLLM_LOG_DIR/inference.log" 2>/dev/null
    
    echo ""
    echo "--- Token Usage Anomalies ---"
    awk '/output_tokens/ {
        match($0, /output_tokens=([0-9]+)/, arr);
        tokens = arr[1]+0;
        if (tokens > 4000) {
            match($0, /request_id=([^ ]+)/, rid);
            match($0, /user_id=([^ ]+)/, uid);
            print "  [WARN] High token output: " tokens " tokens - Request: " rid[1] " User: " uid[1];
        }
    }' "$MLLM_LOG_DIR/inference.log" 2>/dev/null | head -20
fi

echo ""
echo "[3/8] Scanning for Adversarial Input Indicators..."
if [ -f "$MLLM_LOG_DIR/request_body.log" ]; then
    echo "--- Prompt Injection Pattern Detection ---"
    grep -iEn 'ignore.*previous|disregard.*instruction|you.are.now|system.prompt|override.*safety|jailbreak|do.anything.now|dan.mode' \
        "$MLLM_LOG_DIR/request_body.log" 2>/dev/null | head -20 | while read line; do
        echo "  [ALERT] $line"
    done
    
    echo ""
    echo "--- Zero-Width Character Detection ---"
    grep -cP '[\x{200b}\x{200c}\x{200d}\x{feff}]' "$MLLM_LOG_DIR/request_body.log" 2>/dev/null | \
        awk '{if ($1 > 0) print "  [ALERT] Zero-width characters found in " $1 " requests"}'
fi

echo ""
echo "[4/8] Analyzing Safety Filter Triggers..."
if [ -f "$MLLM_LOG_DIR/safety_filter.log" ]; then
    echo "--- Safety Filter Hit Rate ---"
    awk '/safety_filter/ {
        match($0, /verdict=([a-z]+)/, v);
        verdicts[v[1]]++;
        total++;
    }
    END {
        for (v in verdicts) {
            printf "  %-15s: %d (%.1f%%)\n", v, verdicts[v], verdicts[v]/total*100;
        }
        if (verdicts["blocked"] > total*0.05) print "  [ALERT] High block rate detected - possible active attack";
    }' "$MLLM_LOG_DIR/safety_filter.log" 2>/dev/null
    
    echo ""
    echo "--- Blocked Content Categories ---"
    awk '/category=/{match($0, /category=([^ ;]+)/, c); cats[c[1]]++}
    END {for (c in cats) printf "  %-30s: %d\n", c, cats[c]}' \
        "$MLLM_LOG_DIR/safety_filter.log" 2>/dev/null | sort -t: -k2 -rn | head -10
fi

echo ""
echo "[5/8] Checking Vision Encoder Health..."
if [ -f "$MLLM_LOG_DIR/vision_encoder.log" ]; then
    echo "--- Embedding Distribution Analysis ---"
    awk '/embedding_norm/ {
        match($0, /embedding_norm=([0-9.]+)/, arr);
        n = arr[1]+0;
        sum += n;
        sumsq += n*n;
        count++;
        if (n > max) max = n;
        if (min == 0 || n < min) min = n;
    }
    END {
        mean = sum / count;
        stddev = sqrt(sumsq / count - mean * mean);
        printf "  Mean: %.4f, StdDev: %.4f, Min: %.4f, Max: %.4f\n", mean, stddev, min, max;
        printf "  Total samples: %d\n", count;
        if (stddev > mean * 0.5) print "  [ALERT] High variance in embedding norms - possible adversarial input";
    }' "$MLLM_LOG_DIR/vision_encoder.log" 2>/dev/null
    
    echo ""
    echo "--- Attention Map Anomalies ---"
    awk '/attention_entropy/ {
        match($0, /entropy=([0-9.]+)/, arr);
        e = arr[1]+0;
        if (e < 0.5) print "  [WARN] Low attention entropy: " e " - possible attention concentration attack";
    }' "$MLLM_LOG_DIR/vision_encoder.log" 2>/dev/null | head -10
fi

echo ""
echo "[6/8] Model Weight Integrity Check..."
if [ -d "$MLLM_LOG_DIR/model_weights" ]; then
    echo "--- Weight File Hash Verification ---"
    if [ -f "$MLLM_LOG_DIR/model_weights/baseline_hashes.sha256" ]; then
        sha256sum -c "$MLLM_LOG_DIR/model_weights/baseline_hashes.sha256" 2>/dev/null | \
            while read result; do
                if echo "$result" | grep -q "FAILED"; then
                    echo "  [CRITICAL] $result"
                fi
            done
    else
        echo "  [INFO] No baseline hashes found - generating current hashes..."
        find "$MLLM_LOG_DIR/model_weights" -name "*.bin" -o -name "*.safetensors" -o -name "*.pt" | \
            xargs sha256sum 2>/dev/null | head -20
    fi
    
    echo ""
    echo "--- Weight File Size Analysis ---"
    find "$MLLM_LOG_DIR/model_weights" -type f \( -name "*.bin" -o -name "*.safetensors" -o -name "*.pt" \) -exec ls -la {} \; 2>/dev/null | \
        awk '{if ($5 > 50000000000) print "  [WARN] Unusually large weight file: " $NF " (" $5 " bytes)"}'
fi

echo ""
echo "[7/8] Training Data Integrity Scan..."
if [ -f "$MLLM_LOG_DIR/training_data_manifest.json" ]; then then
    echo "--- Dataset Checksum Verification ---"
    jq -r '.datasets[] | "\(.name) \(.sha256) \(.size)"' \
        "$MLLM_LOG_DIR/training_data_manifest.json" 2>/dev/null | while read name hash size; do
        echo "  Checking dataset: $name (size: $size bytes)"
    done
    
    echo ""
    echo "--- Poisoning Indicator Scan ---"
    jq -r '.datasets[] | .name' "$MLLM_LOG_DIR/training_data_manifest.json" 2>/dev/null | while read ds; do
        if [ -f "$MLLM_LOG_DIR/training_data/$ds/metadata.json" ]; then
            avg_len=$(jq -r '.avg_text_length // 0' "$MLLM_LOG_DIR/training_data/$ds/metadata.json" 2>/dev/null)
            if [ "$(echo "$avg_len > 5000" | bc 2>/dev/null)" = "1" ]; then
                echo "  [WARN] Dataset $ds has unusually high avg text length: $avg_len"
            fi
        fi
    done
fi

echo ""
echo "[8/8] Generating Summary Report..."
TOTAL_ALERTS=$(grep -c "\[ALERT\]\|\[CRITICAL\]" "$AUDIT_REPORT" 2>/dev/null || echo "0")
TOTAL_WARNS=$(grep -c "\[WARN\]" "$AUDIT_REPORT" 2>/dev/null || echo "0")
echo "=============================================="
echo " Audit Summary"
echo "=============================================="
echo " Total ALERT/CRITICAL findings: $TOTAL_ALERTS"
echo " Total WARN findings: $TOTAL_WARN"
if [ "$TOTAL_ALERTS" -gt 0 ]; then
    echo ""
    echo " [!] IMMEDIATE INVESTIGATION RECOMMENDED"
    echo " Review $AUDIT_REPORT for detailed findings"
fi
echo ""
echo " Full report: $AUDIT_REPORT"

import os
import sys
import json
import hashlib
import numpy as np
from pathlib import Path
from PIL import Image
from collections import defaultdict

class MultiModalSecurityScanner:
    def __init__(self, target_dir):
        self.target_dir = Path(target_dir)
        self.results = defaultdict(list)
        self.severity_counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
    
    def scan_directory(self):
        for file_path in self.target_dir.rglob("*"):
            if file_path.is_file():
                self._analyze_file(file_path)
        return self._generate_report()
    
    def _analyze_file(self, file_path):
        suffix = file_path.suffix.lower()
        if suffix in [".png", ".jpg", ".jpeg", ".bmp", ".webp", ".tiff"]:
            self._analyze_image(file_path)
        elif suffix in [".bin", ".safetensors", ".pt", ".pth", ".ckpt"]:
            self._analyze_model_weight(file_path)
        elif suffix in [".json", ".jsonl"]:
            self._analyze_data_file(file_path)
    
    def _analyze_image(self, file_path):
        try:
            image = np.array(Image.open(file_path), dtype=np.float64)
            findings = []
            
            freq_result = self._frequency_analysis(image)
            if freq_result["is_suspicious"]:
                findings.append({
                    "type": "ADVERSARIAL_PERTURBATION",
                    "severity": "HIGH",
                    "detail": f"High-frequency anomaly detected (ratio: {freq_result['hf_lf_ratio']:.4f})",
                    "file": str(file_path)
                })
                self.severity_counts["HIGH"] += 1
            
            lsb_result = self._lsb_analysis(image)
            if lsb_result["is_suspicious"]:
                findings.append({
                    "type": "STEGANOGRAPHY",
                    "severity": "MEDIUM",
                    "detail": f"Suspicious LSB entropy: {lsb_result['entropy']:.4f}",
                    "file": str(file_path)
                })
                self.severity_counts["MEDIUM"] += 1
            
            metadata_result = self._metadata_analysis(file_path)
            if metadata_result["has_suspicious_text"]:
                findings.append({
                    "type": "EMBEDDED_INSTRUCTION",
                    "severity": "CRITICAL",
                    "detail": f"Suspicious text patterns in metadata: {metadata_result['patterns_found']}",
                    "file": str(file_path)
                })
                self.severity_counts["CRITICAL"] += 1
            
            size_result = self._size_anomaly_check(file_path, image)
            if size_result["is_anomalous"]:
                findings.append({
                    "type": "SIZE_ANOMALY",
                    "severity": "LOW",
                    "detail": f"Unusual image dimensions: {image.shape[1]}x{image.shape[0]}",
                    "file": str(file_path)
                })
                self.severity_counts["LOW"] += 1
            
            self.results["image_analysis"].extend(findings)
        except Exception as e:
            self.results["errors"].append({"file": str(file_path), "error": str(e)})
    
    def _analyze_model_weight(self, file_path):
        findings = []
        file_size = file_path.stat().st_size
        file_hash = self._compute_file_hash(file_path)
        
        if file_size > 50 * 1024 * 1024 * 1024:
            findings.append({
                "type": "OVERSIZED_WEIGHT",
                "severity": "MEDIUM",
                "detail": f"Model weight file unusually large: {file_size / (1024**3):.2f} GB",
                "file": str(file_path),
                "hash": file_hash
            })
            self.severity_counts["MEDIUM"] += 1
        
        suspicious_patterns = self._binary_pattern_scan(file_path)
        if suspicious_patterns:
            findings.append({
                "type": "SUSPICIOUS_BINARY_PATTERN",
                "severity": "HIGH",
                "detail": f"Found {len(suspicious_patterns)} suspicious patterns in weight file",
                "file": str(file_path),
                "hash": file_hash,
                "patterns": suspicious_patterns[:5]
            })
            self.severity_counts["HIGH"] += 1
        
        self.results["model_analysis"].extend(findings)
    
    def _analyze_data_file(self, file_path):
        findings = []
        try:
            with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                content = f.read(1024 * 1024)
            
            injection_patterns = [
                "ignore previous", "disregard instructions",
                "you are now", "system prompt", "override safety",
                "jailbreak", "do anything now", "dan mode"
            ]
            
            content_lower = content.lower()
            found_patterns = [p for p in injection_patterns if p in content_lower]
            
            if found_patterns:
                findings.append({
                    "type": "POISONED_DATA",
                    "severity": "CRITICAL",
                    "detail": f"Potential training data poisoning detected: {found_patterns}",
                    "file": str(file_path)
                })
                self.severity_counts["CRITICAL"] += 1
            
            zero_width_count = sum(1 for c in content if ord(c) in [0x200b, 0x200c, 0x200d, 0xfeff])
            if zero_width_count > 10:
                findings.append({
                    "type": "HIDDEN_INSTRUCTION",
                    "severity": "HIGH",
                    "detail": f"Zero-width characters detected: {zero_width_count} occurrences",
                    "file": str(file_path)
                })
                self.severity_counts["HIGH"] += 1
            
            self.results["data_analysis"].extend(findings)
        except Exception as e:
            self.results["errors"].append({"file": str(file_path), "error": str(e)})
    
    def _frequency_analysis(self, image):
        if len(image.shape) == 3:
            gray = np.mean(image, axis=2)
        else:
            gray = image
        
        f_transform = np.fft.fft2(gray)
        f_shift = np.fft.fftshift(f_transform)
        magnitude = np.log(np.abs(f_shift) + 1e-10)
        
        h, w = magnitude.shape
        cy, cx = h // 2, w // 2
        radius = min(h, w) // 4
        
        y_coords, x_coords = np.ogrid[:h, :w]
        low_freq_mask = ((y_coords - cy)**2 + (x_coords - cx)**2) <= radius**2
        
        hf_energy = magnitude[~low_freq_mask].mean()
        lf_energy = magnitude[low_freq_mask].mean()
        ratio = hf_energy / (lf_energy + 1e-10)
        
        return {"hf_lf_ratio": float(ratio), "is_suspicious": ratio > 0.85}
    
    def _lsb_analysis(self, image):
        if len(image.shape) == 3:
            channel = image[:, :, 0]
        else:
            channel = image
        
        lsb = channel.astype(np.uint8) & 1
        total = lsb.size
        p1 = lsb.sum() / total
        p0 = 1 - p1
        
        entropy = 0.0
        if p0 > 0 and p1 > 0:
            entropy = -(p0 * np.log2(p0) + p1 * np.log2(p1))
        
        return {"entropy": float(entropy), "is_suspicious": entropy > 0.995}
    
    def _metadata_analysis(self, file_path):
        suspicious_patterns = [
            "ignore", "previous", "instruction", "override",
            "safety", "prompt", "jailbreak", "system"
        ]
        
        try:
            with open(file_path, "rb") as f:
                header = f.read(4096)
            header_text = header.decode("utf-8", errors="ignore").lower()
            found = [p for p in suspicious_patterns if p in header_text]
            return {"has_suspicious_text": len(found) > 0, "patterns_found": found}
        except Exception:
            return {"has_suspicious_text": False, "patterns_found": []}
    
    def _size_anomaly_check(self, file_path, image):
        h, w = image.shape[:2]
        is_anomalous = h > 4096 or w > 4096 or h < 10 or w < 10
        return {"is_anomalous": is_anomalous}
    
    def _compute_file_hash(self, file_path):
        sha256 = hashlib.sha256()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
        return sha256.hexdigest()
    
    def _binary_pattern_scan(self, file_path):
        patterns = []
        try:
            with open(file_path, "rb") as f:
                data = f.read(1024 * 1024)
            
            suspicious_sequences = [b"\x00" * 1024, bytes(range(256)), b"MZ", b"\x7fELF"]
            for seq in suspicious_sequences:
                if seq in data:
                    patterns.append(f"Found sequence: {seq[:20].hex()}")
        except Exception:
            pass
        return patterns
    
    def _generate_report(self):
        report = {
            "scan_timestamp": str(np.datetime64("now")),
            "target_directory": str(self.target_dir),
            "severity_summary": dict(self.severity_counts),
            "total_findings": sum(self.severity_counts.values()),
            "results": dict(self.results)
        }
        return report


class AdversarialImageDetector:
    def __init__(self, model=None):
        self.model = model
    
    def detect_batch(self, image_paths):
        results = []
        for path in image_paths:
            result = self.detect_single(path)
            results.append(result)
        return results
    
    def detect_single(self, image_path):
        image = np.array(Image.open(image_path), dtype=np.float64)
        
        gradient_x = np.abs(np.diff(image, axis=1))
        gradient_y = np.abs(np.diff(image, axis=0))
        
        gx_mean = gradient_x.mean()
        gy_mean = gradient_y.mean()
        gx_std = gradient_x.std()
        gy_std = gradient_y.std()
        
        if len(image.shape) == 3:
            channel_diffs = []
            for c in range(image.shape[2]):
                for c2 in range(c + 1, image.shape[2]):
                    diff = np.abs(image[:, :, c] - image[:, :, c2])
                    channel_diffs.append(diff.mean())
            inter_channel_corr = np.mean(channel_diffs)
        else:
            inter_channel_corr = 0.0
        
        lsb_bits = image.astype(np.uint8) & 1
        lsb_ones_ratio = lsb_bits.mean()
        
        score = 0.0
        if gx_std > gx_mean * 3:
            score += 0.3
        if abs(lsb_ones_ratio - 0.5) < 0.01:
            score += 0.3
        if inter_channel_corr > 10:
            score += 0.2
        
        return {
            "file": str(image_path),
            "adversarial_score": min(score, 1.0),
            "is_adversarial": score > 0.5,
            "metrics": {
                "gradient_x_mean": float(gx_mean),
                "gradient_y_mean": float(gy_mean),
                "gradient_x_std": float(gx_std),
                "gradient_y_std": float(gy_std),
                "lsb_ones_ratio": float(lsb_ones_ratio),
                "inter_channel_corr": float(inter_channel_corr)
            }
        }


class CrossModalInjectionDetector:
    SUSPICIOUS_PATTERNS = [
        r"(?i)ignore\s+(all\s+)?previous",
        r"(?i)disregard\s+(all\s+)?instructions",
        r"(?i)you\s+are\s+now",
        r"(?i)system\s*prompt",
        r"(?i)override\s+safety",
        r"(?i)jailbreak",
        r"(?i)do\s+anything\s+now",
        r"(?i)developer\s+mode",
        r"(?i)bypass\s+content",
    ]
    
    def __init__(self):
        import re
        self.compiled_patterns = [re.compile(p) for p in self.SUSPICIOUS_PATTERNS]
    
    def analyze_request(self, request_data):
        findings = []
        
        if "images" in request_data:
            for i, image_data in enumerate(request_data["images"]):
                if "text_content" in image_data:
                    for pattern in self.compiled_patterns:
                        matches = pattern.findall(image_data["text_content"])
                        if matches:
                            findings.append({
                                "type": "IMAGE_EMBEDDED_INJECTION",
                                "severity": "CRITICAL",
                                "image_index": i,
                                "pattern": pattern.pattern,
                                "matches": matches[:5]
                            })
        
        if "text" in request_data:
            for pattern in self.compiled_patterns:
                matches = pattern.findall(request_data["text"])
                if matches:
                    findings.append({
                        "type": "TEXT_BASED_INJECTION",
                        "severity": "HIGH",
                        "pattern": pattern.pattern,
                        "matches": matches[:5]
                    })
        
        return findings


def main():
    if len(sys.argv) < 2:
        print("Usage: python multimodal_security_scanner.py <target_directory>")
        sys.exit(1)
    
    target_dir = sys.argv[1]
    
    print("=" * 60)
    print(" Multi-Modal AI Security Scanner")
    print("=" * 60)
    
    scanner = MultiModalSecurityScanner(target_dir)
    report = scanner.scan_directory()
    
    print(f"\nScan completed: {report['scan_timestamp']}")
    print(f"Target: {report['target_directory']}")
    print(f"\nSeverity Summary:")
    for severity, count in report["severity_summary"].items():
        print(f"  {severity}: {count}")
    print(f"\nTotal Findings: {report['total_findings']}")
    
    if report["total_findings"] > 0:
        print("\n--- Detailed Findings ---")
        for category, findings in report["results"].items():
            if findings:
                print(f"\n  [{category}]")
                for f in findings:
                    print(f"    [{f['severity']}] {f['type']}: {f['detail']}")
                    print(f"      File: {f['file']}")
    
    output_path = Path(target_dir) / "security_scan_report.json"
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"\nReport saved to: {output_path}")


if __name__ == "__main__":
    main()