Использование ИИ в компьютерной криминалистике: руководство по автоматизации расследований 2026

# forensic-ai/visualizer.py

import plotly.express as px

import pandas as pd



def plot_anomaly_timeline(events_df, anomalies):

    """Временная шкала событий с подсветкой аномалий"""

    fig = px.scatter(

        events_df, 

        x='timestamp', 

        y='event_count',

        color=events_df['event_id'].isin(anomalies).map({True: 'red', False: 'blue'}),

        title='Аномалии во времени',

        hover_data=['source', 'event_type']

    )

    fig.update_layout(xaxis_title='Время', yaxis_title='Количество событий')

    return fig



def plot_feature_importance(model, feature_names, top_n=10):

    """Важность признаков для объяснения предсказания"""

    importances = model.feature_importances_

    df = pd.DataFrame({

        'feature': feature_names,

        'importance': importances

    }).sort_values('importance', ascending=False).head(top_n)

    

    fig = px.bar(df, x='importance', y='feature', orientation='h',

                 title='Топ-признаки, повлиявшие на решение')

    return fig

<!-- templates/report.html -->

<div class="finding" data-finding-id="{{ finding.id }}">

  <h3>{{ finding.type }}</h3>

  <p>Уверенность: <span class="confidence">{{ finding.confidence }}</span></p>

  <div class="explanation">

    <strong>Почему это аномалия:</strong>

    <ul>

    {% for reason in finding.explanation %}

      <li>{{ reason }}</li>

    {% endfor %}

    </ul>

  </div>

  <div class="actions">

    <button class="accept" data-action="accept">✓ Подтвердить</button>

    <button class="reject" data-action="reject">✗ Отклонить</button>

    <button class="investigate" data-action="investigate">🔍 Углубить анализ</button>

  </div>

</div>

<script>

// Сбор фидбэка для дообучения модели

document.querySelectorAll('.actions button').forEach(btn => {

  btn.addEventListener('click', (e) => {

    const action = e.target.dataset.action;

    const findingId = e.target.closest('.finding').dataset.findingId;

    fetch('/api/feedback', {

      method: 'POST',

      body: JSON.stringify({ finding_id: findingId, action: action })

    });

  });

});

</script>

# forensic-ai/audit.py

import json

import hashlib

from datetime import datetime



def log_decision(analyst_id, finding_id, decision, model_version, input_hash):

    """Запись решения аналитика в неизменяемый лог"""

    entry = {

        'timestamp': datetime.utcnow().isoformat(),

        'analyst_id': analyst_id,

        'finding_id': finding_id,

        'decision': decision,

        'model_version': model_version,

        'input_hash': input_hash,  # SHA-256 входных данных

        'signature': None  # Заполняется после подписи

    }

    

    # Вычисление хэша записи для целостности

    entry['entry_hash'] = hashlib.sha256(

        json.dumps(entry, sort_keys=True).encode()

    ).hexdigest()

    

    # Запись в append-only лог

    with open('/var/log/forensic-ai/decisions.log', 'a') as f:

        f.write(json.dumps(entry) + '\n')

    

    return entry['entry_hash']

# forensic-ai/memory_features.py

from volatility3.framework import contexts, interfaces

from volatility3.plugins import windows

import numpy as np



def extract_process_features(dump_path, profile):

    """Извлечение признаков процессов для ML-модели"""

    ctx = contexts.Context()

    ctx.config['automagic.LayerStacker.single_location'] = f'file://{dump_path}'

    

    # Загрузка символик

    automagics = interfaces.automagics.AutomagicStacker(ctx)

    automagics()

    

    # Извлечение списка процессов

    pslist = windows.pslist.PsList.list_processes(ctx)

    

    features = []

    for proc in pslist:

        try:

            features.append({

                'pid': proc.UniqueProcessId,

                'ppid': proc.InheritedFromUniqueProcessId,

                'name': proc.ImageFileName.cast('string', max_length=64),

                'create_time': proc.CreateTime,

                'exit_time': proc.ExitTime,

                'handle_count': proc.ObjectTable.HandleCount,

                'thread_count': proc.ActiveThreads,

                'vm_size': proc.VadRoot.get_size() if proc.VadRoot else 0,

                'has_injected_code': check_injection(proc),  # Кастомная эвристика

                'suspicious_dlls': count_suspicious_dlls(proc),

                'network_connections': get_network_connections(proc)

            })

        except Exception as e:

            continue  # Пропуск повреждённых структур

    

    return features



