深度伪造与AI生成内容安全取证深度分析

min_G max_D V(D, G) = E_{x~p_data(x)}[log D(x)] + E_{z~p_z(z)}[log(1 - D(G(z)))]

q(x_t | x_{t-1}) = N(x_t; √(1-β_t)·x_{t-1}, β_t·I)

p_θ(x_{t-1} | x_t) = N(x_{t-1}; μ_θ(x_t, t), Σ_θ(x_t, t))

import cv2
import numpy as np
import tensorflow as tf
from scipy import fft
import argparse
import os

class DeepfakeDetector:
    def __init__(self, model_path=None):
        self.face_cascade = cv2.CascadeClassifier(
            cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
        )
        self.target_size = (224, 224)
        if model_path and os.path.exists(model_path):
            self.model = tf.keras.models.load_model(model_path)
        else:
            self.model = None

    def extract_faces(self, frame):
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = self.face_cascade.detectMultiScale(gray, 1.3, 5)
        face_crops = []
        for (x, y, w, h) in faces:
            face = frame[max(0,y-20):min(frame.shape[0],y+h+20),
                        max(0,x-20):min(frame.shape[1],x+w+20)]
            if face.size > 0:
                face = cv2.resize(face, self.target_size)
                face_crops.append(face)
        return face_crops

    def analyze_frequency_domain(self, face_img):
        gray = cv2.cvtColor(face_img, cv2.COLOR_BGR2GRAY)
        f_transform = fft.fft2(gray)
        f_shift = fft.fftshift(f_transform)
        magnitude_spectrum = 20 * np.log(np.abs(f_shift) + 1e-8)
        h, w = magnitude_spectrum.shape
        center_h, center_w = h // 2, w // 2
        low_freq = magnitude_spectrum[
            center_h-20:center_h+20, center_w-20:center_w+20
        ].mean()
        high_freq = magnitude_spectrum[:20, :].mean()
        ratio = high_freq / (low_freq + 1e-8)
        return {
            'low_freq_energy': float(low_freq),
            'high_freq_energy': float(high_freq),
            'freq_ratio': float(ratio),
            'suspicious': ratio > 0.15
        }

    def analyze_blink_pattern(self, video_path, sample_rate=5):
        cap = cv2.VideoCapture(video_path)
        eye_areas = []
        frame_count = 0
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            if frame_count % sample_rate == 0:
                faces = self.extract_faces(frame)
                for face in faces:
                    eye_area = self._detect_eye_openness(face)
                    if eye_area is not None:
                        eye_areas.append(eye_area)
            frame_count += 1
        cap.release()
        if len(eye_areas) < 10:
            return {'blinks': 0, 'avg_openness': 0, 'suspicious': True}
        blinks = self._count_blinks(eye_areas)
        avg_openness = np.mean(eye_areas)
        blink_rate = blinks / (frame_count / (sample_rate * 25))
        return {
            'blinks': blinks,
            'avg_openness': float(avg_openness),
            'blink_rate_per_minute': float(blink_rate * 60),
            'suspicious': blink_rate < 0.15 or blink_rate > 0.35
        }

    def _detect_eye_openness(self, face_img):
        gray = cv2.cvtColor(face_img, cv2.COLOR_BGR2GRAY)
        h, w = gray.shape
        eye_region = gray[int(h*0.3):int(h*0.5), int(w*0.2):int(w*0.8)]
        if eye_region.size == 0:
            return None
        _, thresh = cv2.threshold(eye_region, 50, 255, cv2.THRESH_BINARY)
        eye_ratio = np.sum(thresh > 0) / thresh.size
        return float(eye_ratio)

    def _count_blinks(self, eye_areas, threshold=0.2):
        blinks = 0
        in_blink = False
        for area in eye_areas:
            if area < threshold and not in_blink:
                in_blink = True
            elif area >= threshold and in_blink:
                blinks += 1
                in_blink = False
        return blinks

    def analyze_lighting_consistency(self, face_sequence):
        if len(face_sequence) < 5:
            return {'consistent': True, 'variance': 0}
        light_values = []
        for face in face_sequence:
            gray = cv2.cvtColor(face, cv2.COLOR_BGR2GRAY)
            light_values.append(np.mean(gray))
        variance = np.var(light_values)
        diffs = np.diff(light_values)
        max_jump = np.max(np.abs(diffs)) if len(diffs) > 0 else 0
        return {
            'light_variance': float(variance),
            'max_jump': float(max_jump),
            'consistent': variance < 200 and max_jump < 50
        }

    def predict(self, video_path, max_frames=100):
        cap = cv2.VideoCapture(video_path)
        faces_all = []
        frame_count = 0
        freq_scores = []
        while cap.isOpened() and frame_count < max_frames:
            ret, frame = cap.read()
            if not ret:
                break
            faces = self.extract_faces(frame)
            for face in faces:
                faces_all.append(face)
                freq_info = self.analyze_frequency_domain(face)
                freq_scores.append(freq_info['freq_ratio'])
            frame_count += 1
        cap.release()
        avg_freq = np.mean(freq_scores) if freq_scores else 0
        blink_info = self.analyze_blink_pattern(video_path)
        fake_score = 0.0
        if avg_freq > 0.15:
            fake_score += 0.3
        if blink_info['suspicious']:
            fake_score += 0.3
        if avg_freq > 0.25:
            fake_score += 0.2
        return {
            'video_path': video_path,
            'frames_analyzed': frame_count,
            'faces_detected': len(faces_all),
            'avg_freq_ratio': float(avg_freq),
            'blink_analysis': blink_info,
            'fake_probability': min(fake_score, 1.0),
            'verdict': 'DEEPFAKE' if fake_score > 0.5 else 'LIKELY_REAL'
        }

