neo
b9db935389
Исправлено: - engine: _calculate_score() при ошибке загрузки (score не 100 для мёртвых сайтов) - privacy_link: замена page.goto() на aiohttp (гонка данных) - form_checkbox: 3-уровневый поиск label + dark-pattern detection - server_location: синхронный urllib → aiohttp - api/server: перепутанные request/response модели в batch - main.py: csv --checks флаг - engine: фазовая модель (cookie_audit отдельно, остальные параллельно) - engine: retry-логика (3 попытки, exponential backoff) - PDF: emoji заменены на текст (WeasyPrint) Добавлено: - consent_doc: проверка отдельного согласия на обработку ПД - operator_info: проверка ИНН/ОГРН/DPO (читает главную + политику) - cookie_audit: 4-фазный аудит (load→detect→reject→compare) - llm_policy: опциональный LLM-анализ 16 разделов (--deep) - external_scripts: YAML-словари trackers.yaml (critical/warning/russian) - logging (output.py), .gitignore, pyproject.toml - smoke-тесты (16/16)
234 lines
11 KiB
Python
234 lines
11 KiB
Python
"""
|
|
Опциональный LLM-анализ политики конфиденциальности.
|
|
Запускается ПОСЛЕ основных проверок, только если явно запрошен.
|
|
Проверяет 16 обязательных разделов политики по 152-ФЗ.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
from bs4 import BeautifulSoup
|
|
|
|
from scanner.checks.base import BaseCheck
|
|
from scanner.models import CheckResult, CheckStatus, RiskLevel
|
|
|
|
|
|
# 16 обязательных разделов политики конфиденциальности (Reova, 2026)
|
|
REQUIRED_SECTIONS = [
|
|
{"key": "general", "name": "Общие положения", "article": "Ст. 3"},
|
|
{"key": "operator", "name": "Наименование оператора", "article": "Ст. 18.1"},
|
|
{"key": "purposes", "name": "Цели обработки", "article": "Ст. 5 ч.2"},
|
|
{"key": "categories", "name": "Категории ПД", "article": "Ст. 18.1"},
|
|
{"key": "subjects", "name": "Категории субъектов", "article": "Ст. 18.1"},
|
|
{"key": "basis", "name": "Правовые основания", "article": "Ст. 6"},
|
|
{"key": "methods", "name": "Способы обработки", "article": "Ст. 3"},
|
|
{"key": "storage", "name": "Сроки хранения", "article": "Ст. 5 ч.7"},
|
|
{"key": "destruction","name": "Порядок уничтожения", "article": "Ст. 21"},
|
|
{"key": "rights", "name": "Права субъекта", "article": "Ст. 14, 15"},
|
|
{"key": "security", "name": "Меры безопасности", "article": "Ст. 19"},
|
|
{"key": "transfer", "name": "Трансграничная передача", "article": "Ст. 12"},
|
|
{"key": "thirdparty", "name": "Передача третьим лицам", "article": "Ст. 6 ч.3"},
|
|
{"key": "cookies", "name": "Cookies", "article": "Разъяснения РКН"},
|
|
{"key": "dpo", "name": "Ответственный за ПД (DPO)", "article": "Ст. 22.1"},
|
|
{"key": "disputes", "name": "Порядок разрешения споров", "article": "Ст. 17"},
|
|
]
|
|
|
|
|
|
class LLMPolicyCheck(BaseCheck):
|
|
"""Глубокий LLM-анализ текста политики конфиденциальности."""
|
|
|
|
name = "llm_policy_analysis"
|
|
title = "LLM-анализ политики конфиденциальности"
|
|
description = "Проверяет 16 обязательных разделов политики через LLM"
|
|
risk_on_fail = RiskLevel.HIGH
|
|
default_weight = 20
|
|
|
|
def __init__(self, llm_url: str = "http://192.168.1.152:80/v1/chat/completions",
|
|
llm_model: str = "Qwen3.6-35B-A3B-UD-IQ4_NL.gguf"):
|
|
self.llm_url = llm_url
|
|
self.llm_model = llm_model
|
|
|
|
async def check(self, page, url: str) -> CheckResult:
|
|
"""Этот метод вызывается через API, не через Playwright."""
|
|
return self._skip_result(message="LLM-анализ запускается отдельно через analyze_policy()")
|
|
|
|
async def analyze_policy(self, policy_url: str) -> CheckResult:
|
|
"""
|
|
Загрузить страницу политики и проанализировать через LLM.
|
|
|
|
Args:
|
|
policy_url: URL страницы политики конфиденциальности
|
|
|
|
Returns:
|
|
CheckResult с детальным анализом 16 разделов
|
|
"""
|
|
try:
|
|
# 1. Скачиваем текст политики
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
policy_url,
|
|
headers={"User-Agent": "Mozilla/5.0"},
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as resp:
|
|
if resp.status != 200:
|
|
return self._error_result(f"Не удалось загрузить политику: HTTP {resp.status}")
|
|
html = await resp.text()
|
|
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
for tag in soup(["script", "style", "nav", "footer", "header"]):
|
|
tag.decompose()
|
|
policy_text = soup.get_text(separator=" ").strip()
|
|
|
|
if len(policy_text) < 100:
|
|
return self._fail_result(
|
|
message="Текст политики слишком короткий для анализа (<100 символов).",
|
|
details={"char_count": len(policy_text)},
|
|
)
|
|
|
|
# 2. Отправляем на LLM-анализ
|
|
sections_list = "\n".join(
|
|
f"- {s['key']}: {s['name']} ({s['article']})"
|
|
for s in REQUIRED_SECTIONS
|
|
)
|
|
|
|
system_prompt = (
|
|
"Ты юридический аудитор. Проанализируй текст политики конфиденциальности "
|
|
"на соответствие 152-ФЗ. Проверь наличие каждого из 16 обязательных разделов. "
|
|
"Отвечай СТРОГО в формате JSON без markdown-обёрток."
|
|
)
|
|
user_prompt = f"""Проверь, какие из этих разделов присутствуют в тексте политики:
|
|
|
|
{sections_list}
|
|
|
|
Для каждого раздела укажи:
|
|
- present: true/false (присутствует ли)
|
|
- evidence: короткая цитата из текста (до 100 символов), если present=true
|
|
- quality: "full"/"partial"/"missing"
|
|
|
|
Также дай:
|
|
- overall_score: общая оценка от 0 до 100
|
|
- summary: краткое резюме на русском (2-3 предложения)
|
|
|
|
Текст политики (первые 8000 символов):
|
|
{policy_text[:8000]}
|
|
|
|
Ответь ТОЛЬКО JSON, без markdown:"""
|
|
|
|
analysis = await self._call_llm(system_prompt, user_prompt)
|
|
|
|
if not analysis:
|
|
return self._error_result("LLM не вернул результат анализа.")
|
|
|
|
# 3. Формируем результат
|
|
present_count = sum(1 for s in analysis.get("sections", []) if s.get("present"))
|
|
missing_sections = [s["name"] for s in analysis.get("sections", []) if not s.get("present")]
|
|
partial_sections = [s["name"] for s in analysis.get("sections", []) if s.get("quality") == "partial"]
|
|
|
|
overall = analysis.get("overall_score", 0)
|
|
summary = analysis.get("summary", "Анализ завершён.")
|
|
|
|
if present_count >= 14 and not missing_sections:
|
|
return self._pass_result(
|
|
message=f"Политика содержит {present_count}/16 обязательных разделов. {summary}",
|
|
details={
|
|
"present_count": present_count,
|
|
"total_sections": 16,
|
|
"overall_score": overall,
|
|
"missing": missing_sections,
|
|
"partial": partial_sections,
|
|
"llm_summary": summary,
|
|
},
|
|
)
|
|
|
|
if present_count >= 10:
|
|
return self._fail_result(
|
|
message=f"Политика неполная: {present_count}/16 разделов. Отсутствуют: {', '.join(missing_sections[:5])}. {summary}",
|
|
risk_level=RiskLevel.HIGH,
|
|
details={
|
|
"present_count": present_count,
|
|
"total_sections": 16,
|
|
"overall_score": overall,
|
|
"missing": missing_sections,
|
|
"partial": partial_sections,
|
|
"llm_summary": summary,
|
|
},
|
|
)
|
|
|
|
return self._fail_result(
|
|
message=f"Политика критически неполная: {present_count}/16 разделов. Отсутствуют: {', '.join(missing_sections[:8])}. {summary}",
|
|
risk_level=RiskLevel.CRITICAL,
|
|
details={
|
|
"present_count": present_count,
|
|
"total_sections": 16,
|
|
"overall_score": overall,
|
|
"missing": missing_sections,
|
|
"partial": partial_sections,
|
|
"llm_summary": summary,
|
|
},
|
|
)
|
|
|
|
except Exception as e:
|
|
return self._error_result(f"Ошибка LLM-анализа: {str(e)}")
|
|
|
|
async def _call_llm(self, system_prompt: str, user_prompt: str) -> dict | None:
|
|
"""Вызвать LLM для анализа текста политики."""
|
|
try:
|
|
import aiohttp
|
|
payload = {
|
|
"model": self.llm_model,
|
|
"messages": [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
"temperature": 0.1,
|
|
"max_tokens": 2048,
|
|
}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
self.llm_url,
|
|
json=payload,
|
|
timeout=aiohttp.ClientTimeout(total=60),
|
|
) as resp:
|
|
if resp.status != 200:
|
|
return None
|
|
data = await resp.json()
|
|
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
|
|
|
|
# Извлекаем JSON из ответа (LLM может обернуть в ```json)
|
|
content = content.strip()
|
|
if content.startswith("```"):
|
|
content = re.sub(r"^```(?:json)?\s*", "", content)
|
|
content = re.sub(r"\s*```$", "", content)
|
|
|
|
return json.loads(content)
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def run_llm_analysis(policy_url: str, llm_url: str | None = None) -> dict[str, Any]:
|
|
"""
|
|
Удобная функция для запуска LLM-анализа из CLI/API.
|
|
|
|
Returns:
|
|
Словарь с результатом check и деталями
|
|
"""
|
|
checker = LLMPolicyCheck()
|
|
if llm_url:
|
|
checker.llm_url = llm_url
|
|
|
|
result = await checker.analyze_policy(policy_url)
|
|
return {
|
|
"status": result.status.value,
|
|
"title": result.title,
|
|
"message": result.message,
|
|
"risk_level": result.risk_level.value,
|
|
"details": result.details,
|
|
}
|