dokogenMU/dokogen/fstec_threats.py

# -*- coding: utf-8 -*-
"""DokoGen — загрузка перечня угроз безопасности информации с сайта ФСТЭК (БДУ).

Скачивает https://bdu.fstec.ru/threat (таблица угроз, с пагинацией),
парсит названия угроз и сохраняет в dictionaries/threats_fstec.json.
Рядом хранится дата актуализации (threats_fstec_date.txt).
"""

import os
import re
import json
import ssl
import urllib.request
import urllib.parse
from datetime import datetime

BDU_URL = 'https://bdu.fstec.ru/threat'
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'


def _ssl_ctx():
    """SSL-контекст: сначала обычная проверка, при проблемах — без проверки."""
    try:
        ctx = ssl.create_default_context()
        return ctx
    except Exception:
        pass
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def _fetch(url, timeout=30):
    """GET с User-Agent, возвращает текст или None."""
    req = urllib.request.Request(url, headers={'User-Agent': USER_AGENT})
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=_ssl_ctx()) as resp:
            return resp.read().decode('utf-8', errors='ignore')
    except Exception:
        # повтор с отключённой проверкой сертификата (некоторые сети/прокси)
        try:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
                return resp.read().decode('utf-8', errors='ignore')
        except Exception:
            return None


def _parse_threats(html):
    """Извлекает названия угроз из HTML-таблицы БДУ.

    Строки вида: <tr>...<td>УБИ. 001</td><td>Угроза ...</td>...</tr>
    Возвращает список названий (без кода).
    """
    names = []
    if not html:
        return names
    for row in re.findall(r'<tr[^>]*>(.*?)</tr>', html, re.S):
        cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row, re.S)
        cleaned = [re.sub(r'<[^>]+>', '', c).strip() for c in cells]
        # формат: [код, название, ...] — берём ячейку, которая не похожа на код УБИ
        for cell in cleaned:
            if not cell:
                continue
            if re.fullmatch(r'УБИ\.?\s*[\d.]+', cell):
                continue
            if cell in ('№', 'Идентификатор', 'Наименование', 'Дата', 'Уровень опасности'):
                continue
            names.append(cell)
            break
    return names


def download_fstec_threats(dict_dir=None, max_pages=8, page_size=100):
    """Скачивает перечень угроз с bdu.fstec.ru.

    Возвращает (ok, сообщение, список_угроз). При успехе сохраняет:
    - dictionaries/threats_fstec.json — список названий угроз
    - dictionaries/threats_fstec_date.txt — дата актуализации (ГГГГ-ММ-ДД)
    """
    if dict_dir is None:
        dict_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dictionaries')

    threats = []
    for page in range(1, max_pages + 1):
        url = f"{BDU_URL}?size={page_size}&page={page}"
        html = _fetch(url)
        if not html:
            break
        names = _parse_threats(html)
        if not names:
            break
        for n in names:
            if n not in threats:
                threats.append(n)
        # если на странице меньше page_size — это последняя
        if len(names) < page_size:
            break

    if not threats:
        return False, "Не удалось загрузить угрозы с сайта ФСТЭК (bdu.fstec.ru)", []

    try:
        os.makedirs(dict_dir, exist_ok=True)
        with open(os.path.join(dict_dir, 'threats_fstec.json'), 'w', encoding='utf-8') as f:
            json.dump(threats, f, ensure_ascii=False, indent=2)
        today = datetime.now().strftime('%Y-%m-%d')
        with open(os.path.join(dict_dir, 'threats_fstec_date.txt'), 'w', encoding='utf-8') as f:
            f.write(today)
    except Exception as e:
        return False, f"Ошибка сохранения: {e}", threats

    return True, f"Загружено угроз: {len(threats)}", threats


def get_threats_date(dict_dir=None):
    """Дата последней актуализации списка угроз (или None)."""
    if dict_dir is None:
        dict_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dictionaries')
    path = os.path.join(dict_dir, 'threats_fstec_date.txt')
    if os.path.exists(path):
        try:
            with open(path, encoding='utf-8') as f:
                return f.read().strip()
        except Exception:
            pass
    return None


def load_threats(dict_dir=None):
    """Загружает список угроз: сначала скачанный с ФСТЭК, иначе базовый threats.json."""
    if dict_dir is None:
        dict_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dictionaries')
    fstec = os.path.join(dict_dir, 'threats_fstec.json')
    if os.path.exists(fstec):
        try:
            with open(fstec, encoding='utf-8') as f:
                return json.load(f)
        except Exception:
            pass
    base = os.path.join(dict_dir, 'threats.json')
    try:
        with open(base, encoding='utf-8') as f:
            return json.load(f)
    except Exception:
        return []