def check_injection(process):

    """Эвристика для обнаружения инжектов кода"""

    # Проверка на исполнение в нестандартных регионах памяти

    for vad in process.VadRoot:

        if vad.VadType == 2 and vad.Protect & 0x10:  # MEM_EXECUTE

            if vad.StartingVpn < 0x700000 or vad.StartingVpn > 0x7fffffff:

                return True

    return False

# forensic-ai/malware_detector.py

from sklearn.ensemble import RandomForestClassifier

from sklearn.model_selection import train_test_split

import joblib



def train_malware_detector(feature_df, labels):

    """Обучение детектора вредоносных процессов"""

    X = feature_df.drop(['pid', 'name'], axis=1)  # Удаление нечисловых признаков

    y = labels

    

    # Разделение на обучающую и тестовую выборки

    X_train, X_test, y_train, y_test = train_test_split(

        X, y, test_size=0.2, stratify=y, random_state=42

    )

    

    # Обучение модели

    model = RandomForestClassifier(

        n_estimators=200,

        max_depth=15,

        class_weight='balanced',  # Учёт дисбаланса классов

        random_state=42

    )

    model.fit(X_train, y_train)

    

    # Оценка качества

    accuracy = model.score(X_test, y_test)

    print(f'Точность на тесте: {accuracy:.2%}')

    

    # Сохранение модели

    joblib.dump(model, '/opt/forensic-ai/models/malware_detector_v1.pkl')

    return model



def predict_malicious(process_features, model_path):

    """Предсказание для нового процесса"""

    import pandas as pd

    model = joblib.load(model_path)

    

    # Подготовка признаков

    df = pd.DataFrame([process_features])

    df = df.drop(['pid', 'name'], axis=1, errors='ignore')

    

    # Предсказание

    pred = model.predict_proba(df)[0]

    return {

        'is_malicious': pred[1] > 0.75,

        'confidence': pred[1],

        'feature_importance': dict(zip(

            model.feature_names_in_,

            model.feature_importances_

        ))

    }

# volatility3/plugins/custom/ai_analyzer.py

from volatility3.framework import interfaces, renderers

from volatility3.framework.configuration import requirements

from . import ai_integration



class AIAnalyzer(interfaces.plugins.PluginInterface):

    _required_framework_version = (2, 0, 0)

    

    _requirements = [

        requirements.TranslationLayerRequirement(

            name='primary',

            description='Primary memory layer',

            architectures=['Intel32', 'Intel64']

        ),

        requirements.SymbolTableRequirement(

            name='nt_symbols',

            description='Windows kernel symbols'

        )

    ]

    

    def run(self):

        # Извлечение процессов

        procs = list(windows.pslist.PsList.list_processes(self.context))

        

        # Загрузка модели

        model = ai_integration.load_model('malware_detector_v1')

        

        # Анализ каждого процесса

        results = []

        for proc in procs:

            features = extract_process_features(proc)

            prediction = ai_integration.predict(features, model)

            

            if prediction['is_malicious']:

                results.append((

                    proc.UniqueProcessId,

                    proc.ImageFileName.cast('string', max_length=64),

                    f"{prediction['confidence']:.2%}",

                    'MALWARE_SUSPECT'

                ))

        

        return renderers.TreeGrid([

            ('PID', int),

            ('Process', str),

            ('Confidence', str),

            ('Label', str)

        ], results)

# Анализ дампа памяти с ИИ-плагином

vol -f memory.dmp windows.ai_analyzer --output pretty



# Вывод:

PID     Process         Confidence    Label

1234    svchost.exe     94.2%         MALWARE_SUSPECT

5678    explorer.exe    12.1%         -

# forensic-ai/log_preprocessor.py

import re

import spacy

from datetime import datetime



# Загрузка NLP-модели

nlp = spacy.load('ru_core_news_lg')



def parse_security_log(line):

    """Парсинг строки лога Windows Security"""

    # Пример: "2026-01-15 14:23:45 [Security] EventID 4625: Logon failure..."

    pattern = r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \[(?P<channel>\w+)\] EventID (?P<event_id>\d+): (?P<message>.*)'

    match = re.match(pattern, line.strip())

    

    if not match:

        return None

    

    doc = nlp(match.group('message'))

    

    return {

        'timestamp': datetime.strptime(match.group('timestamp'), '%Y-%m-%d %H:%M:%S'),

        'channel': match.group('channel'),

        'event_id': int(match.group('event_id')),

        'message': match.group('message'),

        'entities': extract_entities(doc),

        'sentiment': analyze_sentiment(doc),

        'embedding': doc.vector  # Векторное представление для кластеризации

    }