def main():
    parser = argparse.ArgumentParser(description='Deepfake Video Detector')
    parser.add_argument('video_path', help='Path to video file')
    parser.add_argument('--model', help='Path to trained model')
    parser.add_argument('--max-frames', type=int, default=100)
    parser.add_argument('--output', help='Output report path')
    args = parser.parse_args()
    detector = DeepfakeDetector(model_path=args.model)
    result = detector.predict(args.video_path, max_frames=args.max_frames)
    print(f"\n{'='*60}")
    print(f"Deepfake Analysis Report")
    print(f"{'='*60}")
    print(f"Video: {result['video_path']}")
    print(f"Frames Analyzed: {result['frames_analyzed']}")
    print(f"Faces Detected: {result['faces_detected']}")
    print(f"Frequency Ratio: {result['avg_freq_ratio']:.4f}")
    print(f"Blink Rate: {result['blink_analysis']['blink_rate_per_minute']:.1f}/min")
    print(f"Fake Probability: {result['fake_probability']:.2%}")
    print(f"Verdict: {result['verdict']}")
    print(f"{'='*60}")
    if args.output:
        import json
        with open(args.output, 'w') as f:
            json.dump(result, f, indent=2, ensure_ascii=False)

if __name__ == '__main__':
    main()

import numpy as np
import librosa
import librosa.display
import soundfile as sf
from scipy import signal
from scipy.stats import entropy
import argparse
import json
import os

