neo
de18090304
- scanner/engine.py — Playwright-движок сканирования - scanner/models.py — Pydantic модели - scanner/checks/ — 5 проверок (cookie, privacy, forms, server, external) - api/server.py — FastAPI REST API - reports/generator.py — PDF через WeasyPrint + Jinja2 - main.py — CLI + сервер Проверки: ✅ cookie_banner — поиск cookie-баннера ✅ privacy_link — проверка ссылки на политику ✅ form_checkbox — проверка чекбоксов согласия ✅ server_location — геолокация сервера (RФ или нет) ✅ external_scripts — внешние скрипты аналитики
128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
"""
|
|
Движок сканирования 152-ФЗ.
|
|
Использует Playwright для загрузки сайтов и выполнения проверок.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from playwright.async_api import async_playwright, Page
|
|
|
|
from scanner.checks import ALL_CHECKS
|
|
from scanner.models import CheckResult, CheckStatus, RiskLevel, SiteScan
|
|
|
|
|
|
class ScannerEngine:
|
|
"""Управляет процессом сканирования сайтов."""
|
|
|
|
def __init__(self, timeout: int = 15000, headless: bool = True):
|
|
self.timeout = timeout
|
|
self.headless = headless
|
|
|
|
async def scan_url(self, url: str, checks: list[str] | None = None) -> SiteScan:
|
|
"""
|
|
Просканировать один URL.
|
|
|
|
Args:
|
|
url: URL для сканирования
|
|
checks: список имен проверок (None = все)
|
|
|
|
Returns:
|
|
SiteScan с результатами
|
|
"""
|
|
scan_id = str(uuid.uuid4())[:8]
|
|
result = SiteScan(url=url, scanned_at=datetime.now())
|
|
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(
|
|
headless=self.headless,
|
|
args=[
|
|
"--no-sandbox",
|
|
"--disable-setuid-sandbox",
|
|
"--disable-dev-shm-usage",
|
|
"--disable-gpu",
|
|
],
|
|
)
|
|
|
|
try:
|
|
context = await browser.new_context(
|
|
viewport={"width": 1920, "height": 1080},
|
|
user_agent=(
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/120.0.0.0 Safari/537.36"
|
|
),
|
|
)
|
|
|
|
page = await context.new_page()
|
|
|
|
# Устанавливаем stealth (отключаем webdriver flag)
|
|
await page.add_init_script("""
|
|
Object.defineProperty(navigator, 'webdriver', {
|
|
get: () => undefined
|
|
});
|
|
""")
|
|
|
|
# Загружаем страницу
|
|
try:
|
|
await page.goto(url, timeout=self.timeout, wait_until="networkidle")
|
|
except Exception as e:
|
|
result.checks.append(
|
|
CheckResult(
|
|
name="page_load",
|
|
title="Загрузка страницы",
|
|
status=CheckStatus.ERROR,
|
|
risk_level=RiskLevel.CRITICAL,
|
|
message=f"Не удалось загрузить страницу: {str(e)}",
|
|
)
|
|
)
|
|
return result
|
|
|
|
# Подождем, пока JavaScript отрендерит динамический контент
|
|
await page.wait_for_timeout(3000)
|
|
|
|
# Запускаем все проверки
|
|
check_names = checks if checks else list(ALL_CHECKS.keys())
|
|
|
|
tasks = []
|
|
for check_name in check_names:
|
|
check_cls = ALL_CHECKS.get(check_name)
|
|
if check_cls:
|
|
check_instance = check_cls()
|
|
tasks.append(check_instance.check(page, url))
|
|
|
|
if tasks:
|
|
check_results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
for cr in check_results:
|
|
if isinstance(cr, CheckResult):
|
|
result.checks.append(cr)
|
|
elif isinstance(cr, Exception):
|
|
result.checks.append(
|
|
CheckResult(
|
|
name="unknown",
|
|
title="Неизвестная ошибка",
|
|
status=CheckStatus.ERROR,
|
|
risk_level=RiskLevel.CRITICAL,
|
|
message=f"Необработанная ошибка: {str(cr)}",
|
|
)
|
|
)
|
|
|
|
finally:
|
|
await browser.close()
|
|
|
|
# Вычисляем итоговые метрики
|
|
result._calculate_score()
|
|
return result
|
|
|
|
async def scan_urls_batch(self, urls: list[str], checks: list[str] | None = None) -> list[SiteScan]:
|
|
"""Параллельно сканировать несколько URL."""
|
|
tasks = [self.scan_url(url, checks) for url in urls]
|
|
return await asyncio.gather(*tasks)
|
|
|
|
|
|
# Глобальный экземпляр для переиспользования
|
|
scanner_engine = ScannerEngine()
|