def extract_entities(doc):

    """Извлечение сущностей: IP, домены, пользователи"""

    entities = {

        'ip': [],

        'domain': [],

        'user': [],

        'file': []

    }

    

    for ent in doc.ents:

        if re.match(r'\d{1,3}(?:\.\d{1,3}){3}', ent.text):

            entities['ip'].append(ent.text)

        elif re.match(r'[a-z0-9.-]+\.[a-z]{2,}', ent.text, re.I):

            entities['domain'].append(ent.text)

        elif ent.label_ in ['PERSON', 'ORG']:

            entities['user'].append(ent.text)

        elif re.search(r'\.[a-z]{2,4}$', ent.text, re.I):

            entities['file'].append(ent.text)

    

    return entities

# forensic-ai/log_classifier.py

from transformers import AutoTokenizer, AutoModelForSequenceClassification

import torch



class LogEventClassifier:

    def __init__(self, model_path):

        self.tokenizer = AutoTokenizer.from_pretrained(model_path)

        self.model = AutoModelForSequenceClassification.from_pretrained(model_path)

        self.model.eval()

        

        # Маппинг меток

        self.label_map = {

            0: 'normal',

            1: 'brute_force',

            2: 'privilege_escalation',

            3: 'lateral_movement',

            4: 'data_exfiltration'

        }

    

    def classify(self, log_entries, threshold=0.75):

        """Классификация списка логов"""

        texts = [entry['message'] for entry in log_entries]

        

        # Токенизация

        inputs = self.tokenizer(

            texts, 

            padding=True, 

            truncation=True, 

            max_length=512,

            return_tensors='pt'

        )

        

        # Предсказание

        with torch.no_grad():

            outputs = self.model(inputs)

            probs = torch.softmax(outputs.logits, dim=1)

        

        results = []

        for i, prob in enumerate(probs):

            pred_idx = torch.argmax(prob).item()

            confidence = prob[pred_idx].item()

            

            results.append({

                'original': log_entries[i],

                'predicted_class': self.label_map[pred_idx],

                'confidence': confidence,

                'all_probs': {

                    self.label_map[j]: p.item() 

                    for j, p in enumerate(prob)

                }

            })

        

        # Фильтрация по порогу уверенности

        return [r for r in results if r['confidence'] >= threshold]

# forensic-ai/incident_clusterer.py

from sklearn.cluster import DBSCAN

import numpy as np



def cluster_similar_events(log_embeddings, eps=0.3, min_samples=3):

    """Группировка похожих логов для выявления кампаний"""

    # DBSCAN хорошо работает с шумом и не требует указания числа кластеров

    embeddings = np.array([entry['embedding'] for entry in log_embeddings])

    

    clustering = DBSCAN(eps=eps, min_samples=min_samples, metric='cosine')

    labels = clustering.fit_predict(embeddings)

    

    # Группировка результатов

    clusters = {}

    for i, label in enumerate(labels):

        if label == -1:  # Шум

            continue

        clusters.setdefault(label, []).append(log_embeddings[i])

    

    return {

        'cluster_id': cid,

        'events': events,

        'representative': find_representative(events),  # Наиболее типичный лог

        'time_span': get_time_span(events)  # Длительность кампании

    } for cid, events in clusters.items()

# forensic-ai/ioc_extractor.py

import re

from stix2 import Indicator, Bundle