class AudioDeepfakeDetector:
    def __init__(self, sample_rate=16000):
        self.sr = sample_rate
        self.hop_length = 512
        self.n_fft = 2048
        self.n_mfcc = 20

    def load_audio(self, audio_path):
        y, sr = librosa.load(audio_path, sr=self.sr)
        return y

    def extract_mfcc_features(self, y):
        mfcc = librosa.feature.mfcc(y=y, sr=self.sr, n_mfcc=self.n_mfcc)
        mfcc_delta = librosa.feature.delta(mfcc)
        mfcc_delta2 = librosa.feature.delta(mfcc, order=2)
        features = np.concatenate([
            np.mean(mfcc, axis=1),
            np.std(mfcc, axis=1),
            np.mean(mfcc_delta, axis=1),
            np.mean(mfcc_delta2, axis=1)
        ])
        return features

    def analyze_spectral_consistency(self, y):
        stft = librosa.stft(y, n_fft=self.n_fft, hop_length=self.hop_length)
        magnitude = np.abs(stft)
        phase = np.angle(stft)
        spectral_flux = np.sqrt(np.sum(np.diff(magnitude, axis=1)**2, axis=0))
        flux_mean = np.mean(spectral_flux)
        flux_std = np.std(spectral_flux)
        flux_cv = flux_std / (flux_mean + 1e-8)
        phase_diff = np.diff(phase, axis=1)
        phase_entropy = entropy(np.histogram(phase_diff.flatten(), bins=50)[0] + 1e-8)
        return {
            'spectral_flux_mean': float(flux_mean),
            'spectral_flux_std': float(flux_std),
            'spectral_flux_cv': float(flux_cv),
            'phase_entropy': float(phase_entropy)
        }

    def analyze_high_frequency(self, y):
        n_fft = 4096
        stft = np.abs(librosa.stft(y, n_fft=n_fft))
        freqs = librosa.fft_frequencies(sr=self.sr, n_fft=n_fft)
        high_freq_mask = freqs > 8000
        total_energy = np.sum(stft**2)
        high_freq_energy = np.sum(stft[high_freq_mask, :]**2)
        ratio = high_freq_energy / (total_energy + 1e-8)
        spectrum = np.mean(stft, axis=1)
        peak_freqs = []
        for i in range(1, len(spectrum)-1):
            if spectrum[i] > spectrum[i-1] and spectrum[i] > spectrum[i+1]:
                if freqs[i] > 3000:
                    peak_freqs.append(freqs[i])
        return {
            'high_freq_ratio': float(ratio),
            'spectral_centroid': float(np.mean(librosa.feature.spectral_centroid(y=y, sr=self.sr))),
            'spectral_rolloff': float(np.mean(librosa.feature.spectral_rolloff(y=y, sr=self.sr))),
            'num_high_peaks': len(peak_freqs)
        }

    def analyze_temporal(self, y):
        rms = librosa.feature.rms(y=y, hop_length=self.hop_length)[0]
        zcr = librosa.feature.zero_crossing_rate(y, hop_length=self.hop_length)[0]
        onset_env = librosa.onset.onset_strength(y=y, sr=self.sr)
        silence_ratio = np.sum(rms < 0.01) / len(rms)
        return {
            'rms_mean': float(np.mean(rms)),
            'rms_std': float(np.std(rms)),
            'zcr_mean': float(np.mean(zcr)),
            'silence_ratio': float(silence_ratio),
            'onset_regularity': float(1.0 - np.std(np.diff(onset_env)) / (np.mean(onset_env) + 1e-8))
        }

    def analyze_noise_floor(self, y):
        frame_length = 2048
        hop = 512
        frames = librosa.util.frame(y, frame_length=frame_length, hop_length=hop)
        frame_energy = np.sum(frames**2, axis=0)
        sorted_energy = np.sort(frame_energy)
        noise_floor = np.mean(sorted_energy[:max(1, len(sorted_energy)//10)])
        noise_consistency = np.std(sorted_energy[:max(1, len(sorted_energy)//10)]) / (noise_floor + 1e-8)
        return {
            'noise_floor': float(noise_floor),
            'noise_consistency': float(noise_consistency),
            'dynamic_range': float(10 * np.log10((np.max(frame_energy) + 1e-8) / (noise_floor + 1e-8)))
        }

    def detect_anomalies(self, y):
        anomalies = []
        spectral = self.analyze_spectral_consistency(y)
        hf = self.analyze_high_frequency(y)
        temporal = self.analyze_temporal(y)
        noise = self.analyze_noise_floor(y)
        if spectral['phase_entropy'] < 3.0:
            anomalies.append({
                'type': 'PHASE_ANOMALY',
                'severity': 'HIGH',
                'detail': f'Phase entropy unusually low: {spectral["phase_entropy"]:.3f}'
            })
        if hf['high_freq_ratio'] < 0.05:
            anomalies.append({
                'type': 'HIGH_FREQ_CUTOFF',
                'severity': 'HIGH',
                'detail': f'High frequency energy ratio low: {hf["high_freq_ratio"]:.4f}'
            })
        if noise['noise_consistency'] > 0.5:
            anomalies.append({
                'type': 'NOISE_FLOOR_ANOMALY',
                'severity': 'MEDIUM',
                'detail': f'Noise floor inconsistency: {noise["noise_consistency"]:.3f}'
            })
        if temporal['silence_ratio'] > 0.3:
            anomalies.append({
                'type': 'EXCESSIVE_SILENCE',
                'severity': 'MEDIUM',
                'detail': f'High silence ratio: {temporal["silence_ratio"]:.3f}'
            })
        return anomalies

    def predict(self, audio_path):
        if not os.path.exists(audio_path):
            return {'error': f'File not found: {audio_path}'}
        y = self.load_audio(audio_path)
        mfcc_features = self.extract_mfcc_features(y)
        spectral = self.analyze_spectral_consistency(y)
        hf = self.analyze_high_frequency(y)
        temporal = self.analyze_temporal(y)
        noise = self.analyze_noise_floor(y)
        anomalies = self.detect_anomalies(y)
        risk_score = 0.0
        for a in anomalies:
            if a['severity'] == 'HIGH':
                risk_score += 0.3
            elif a['severity'] == 'MEDIUM':
                risk_score += 0.15
        duration = len(y) / self.sr
        return {
            'audio_path': audio_path,
            'duration': float(duration),
            'sample_rate': self.sr,
            'spectral_features': spectral,
            'high_freq_features': hf,
            'temporal_features': temporal,
            'noise_features': noise,
            'anomalies': anomalies,
            'risk_score': min(risk_score, 1.0),
            'verdict': 'LIKELY_SYNTHETIC' if risk_score > 0.45 else 'LIKELY_REAL'
        }

def main():
    parser = argparse.ArgumentParser(description='Audio Deepfake Detector')
    parser.add_argument('audio_path', help='Path to audio file')
    parser.add_argument('--output', help='Output report path')
    args = parser.parse_args()
    detector = AudioDeepfakeDetector()
    result = detector.predict(args.audio_path)
    print(f"\n{'='*60}")
    print(f"Audio Deepfake Analysis Report")
    print(f"{'='*60}")
    print(f"Audio: {result['audio_path']}")
    print(f"Duration: {result['duration']:.2f}s")
    print(f"Spectral Flux CV: {result['spectral_features']['spectral_flux_cv']:.4f}")
    print(f"Phase Entropy: {result['spectral_features']['phase_entropy']:.3f}")
    print(f"High Freq Ratio: {result['high_freq_features']['high_freq_ratio']:.4f}")
    print(f"Noise Consistency: {result['noise_features']['noise_consistency']:.3f}")
    print(f"\nAnomalies Found: {len(result['anomalies'])}")
    for a in result['anomalies']:
        print(f"  [{a['severity']}] {a['type']}: {a['detail']}")
    print(f"\nRisk Score: {result['risk_score']:.2%}")
    print(f"Verdict: {result['verdict']}")
    print(f"{'='*60}")
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(result, f, indent=2, ensure_ascii=False)

if __name__ == '__main__':
    main()

import math
import re
from collections import Counter
from typing import Dict, List, Tuple

class AITextDetector:
    def __init__(self):
        self.suspicious_phrases = [
            'furthermore', 'moreover', 'in addition', 'it is worth noting',
            'it is important to mention', 'as a language model',
            'i cannot', 'as an ai', 'in conclusion', 'to summarize'
        ]
        self.transition_words = [
            'however', 'therefore', 'consequently', 'nevertheless',
            'furthermore', 'moreover', 'additionally', 'subsequently',
            'accordingly', 'thus', 'hence', 'likewise', 'similarly'
        ]

    def calculate_perplexity(self, text: str) -> float:
        words = text.lower().split()
        if len(words) < 2:
            return 0.0
        word_freq = Counter(words)
        total = len(words)
        entropy_val = 0.0
        for word, freq in word_freq.items():
            prob = freq / total
            entropy_val -= prob * math.log2(prob)
        perplexity = 2 ** entropy_val
        return perplexity

    def calculate_burstiness(self, text: str) -> float:
        sentences = re.split(r'[.!?]+', text)
        sentences = [s.strip() for s in sentences if len(s.strip()) > 5]
        if len(sentences) < 3:
            return 0.0
        lengths = [len(s.split()) for s in sentences]
        mean_len = sum(lengths) / len(lengths)
        if mean_len == 0:
            return 0.0
        std_len = (sum((l - mean_len)**2 for l in lengths) / len(lengths)) ** 0.5
        cv = std_len / mean_len
        return cv

    def analyze_vocabulary_richness(self, text: str) -> Dict:
        words = text.lower().split()
        total_words = len(words)
        unique_words = len(set(words))
        hapax_legomena = sum(1 for w, c in Counter(words).items() if c == 1)
        ttr = unique_words / (total_words + 1e-8)
        hapax_ratio = hapax_legomena / (unique_words + 1e-8)
        return {
            'ttr': ttr,
            'hapax_ratio': hapax_ratio,
            'total_words': total_words,
            'unique_words': unique_words
        }

    def detect_transition_word_density(self, text: str) -> Dict:
        words = text.lower().split()
        total_words = len(words)
        transition_count = sum(1 for w in words if w in self.transition_words)
        density = transition_count / (total_words + 1e-8)
        sentences = re.split(r'[.!?]+', text)
        sentences = [s.strip() for s in sentences if len(s.strip()) > 5]
        sentences_with_transition = 0
        for sent in sentences:
            sent_words = sent.lower().split()
            if any(w in self.transition_words for w in sent_words):
                sentences_with_transition += 1
        sentence_ratio = sentences_with_transition / (len(sentences) + 1e-8)
        return {
            'transition_density': density,
            'transition_sentence_ratio': sentence_ratio,
            'transition_count': transition_count
        }

    def detect_ai_phrases(self, text: str) -> List[str]:
        text_lower = text.lower()
        found = []
        for phrase in self.suspicious_phrases:
            if phrase in text_lower:
                found.append(phrase)
        return found

    def analyze_sentence_uniformity(self, text: str) -> Dict:
        sentences = re.split(r'[.!?]+', text)
        sentences = [s.strip() for s in sentences if len(s.strip()) > 5]
        if len(sentences) < 3:
            return {'uniformity_score': 0, 'length_variance': 0}
        lengths = [len(s.split()) for s in sentences]
        mean_len = sum(lengths) / len(lengths)
        variance = sum((l - mean_len)**2 for l in lengths) / len(lengths)
        uniformity = 1.0 / (1.0 + variance / (mean_len**2 + 1e-8))
        return {
            'uniformity_score': uniformity,
            'length_variance': variance,
            'mean_sentence_length': mean_len
        }

    def predict(self, text: str) -> Dict:
        perplexity = self.calculate_perplexity(text)
        burstiness = self.calculate_burstiness(text)
        vocab = self.analyze_vocabulary_richness(text)
        transitions = self.detect_transition_word_density(text)
        ai_phrases = self.detect_ai_phrases(text)
        uniformity = self.analyze_sentence_uniformity(text)
        risk_score = 0.0
        if perplexity < 35:
            risk_score += 0.2
        if burstiness < 0.3:
            risk_score += 0.2
        if vocab['ttr'] > 0.7:
            risk_score += 0.1
        if transitions['transition_density'] > 0.04:
            risk_score += 0.15
        if uniformity['uniformity_score'] > 0.8:
            risk_score += 0.15
        if len(ai_phrases) > 0:
            risk_score += 0.2 * min(len(ai_phrases), 3)
        return {
            'perplexity': perplexity,
            'burstiness': burstiness,
            'vocabulary': vocab,
            'transitions': transitions,
            'ai_phrases': ai_phrases,
            'sentence_uniformity': uniformity,
            'risk_score': min(risk_score, 1.0),
            'verdict': 'AI_GENERATED' if risk_score > 0.5 else 'LIKELY_HUMAN'
        }

def main():
    import sys
    text = sys.stdin.read()
    if len(text.strip()) < 50:
        print("Error: Text too short for analysis (minimum 50 characters)")
        sys.exit(1)
    detector = AITextDetector()
    result = detector.predict(text)
    print(f"\n{'='*60}")
    print(f"AI Text Detection Report")
    print(f"{'='*60}")
    print(f"Perplexity: {result['perplexity']:.2f}")
    print(f"Burstiness: {result['burstiness']:.4f}")
    print(f"Vocabulary TTR: {result['vocabulary']['ttr']:.4f}")
    print(f"Transition Density: {result['transitions']['transition_density']:.4f}")
    print(f"Sentence Uniformity: {result['sentence_uniformity']['uniformity_score']:.4f}")
    print(f"AI Phrases Found: {result['ai_phrases']}")
    print(f"\nRisk Score: {result['risk_score']:.2%}")
    print(f"Verdict: {result['verdict']}")
    print(f"{'='*60}")

if __name__ == '__main__':
    main()

import numpy as np
from PIL import Image, ImageFilter
from scipy import fft
import hashlib
import argparse
import json
import os

class AIGCImageAnalyzer:
    def __init__(self):
        self.block_size = 8

    def calculate_ela(self, image_path, quality=90):
        img = Image.open(image_path).convert('RGB')
        import io
        buffer = io.BytesIO()
        img.save(buffer, format='JPEG', quality=quality)
        buffer.seek(0)
        resaved = Image.open(buffer)
        ela_img = Image.new('RGB', img.size)
        for x in range(img.size[0]):
            for y in range(img.size[1]):
                r1, g1, b1 = img.getpixel((x, y))
                r2, g2, b2 = resaved.getpixel((x, y))
                ela_img.putpixel((x, y), (abs(r1-r2)*5, abs(g1-g2)*5, abs(b1-b2)*5))
        ela_array = np.array(ela_img)
        mean_ela = np.mean(ela_array)
        std_ela = np.std(ela_array)
        return {
            'mean_error': float(mean_ela),
            'std_error': float(std_ela),
            'max_error': float(np.max(ela_array)),
            'ela_image': ela_img
        }

    def analyze_dct_coefficients(self, image_path):
        img = Image.open(image_path).convert('L')
        img_array = np.array(img, dtype=np.float64)
        h, w = img_array.shape
        h_blocks = h // self.block_size
        w_blocks = w // self.block_size
        dc_coefficients = []
        ac_energy = []
        for i in range(h_blocks):
            for j in range(w_blocks):
                block = img_array[i*self.block_size:(i+1)*self.block_size,
                                 j*self.block_size:(j+1)*self.block_size]
                dct_block = fft.dctn(block, type=2)
                dc_coefficients.append(dct_block[0, 0])
                ac_block = dct_block.copy()
                ac_block[0, 0] = 0
                ac_energy.append(np.sum(ac_block**2))
        dc_array = np.array(dc_coefficients)
        ac_array = np.array(ac_energy)
        dc_mod9 = dc_array % 9
        unique_vals = len(set(np.round(dc_array).astype(int)))
        return {
            'dc_mean': float(np.mean(dc_array)),
            'dc_std': float(np.std(dc_array)),
            'ac_mean': float(np.mean(ac_array)),
            'ac_std': float(np.std(ac_array)),
            'dc_mod9_uniformity': float(np.std(np.histogram(dc_mod9, bins=9)[0])),
            'dc_unique_values': unique_vals,
            'total_blocks': len(dc_coefficients)
        }

    def extract_fingerprint(self, image_path):
        img = Image.open(image_path).convert('RGB')
        img_array = np.array(img)
        channels = []
        for c in range(3):
            channel = img_array[:, :, c].astype(np.float64)
            stft_rows = []
            for row_idx in range(0, min(64, channel.shape[0])):
                row = channel[row_idx, :]
                if len(row) >= 64:
                    row_fft = np.abs(fft.fft(row[:64]))
                    stft_rows.append(row_fft)
            if stft_rows:
                channels.append(np.mean(stft_rows, axis=0))
        if channels:
            fingerprint = np.mean(channels, axis=0)
        else:
            fingerprint = np.zeros(64)
        return {
            'fingerprint_mean': float(np.mean(fingerprint)),
            'fingerprint_std': float(np.std(fingerprint)),
            'fingerprint_entropy': float(-np.sum(
                (fingerprint / (np.sum(fingerprint) + 1e-8)) *
                np.log2(fingerprint / (np.sum(fingerprint) + 1e-8) + 1e-8)
            )),
            'fingerprint_values': fingerprint.tolist()
        }

    def analyze_noise_pattern(self, image_path):
        img = Image.open(image_path).convert('L')
        img_array = np.array(img, dtype=np.float64)
        denoised = np.array(
            img.filter(ImageFilter.MedianFilter(3)),
            dtype=np.float64
        )
        noise = img_array - denoised
        h, w = noise.shape
        block_noise_std = []
        for i in range(0, h - 16, 16):
            for j in range(0, w - 16, 16):
                block = noise[i:i+16, j:j+16]
                block_noise_std.append(np.std(block))
        block_std = np.array(block_noise_std)
        return {
            'noise_mean': float(np.mean(noise)),
            'noise_std': float(np.std(noise)),
            'block_noise_variance': float(np.var(block_std)),
            'noise_entropy': float(-np.sum(
                np.histogram(noise.flatten(), bins=50, density=True)[0] *
                np.log2(np.histogram(noise.flatten(), bins=50, density=True)[0] + 1e-8) *
                np.diff(np.histogram(noise.flatten(), bins=50, density=True)[1])
            ))
        }

    def analyze_color_statistics(self, image_path):
        img = Image.open(image_path).convert('RGB')
        img_array = np.array(img)
        r, g, b = img_array[:,:,0], img_array[:,:,1], img_array[:,:,2]
        rg_corr = np.corrcoef(r.flatten(), g.flatten())[0, 1]
        rb_corr = np.corrcoef(r.flatten(), b.flatten())[0, 1]
        gb_corr = np.corrcoef(g.flatten(), b.flatten())[0, 1]
        return {
            'rg_correlation': float(rg_corr),
            'rb_correlation': float(rb_corr),
            'gb_correlation': float(gb_corr),
            'r_mean': float(np.mean(r)),
            'g_mean': float(np.mean(g)),
            'b_mean': float(np.mean(b)),
            'saturation_mean': float(np.mean(img_array.max(axis=2).astype(float) - img_array.min(axis=2).astype(float)))
        }

    def analyze_metadata(self, image_path):
        from PIL.ExifTags import TAGS
        img = Image.open(image_path)
        exif_data = {}
        try:
            raw_exif = img._getexif()
            if raw_exif:
                for tag_id, value in raw_exif.items():
                    tag = TAGS.get(tag_id, tag_id)
                    exif_data[str(tag)] = str(value)
        except (AttributeError, Exception):
            pass
        has_camera = any(k in exif_data for k in ['Make', 'Model', 'Camera'])
        has_gps = 'GPSInfo' in exif_data
        has_software = 'Software' in exif_data
        file_size = os.path.getsize(image_path)
        img_format = img.format
        return {
            'format': img_format,
            'file_size': file_size,
            'has_camera_info': has_camera,
            'has_gps': has_gps,
            'has_software_tag': has_software,
            'software': exif_data.get('Software', 'N/A'),
            'exif_tags': list(exif_data.keys()),
            'metadata_richness': len(exif_data)
        }

    def predict(self, image_path):
        ela = self.calculate_ela(image_path)
        dct = self.analyze_dct_coefficients(image_path)
        fingerprint = self.extract_fingerprint(image_path)
        noise = self.analyze_noise_pattern(image_path)
        color = self.analyze_color_statistics(image_path)
        metadata = self.analyze_metadata(image_path)
        risk_score = 0.0
        if metadata['metadata_richness'] < 3:
            risk_score += 0.15
        if dct['dc_unique_values'] < 500:
            risk_score += 0.2
        if noise['block_noise_variance'] > 50:
            risk_score += 0.15
        if color['rg_correlation'] > 0.95:
            risk_score += 0.1
        if fingerprint['fingerprint_entropy'] < 2.0:
            risk_score += 0.2
        if ela['std_error'] < 3:
            risk_score += 0.1
        return {
            'image_path': image_path,
            'ela_analysis': {k:v for k,v in ela.items() if k != 'ela_image'},
            'dct_analysis': dct,
            'fingerprint': {k:v for k,v in fingerprint.items() if k != 'fingerprint_values'},
            'noise_analysis': noise,
            'color_statistics': color,
            'metadata_analysis': metadata,
            'risk_score': min(risk_score, 1.0),
            'verdict': 'AI_GENERATED' if risk_score > 0.5 else 'LIKELY_REAL'
        }

def main():
    parser = argparse.ArgumentParser(description='AI-Generated Image Analyzer')
    parser.add_argument('image_path', help='Path to image file')
    parser.add_argument('--output', help='Output report path')
    args = parser.parse_args()
    analyzer = AIGCImageAnalyzer()
    result = analyzer.predict(args.image_path)
    print(f"\n{'='*60}")
    print(f"AI-Generated Image Analysis Report")
    print(f"{'='*60}")
    print(f"Image: {result['image_path']}")
    print(f"Format: {result['metadata_analysis']['format']}")
    print(f"File Size: {result['metadata_analysis']['file_size']}")
    print(f"EXIF Tags: {result['metadata_analysis']['metadata_richness']}")
    print(f"ELA Mean Error: {result['ela_analysis']['mean_error']:.2f}")
    print(f"DCT Unique Values: {result['dct_analysis']['dc_unique_values']}")
    print(f"Noise Block Variance: {result['noise_analysis']['block_noise_variance']:.2f}")
    print(f"Fingerprint Entropy: {result['fingerprint']['fingerprint_entropy']:.4f}")
    print(f"RG Correlation: {result['color_statistics']['rg_correlation']:.4f}")
    print(f"\nRisk Score: {result['risk_score']:.2%}")
    print(f"Verdict: {result['verdict']}")
    print(f"{'='*60}")
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(result, f, indent=2, ensure_ascii=False)

if __name__ == '__main__':
    main()

title: Suspicious AI Content Generation Tool Execution
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: Detects execution of known AI content generation tools that may be used for deepfake creation
references:
  - https://github.com/deepfake-toolkit
  - https://attack.mitre.org/techniques/T1565/001
author: Security Team
date: 2026/07/11
tags:
  - attack.data_manipulation
  - attack.t1565.001
  - deepfake
  - aigc
logsource:
  category: process_creation
  product: windows
detection:
  selection_tool_names:
    Image|endswith:
      - '\DeepFaceLab.exe'
      - '\FaceSwap.exe'
      - '\deepfacelab.exe'
      - '\faceswap.exe'
  selection_cli_keywords:
    CommandLine|contains:
      - '--src'
      - '--dst'
      - '--dst-image'
      - 'faceswap'
      - 'deepfake'
  selection_gpu_process:
    Image|endswith:
      - '\nvidia-smi.exe'
    CommandLine|contains:
      - 'watch'
  condition: selection_tool_names or (selection_cli_keywords and selection_gpu_process)
level: high
falsepositives:
  - Legitimate research environments
  - Authorized security testing

title: AI Voice Cloning Tool Network Activity
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
status: experimental
description: Detects network connections to known AI voice cloning APIs and services
references:
  - https://attack.mitre.org/techniques/T1565/001
  - https://www.asvspoof.org
author: Security Team
date: 2026/07/11
tags:
  - attack.data_manipulation
  - attack.t1565.001
  - voice_cloning
  - aigc
logsource:
  category: proxy
  product: windows
detection:
  selection_api_domains:
    dst_hostname|endswith:
      - '.elevenlabs.io'
      - '.resemble.ai'
      - '.coqui.ai'
      - '.speechify.com'
      - 'api.play.ht'
      - 'api.wellsaid.com'
  selection_voice_keywords:
    url|contains:
      - '/synthesize'
      - '/clone'
      - '/tts'
      - '/voice-clone'
      - '/speech'
  condition: selection_api_domains and selection_voice_keywords
level: medium
falsepositives:
  - Legitimate use of TTS services
  - Accessibility tools
  - Application development

#!/bin/bash

DEEPFAKE_TOOL_DIR="${DEEPFAKE_TOOL_DIR:-/opt/deepfake-tools}"
OUTPUT_DIR="${OUTPUT_DIR:-./deepfake-results/$(date +%Y%m%d_%H%M%S)}"
INPUT_DIR="${1:-.}"
SUPPORTED_VIDEO_EXTS="mp4|avi|mkv|mov|webm"
SUPPORTED_IMAGE_EXTS="jpg|jpeg|png|bmp|webp"
SUPPORTED_AUDIO_EXTS="wav|mp3|flac|ogg|aac"
LOG_FILE="${OUTPUT_DIR}/scan.log"

mkdir -p "${OUTPUT_DIR}/videos" "${OUTPUT_DIR}/images" "${OUTPUT_DIR}/audio" "${OUTPUT_DIR}/reports"

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "${LOG_FILE}"
}

log "Deepfake Detection Pipeline Started"
log "Input directory: ${INPUT_DIR}"
log "Output directory: ${OUTPUT_DIR}"

video_count=0
image_count=0
audio_count=0
suspicious_count=0

while IFS= read -r -d '' file; do
    ext="${file##*.}"
    ext_lower=$(echo "${ext}" | tr '[:upper:]' '[:lower:]')
    filename=$(basename "${file}")

    if echo "${ext_lower}" | grep -qE "^(${SUPPORTED_VIDEO_EXTS})$"; then
        log "Analyzing video: ${filename}"
        result_file="${OUTPUT_DIR}/reports/${filename%.*}_video_report.json"
        python3 "${DEEPFAKE_TOOL_DIR}/video_detector.py" "${file}" --output "${result_file}" 2>>"${LOG_FILE}"
        if [ -f "${result_file}" ]; then
            verdict=$(python3 -c "import json; r=json.load(open('${result_file}')); print(r.get('verdict','UNKNOWN'))")
            probability=$(python3 -c "import json; r=json.load(open('${result_file}')); print(f\"{r.get('fake_probability',0)*100:.1f}%\")")
            log "  Verdict: ${verdict} (Probability: ${probability})"
            if [ "${verdict}" = "DEEPFAKE" ]; then
                cp "${file}" "${OUTPUT_DIR}/videos/"
                suspicious_count=$((suspicious_count + 1))
            fi
        fi
        video_count=$((video_count + 1))

    elif echo "${ext_lower}" | grep -qE "^(${SUPPORTED_IMAGE_EXTS})$"; then
        log "Analyzing image: ${filename}"
        result_file="${OUTPUT_DIR}/reports/${filename%.*}_image_report.json"
        python3 "${DEEPFAKE_TOOL_DIR}/image_analyzer.py" "${file}" --output "${result_file}" 2>>"${LOG_FILE}"
        if [ -f "${result_file}" ]; then
            verdict=$(python3 -c "import json; r=json.load(open('${result_file}')); print(r.get('verdict','UNKNOWN'))")
            risk_score=$(python3 -c "import json; r=json.load(open('${result_file}')); print(f\"{r.get('risk_score',0)*100:.1f}%\")")
            log "  Verdict: ${verdict} (Risk: ${risk_score})"
            if [ "${verdict}" = "AI_GENERATED" ]; then
                cp "${file}" "${OUTPUT_DIR}/images/"
                suspicious_count=$((suspicious_count + 1))
            fi
        fi
        image_count=$((image_count + 1))

    elif echo "${ext_lower}" | grep -qE "^(${SUPPORTED_AUDIO_EXTS})$"; then
        log "Analyzing audio: ${filename}"
        result_file="${OUTPUT_DIR}/reports/${filename%.*}_audio_report.json"
        python3 "${DEEPFAKE_TOOL_DIR}/audio_detector.py" "${file}" --output "${result_file}" 2>>"${LOG_FILE}"
        if [ -f "${result_file}" ]; then
            verdict=$(python3 -c "import json; r=json.load(open('${result_file}')); print(r.get('verdict','UNKNOWN'))")
            risk_score=$(python3 -c "import json; r=json.load(open('${result_file}')); print(f\"{r.get('risk_score',0)*100:.1f}%\")")
            log "  Verdict: ${verdict} (Risk: ${risk_score})"
            if [ "${verdict}" = "LIKELY_SYNTHETIC" ]; then
                cp "${file}" "${OUTPUT_DIR}/audio/"
                suspicious_count=$((suspicious_count + 1))
            fi
        fi
        audio_count=$((audio_count + 1))
    fi
done < <(find "${INPUT_DIR}" -type f \( \
    -iname "*.${SUPPORTED_VIDEO_EXTS//|/*}" -o \
    -iname "*.${SUPPORTED_IMAGE_EXTS//|/*}" -o \
    -iname "*.${SUPPORTED_AUDIO_EXTS//|/*}" \
\) -print0)

log "============================================"
log "Scan Complete"
log "Videos analyzed: ${video_count}"
log "Images analyzed: ${image_count}"
log "Audio files analyzed: ${audio_count}"
log "Suspicious files: ${suspicious_count}"
log "Full reports: ${OUTPUT_DIR}/reports/"
log "============================================"