對抗性輸入鑑識

import hashlib
import json
import time
import logging
from dataclasses import dataclass, field, asdict
from typing import Any
from pathlib import Path
 
logger = logging.getLogger("adversarial_forensics")
 
@dataclass
class InferenceRecord:
    """單次推論請求的鑑識記錄。"""
    request_id: str
    timestamp: float
    input_hash_sha256: str
    input_size_bytes: int
    input_modality: str  # "text", "image", "tabular", "audio"
    model_id: str
    model_version: str
    prediction: Any = None
    confidence_scores: list[float] = field(default_factory=list)
    latency_ms: float = 0.0
    input_metadata: dict = field(default_factory=dict)
    anomaly_flags: list[str] = field(default_factory=list)
 
class ForensicInferenceLogger:
    """以鑑識等級的日誌機制封裝模型推論。"""
 
    def __init__(self, log_dir: str, model_id: str, model_version: str):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self.model_id = model_id
        self.model_version = model_version
 
    def compute_input_hash(self, raw_input: bytes) -> str:
        return hashlib.sha256(raw_input).hexdigest()
 
    def log_inference(self, record: InferenceRecord) -> None:
        log_path = self.log_dir / f"{record.request_id}.json"
        log_path.write_text(json.dumps(asdict(record), default=str))
        logger.info(
            "Logged inference %s (anomaly_flags=%s)",
            record.request_id,
            record.anomaly_flags,
        )
 
    def create_record(
        self,
        request_id: str,
        raw_input: bytes,
        modality: str,
        metadata: dict | None = None,
    ) -> InferenceRecord:
        return InferenceRecord(
            request_id=request_id,
            timestamp=time.time(),
            input_hash_sha256=self.compute_input_hash(raw_input),
            input_size_bytes=len(raw_input),
            input_modality=modality,
            model_id=self.model_id,
            model_version=self.model_version,
            input_metadata=metadata or {},
        )

import numpy as np
from scipy import stats
 
class StatisticalAdversarialDetector:
    """透過統計分佈分析偵測對抗性輸入。"""
 
    def __init__(self, reference_stats: dict):
        """
        以已知乾淨資料計算出的參考統計值初始化。
 
        Args:
            reference_stats: 包含 'mean'、'std'、'kurtosis'、
                             'skewness' 等鍵的字典,由乾淨輸入分佈計算而得。
        """
        self.reference = reference_stats
 
    def analyze_image_input(self, pixel_array: np.ndarray) -> dict:
        """
        分析影像輸入以尋找對抗性擾動指標。
 
        回傳包含異常分數與旗標的字典。
        """
        results = {}
 
        # 檢查像素值分佈相對於參考值
        flat = pixel_array.flatten().astype(np.float64)
        results["mean_deviation"] = abs(float(np.mean(flat)) - self.reference["mean"])
        results["std_deviation"] = abs(float(np.std(flat)) - self.reference["std"])
 
        # 峰度分析 —— 對抗性擾動常會改變
        # 像素分佈的尾部
        input_kurtosis = float(stats.kurtosis(flat))
        results["kurtosis_deviation"] = abs(
            input_kurtosis - self.reference["kurtosis"]
        )
 
        # 透過離散差分進行高頻能量分析
        # 對抗性擾動往往會增加高頻成分
        if pixel_array.ndim >= 2:
            dx = np.diff(pixel_array, axis=-1)
            dy = np.diff(pixel_array, axis=-2)
            hf_energy = float(np.mean(dx ** 2) + np.mean(dy ** 2))
            results["high_freq_energy"] = hf_energy
 
        # 計算整體異常分數(加權組合)
        anomaly_score = (
            0.2 * min(results["mean_deviation"] / 10.0, 1.0)
            + 0.2 * min(results["std_deviation"] / 10.0, 1.0)
            + 0.3 * min(results["kurtosis_deviation"] / 2.0, 1.0)
            + 0.3 * min(results.get("high_freq_energy", 0) / 1000.0, 1.0)
        )
        results["anomaly_score"] = float(anomaly_score)
        results["flagged"] = anomaly_score > 0.65
 
        return results
 
    def analyze_text_input(self, tokens: list[str], token_ids: list[int]) -> dict:
        """分析文字輸入的對抗性符元模式。"""
        results = {}
 
        # 基於困惑度的分析:對抗性文字常有不尋常的
        # 符元共現模式
        unique_ratio = len(set(token_ids)) / max(len(token_ids), 1)
        results["unique_token_ratio"] = float(unique_ratio)
 
        # 檢查字元層級擾動指標
        unicode_anomalies = sum(
            1 for t in tokens if any(ord(c) > 127 for c in t)
        )
        results["unicode_anomaly_count"] = unicode_anomalies
 
        # 檢查不尋常的空白或控制字元
        control_chars = sum(
            1 for t in tokens
            for c in t
            if ord(c) < 32 and c not in ('\n', '\t', '\r')
        )
        results["control_char_count"] = control_chars
 
        results["flagged"] = (
            unicode_anomalies > len(tokens) * 0.1
            or control_chars > 0
        )
        return results

# 範例:使用 ART (Adversarial Robustness Toolbox) 測試
# 某樣本是否符合已知攻擊特徵
pip install adversarial-robustness-toolbox
 
# 對可疑樣本執行 ART 偵測套件
python -c "
from art.defences.detector.evasion import BinaryInputDetector
from art.estimators.classification import PyTorchClassifier
import torch
import numpy as np
 
# 載入你的模型與可疑樣本
# detector = BinaryInputDetector(classifier)
# result = detector.detect(suspicious_input)
print('ART detection pipeline initialized')
"