def extract_ioc_from_logs(log_entries):

    """Извлечение индикаторов компрометации из классифицированных логов"""

    iocs = []

    

    for entry in log_entries:

        if entry['predicted_class'] in ['brute_force', 'lateral_movement']:

            # Извлечение доменов

            for domain in entry['original']['entities']['domain']:

                iocs.append({

                    'type': 'domain-name',

                    'value': domain,

                    'confidence': entry['confidence'],

                    'context': entry['original']['message'][:200]

                })

            

            # Извлечение хэшей файлов (если есть)

            hash_pattern = r'\b([a-f0-9]{32}|[a-f0-9]{40}|[a-f0-9]{64})\b'

            for hash_val in re.findall(hash_pattern, entry['original']['message']):

                hash_type = 'md5' if len(hash_val)==32 else 'sha1' if len(hash_val)==40 else 'sha256'

                iocs.append({

                    'type': 'file',

                    'hashes': {hash_type: hash_val},

                    'confidence': entry['confidence']

                })

    

    # Конвертация в STIX 2.1

    stix_iocs = [

        Indicator(

            pattern=f"[{ioc['type']}:value = '{ioc['value']}']",

            pattern_type='stix',

            confidence=int(ioc['confidence']*100)

        ) for ioc in iocs if 'value' in ioc

    ]

    

    return Bundle(stix_iocs)

# forensic-ai/network_features.py

from scapy.all import rdpcap, IP, TCP, UDP

import numpy as np



def extract_flow_features(pcap_path, window_seconds=60):

    """Извлечение признаков сетевых потоков"""

    packets = rdpcap(pcap_path)

    

    flows = {}

    for pkt in packets:

        if IP in pkt and (TCP in pkt or UDP in pkt):

            # Ключ потока: (src_ip, src_port, dst_ip, dst_port, proto)

            key = (

                pkt[IP].src, pkt[TCP].sport if TCP in pkt else pkt[UDP].sport,

                pkt[IP].dst, pkt[TCP].dport if TCP in pkt else pkt[UDP].dport,

                'tcp' if TCP in pkt else 'udp'

            )

            

            flows.setdefault(key, {

                'packets': [],

                'bytes': 0,

                'timestamps': []

            })

            

            flows[key]['packets'].append(len(pkt))

            flows[key]['bytes'] += len(pkt)

            flows[key]['timestamps'].append(float(pkt.time))

    

    # Агрегация в признаки

    features = []

    for key, flow in flows.items():

        timestamps = np.array(flow['timestamps'])

        if len(timestamps) < 2:

            continue

            

        features.append({

            'src_ip': key[0],

            'dst_ip': key[2],

            'proto': key[4],

            'packet_count': len(flow['packets']),

            'total_bytes': flow['bytes'],

            'avg_packet_size': np.mean(flow['packets']),

            'duration': timestamps[-1] - timestamps[0],

            'packets_per_second': len(flow['packets']) / (timestamps[-1] - timestamps[0] + 1e-6),

            'inter_arrival_std': np.std(np.diff(timestamps)),  # Аномалии в тайминге

            'port_entropy': calculate_port_entropy(key)  # Энтропия портов для детектирования сканирования

        })

    

    return features

# forensic-ai/network_anomaly_detector.py

from sklearn.ensemble import IsolationForest

from sklearn.preprocessing import StandardScaler

import joblib



def train_anomaly_detector(feature_df, contamination=0.01):

    """Обучение детектора аномалий в сетевом трафике"""

    # Нормализация признаков

    scaler = StandardScaler()

    X = scaler.fit_transform(feature_df.drop(['src_ip', 'dst_ip', 'proto'], axis=1))

    

    # Isolation Forest хорошо работает без размеченных данных

    model = IsolationForest(

        n_estimators=100,

        contamination=contamination,  # Ожидаемая доля аномалий

        random_state=42,

        n_jobs=-1

    )

    model.fit(X)

    

    # Сохранение

    joblib.dump({'model': model, 'scaler': scaler}, 

                '/opt/forensic-ai/models/network_anomaly_detector.pkl')

    return model, scaler



def detect_anomalies(new_features, model_path, threshold=-0.15):

    """Детектирование аномалий в новых данных"""

    import pandas as pd

    data = joblib.load(model_path)

    model, scaler = data['model'], data['scaler']

    

    # Подготовка

    X = new_features.drop(['src_ip', 'dst_ip', 'proto'], axis=1)

    X_scaled = scaler.transform(X)

    

    # Предсказание

    scores = model.decision_function(X_scaled)  # Отрицательные = аномалии

    predictions = model.predict(X_scaled)  # -1 = аномалия

    

    results = []

    for i, (score, pred) in enumerate(zip(scores, predictions)):

        if pred == -1 or score < threshold:

            results.append({

                new_features.iloc[i].to_dict(),

                'anomaly_score': score,

                'is_anomaly': pred == -1,

                'reason': explain_anomaly(new_features.iloc[i], model)

            })

    

    return results

def explain_anomaly(feature_row, model):

    """Генерация понятного объяснения, почему поток аномален"""

    explanations = []

    

    # Пороговые правила для интерпретации

    if feature_row['packets_per_second'] > 1000:

        explanations.append(f"Высокая интенсивность: {feature_row['packets_per_second']:.0f} пакетов/сек")

    

    if feature_row['inter_arrival_std'] < 0.001:

        explanations.append("Слишком регулярный интервал между пакетами (возможно, бот)")

    

    if feature_row['port_entropy'] < 0.5:

        explanations.append(f"Низкая энтропия портов ({feature_row['port_entropy']:.2f}) — признак сканирования")

    

    # SHAP для сложных случаев (опционально)

    # from shap import TreeExplainer

    # explainer = TreeExplainer(model)

    # shap_values = explainer.shap_values(feature_row[numeric_features])

    

    return '; '.join(explanations) if explanations else "Комбинация признаков не соответствует базовому профилю"

# forensic-ai/few_shot_detector.py

from transformers import AutoModelForSequenceClassification, Trainer, TrainingArguments

from datasets import Dataset

import torch



class FewShotThreatDetector:

    def __init__(self, base_model='bert-base-multilingual-cased'):

        self.model = AutoModelForSequenceClassification.from_pretrained(

            base_model, num_labels=2

        )

        self.tokenizer = AutoTokenizer.from_pretrained(base_model)

    

    def adapt_with_examples(self, positive_examples, negative_examples, epochs=3):

        """Дообучение модели на новых примерах угроз"""

        # Подготовка датасета

        texts = [ex['log'] for ex in positive_examples + negative_examples]

        labels = [1]*len(positive_examples) + [0]*len(negative_examples)

        

        dataset = Dataset.from_dict({'text': texts, 'label': labels})

        dataset = dataset.train_test_split(test_size=0.2)

        

        # Токенизация

        def tokenize(batch):

            return self.tokenizer(batch['text'], padding=True, truncation=True)

        

        dataset = dataset.map(tokenize, batched=True)

        

        # Настройка обучения

        training_args = TrainingArguments(

            output_dir='/tmp/fewshot',

            num_train_epochs=epochs,

            per_device_train_batch_size=8,

            learning_rate=2e-5,

            weight_decay=0.01,

            logging_dir='/tmp/logs',

            logging_steps=10

        )

        

        # Обучение

        trainer = Trainer(

            model=self.model,

            args=training_args,

            train_dataset=dataset['train'],

            eval_dataset=dataset['test']

        )

        trainer.train()

        

        # Сохранение адаптивной модели

        self.model.save_pretrained('/opt/forensic-ai/models/fewshot_adapter')

        self.tokenizer.save_pretrained('/opt/forensic-ai/models/fewshot_adapter')

    

    def predict(self, log_text):

        """Предсказание для нового лога"""

        inputs = self.tokenizer(log_text, return_tensors='pt', truncation=True)

        

        with torch.no_grad():

            outputs = self.model(inputs)

            prob = torch.softmax(outputs.logits, dim=1)[0][1].item()

        

        return {

            'is_threat': prob > 0.75,

            'confidence': prob,

            'model_version': 'fewshot_adapter_v1'

        }

# Пример использования

detector = FewShotThreatDetector()



# Аналитик помечает 3 лога как "новая техника"

new_technique_examples = [

    {'log': 'PsExec executed from 192.168.1.10 to 192.168.1.25 with admin credentials'},

    {'log': 'WMI query executed remotely to enumerate processes on DC01'},

    {'log': 'Scheduled task created on remote host via schtasks.exe'}

]



# И 3 нормальных лога для контраста

normal_examples = [

    {'log': 'User logged in successfully via RDP'},

    {'log': 'File backup completed to network share'},

    {'log': 'Antivirus signature update applied'}

]



# Адаптация модели за 5 минут

detector.adapt_with_examples(new_technique_examples, normal_examples)



# Теперь модель детектирует похожие техники в реальном времени

# forensic-ai/explainer.py

import lime

import lime.lime_tabular



def explain_prediction_with_lime(model, feature_df, instance_idx, feature_names):

    """Объяснение предсказания для конкретного события"""

    # Подготовка данных для LIME

    X = feature_df.drop(['src_ip', 'dst_ip'], axis=1).values

    explainer = lime.lime_tabular.LimeTabularExplainer(

        X,

        feature_names=feature_names,

        class_names=['normal', 'anomaly'],

        mode='classification'

    )

    

    # Генерация объяснения

    exp = explainer.explain_instance(

        X[instance_idx], 

        model.predict_proba, 

        num_features=5

    )

    

    return {

        'predicted_class': model.predict([X[instance_idx]])[0],

        'confidence': model.predict_proba([X[instance_idx]])[0][1],

        'top_features': exp.as_list(),  # [(feature, weight), ...]

        'visualization_html': exp.as_pyplot_figure()

    }

def generate_counterfactual(instance, model, feature_ranges):

    """Поиск минимальных изменений, которые изменят предсказание"""

    from scipy.optimize import minimize

    

    def objective(perturbation):

        # Применение возмущения к признакам

        modified = instance.copy()

        for i, delta in enumerate(perturbation):

            modified[i] = np.clip(modified[i] + delta, *feature_ranges[i])

        

        # Штраф за величину изменений + за сохранение предсказания

        prediction = model.predict([modified])[0]

        change_penalty = np.sum(np.abs(perturbation))

        prediction_penalty = 1 if prediction == instance['predicted'] else 0

        

        return change_penalty + 10 * prediction_penalty

    

    # Оптимизация

    result = minimize(objective, x0=np.zeros(len(instance)), method='Nelder-Mead')

    

    return {

        'original': instance,

        'counterfactual': instance + result.x,

        'changes': {

            feature: f"{instance[i]:.2f} → {instance[i] + result.x[i]:.2f}"

            for i, feature in enumerate(feature_names)

            if abs(result.x[i]) > 0.01

        }

    }

# forensic-ai/siem_connector.py

from elasticsearch import Elasticsearch

from datetime import datetime, timedelta



class SIEMConnector:

    def __init__(self, es_host, index_pattern='logs-*'):

        self.es = Elasticsearch([es_host])

        self.index_pattern = index_pattern

    

    def fetch_recent_events(self, hours=1, query_filter=None):

        """Получение событий для анализа"""

        query = {

            "query": {

                "bool": {

                    "must": [

                        {"range": {"@timestamp": {

                            "gte": f"now-{hours}h",

                            "format": "epoch_millis"

                        }}}

                    ],

                    "filter": query_filter or []

                }

            },

            "size": 10000,

            "sort": [{"@timestamp": {"order": "asc"}}]

        }

        

        response = self.es.search(index=self.index_pattern, body=query)

        return [hit['_source'] for hit in response['hits']['hits']]

    

    def enrich_with_ai_predictions(self, events, predictions):

        """Добавление предсказаний ИИ к событиям в SIEM"""

        for event, pred in zip(events, predictions):

            event['ai_analysis'] = {

                'predicted_class': pred['predicted_class'],

                'confidence': pred['confidence'],

                'model_version': pred['model_version'],

                'analyzed_at': datetime.utcnow().isoformat()

            }

            

            # Обновление документа в Elasticsearch

            self.es.update(

                index=event['_index'],

                id=event['_id'],

                body={'doc': {'ai_analysis': event['ai_analysis']}}

            )

# forensic-ai/thehive_connector.py

from thehive4py.api import TheHiveApi

from thehive4py.models import Alert, AlertArtifact



def create_alert_from_finding(finding, hive_url, api_key):

    """Создание алерта в TheHive на основе вывода ИИ"""

    api = TheHiveApi(hive_url, api_key)

    

    artifacts = [

        AlertArtifact(dataType='ip', data=finding.get('src_ip')),

        AlertArtifact(dataType='domain', data=finding.get('dst_domain')),

        AlertArtifact(dataType='hash', data=finding.get('file_hash'))

    ]

    

    alert = Alert(

        title=f"[AI] {finding['type']} - Confidence {finding['confidence']:.0%}",

        description=generate_alert_description(finding),

        type='external',

        source='forensic-ai',

        sourceRef=finding['finding_id'],

        artifacts=[a for a in artifacts if a.data],  # Фильтрация пустых

        customFields={

            'ai_confidence': {'float': finding['confidence']},

            'ai_model': {'string': finding['model_version']},

            'requires_review': {'boolean': finding['confidence'] < 0.9}

        }

    )

    

    response = api.create_alert(alert)

    return response.json() if response.status_code == 201 else None

# config/auto_response_rules.yaml

rules:

  - name: high_confidence_malware

    condition:

      ai_prediction: malware

      confidence: ">= 0.95"

      asset_criticality: "high"

    actions:

      - isolate_host: true

      - create_ticket: true

      - notify_soc: true

    requires_approval: false  # Автоматическое выполнение

    

  - name: medium_confidence_anomaly

    condition:

      ai_prediction: anomaly

      confidence: ">= 0.75"

      confidence: "< 0.95"

    actions:

      - enrich_alert: true

      - request_analyst_review: true

    requires_approval: true  # Требует подтверждения аналитика

    

  - name: new_technique_detected

    condition:

      model_type: fewshot_adapter

      prediction: threat

    actions:

      - create_investigation_case: true

      - notify_threat_intel_team: true

      - trigger_model_retraining: true

    requires_approval: true

# forensic-ai/report_generator.py

from langchain.chains import RetrievalQA

from langchain.vectorstores import FAISS

from langchain.embeddings import HuggingFaceEmbeddings



class ForensicReportGenerator:

    def __init__(self, knowledge_base_path):

        # Векторизация базы знаний (политики, методики, прецеденты)

        embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-MiniLM-L6-v2')

        self.vectorstore = FAISS.load_local(knowledge_base_path, embeddings)

        

        # Настройка QA-цепочки с ограничением на галлюцинации

        self.qa_chain = RetrievalQA.from_chain_type(

            llm=load_restricted_llm(),  # LLM с отключённой креативностью

            chain_type='stuff',

            retriever=self.vectorstore.as_retriever(search_kwargs={'k': 5}),

            return_source_documents=True

        )

    

    def generate_section(self, section_prompt, findings, context):

        """Генерация раздела отчёта с привязкой к фактам"""

        # Формирование контекста из находок

        facts_context = "\n".join([

            f"- {f['type']}: {f['description']} (уверенность: {f['confidence']:.0%})"

            for f in findings

        ])

        

        # Запрос к LLM с жёстким ограничением

        full_prompt = f"""

Ты — эксперт по компьютерной криминалистике. Сформируй раздел отчёта на основе ТОЛЬКО предоставленных фактов.



Контекст расследования:

{context}



Обнаруженные артефакты:

{facts_context}



Задача: {section_prompt}



Требования:

1. Используй только факты из списка выше

2. Указывай уверенность для каждого вывода

3. Не делай предположений без данных

4. Форматируй как маркированный список



Ответ:"""

        

        result = self.qa_chain.run(full_prompt)

        return {

            'content': result,

            'sources': result.get('source_documents', []),  # Для аудита

            'fact_check_passed': verify_against_facts(result, findings)

        }

# forensic-ai/timeline_visualizer.py

import plotly.graph_objects as go



def create_incident_timeline(findings, start_time, end_time):

    """Интерактивная временная шкала с аннотациями"""

    fig = go.Figure()

    

    # Добавление событий

    for f in findings:

        fig.add_trace(go.Scatter(

            x=[f['timestamp']],

            y=[f['severity']],

            mode='markers+text',

            name=f['type'],

            marker=dict(

                size=10 + f['confidence']*20,

                color='red' if f['confidence']>0.9 else 'orange',

                symbol='diamond' if f['requires_review'] else 'circle'

            ),

            text=f['short_description'],

            textposition='top center',

            hovertemplate='<b>%{text}</b><br>Уверенность: %{marker.size}<br>Время: %{x}<extra></extra>'

        ))

    

    fig.update_layout(

        title='Временная шкала инцидента',

        xaxis_title='Время',

        yaxis_title='Критичность',

        hovermode='x unified',

        height=400

    )

    

    return fig.to_html(include_plotlyjs='cdn')

# forensic-ai/compliance.py

def ensure_forensic_soundness(analysis_result):

    """Проверка соответствия вывода требованиям доказательности"""

    checks = {

        'reproducibility': analysis_result.get('input_hash') is not None,

        'model_versioning': analysis_result.get('model_version') is not None,

        'chain_of_custody': analysis_result.get('custody_log') is not None,

        'human_review': analysis_result.get('analyst_approval') is True or analysis_result.get('confidence', 0) >= 0.95,

        'explanation_provided': len(analysis_result.get('explanation', [])) > 0

    }

    

    if not all(checks.values()):

        failed = [k for k, v in checks.items() if not v]

        raise ComplianceError(f"Нарушены требования: {', '.join(failed)}")

    

    return True