fix(pdf): brand-spoof + median=0 empty-state + SSRF + дисклеймер + sources (Refs #772) (#789)
Some checks failed
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / deploy (push) Has been cancelled
Deploy Trade-In / test (push) Successful in 25s
Deploy Trade-In / build-backend (push) Successful in 39s
Some checks failed
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / deploy (push) Has been cancelled
Deploy Trade-In / test (push) Successful in 25s
Deploy Trade-In / build-backend (push) Successful in 39s
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
This commit is contained in:
parent
96027357b5
commit
38d44f46b7
3 changed files with 628 additions and 25 deletions
|
|
@ -342,11 +342,13 @@ def get_estimate(
|
||||||
def estimate_pdf(
|
def estimate_pdf(
|
||||||
estimate_id: UUID,
|
estimate_id: UUID,
|
||||||
db: Annotated[Session, Depends(get_db)],
|
db: Annotated[Session, Depends(get_db)],
|
||||||
brand: str | None = None,
|
|
||||||
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
|
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
|
||||||
) -> Response:
|
) -> Response:
|
||||||
"""Скачать 4-страничный PDF-отчёт для оценки trade-in.
|
"""Скачать 4-страничный PDF-отчёт для оценки trade-in.
|
||||||
|
|
||||||
|
Бренд PDF определяется по владельцу оценки (estimate.created_by → brand slug),
|
||||||
|
а НЕ из ?brand= query param (#7 brand-spoofing fix).
|
||||||
|
|
||||||
Возвращает application/pdf attachment.
|
Возвращает application/pdf attachment.
|
||||||
404 — оценка не найдена.
|
404 — оценка не найдена.
|
||||||
410 — оценка просрочена (TTL 24ч).
|
410 — оценка просрочена (TTL 24ч).
|
||||||
|
|
@ -430,9 +432,13 @@ def estimate_pdf(
|
||||||
"has_balcony": row.has_balcony,
|
"has_balcony": row.has_balcony,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
from app.core.auth import get_brand_for_user as _brand_for_user
|
||||||
from app.services.brand import get_brand as _resolve_brand
|
from app.services.brand import get_brand as _resolve_brand
|
||||||
|
|
||||||
brand_obj = _resolve_brand(brand, db)
|
# #7 brand-spoofing fix: бренд определяется по владельцу оценки,
|
||||||
|
# а не по ?brand= query param (который был удалён из сигнатуры).
|
||||||
|
owner_brand_slug = _brand_for_user(row.created_by) if row.created_by else None
|
||||||
|
brand_obj = _resolve_brand(owner_brand_slug, db)
|
||||||
pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj)
|
pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj)
|
||||||
filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf"
|
filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf"
|
||||||
logger.info(
|
logger.info(
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,7 @@ import datetime as dt
|
||||||
import html as _html
|
import html as _html
|
||||||
import io
|
import io
|
||||||
import logging
|
import logging
|
||||||
|
import re
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
|
|
@ -95,6 +96,85 @@ def _safe_url(url: str | None, *, cdn: bool = False) -> str | None:
|
||||||
return url
|
return url
|
||||||
|
|
||||||
|
|
||||||
|
# ── Logo URL и color-sanitation (#13 SSRF / CSS-injection guard) ─────────────
|
||||||
|
# Допустимые домены для логотипов брендов (white-label).
|
||||||
|
_ALLOWED_LOGO_CDN: frozenset[str] = frozenset(
|
||||||
|
{
|
||||||
|
"gk-praktika.ru",
|
||||||
|
"www.gk-praktika.ru",
|
||||||
|
"prinzip.ru",
|
||||||
|
"www.prinzip.ru",
|
||||||
|
"cdn.gk-praktika.ru",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
_COLOR_RE = re.compile(r"^#[0-9A-Fa-f]{6}$")
|
||||||
|
|
||||||
|
|
||||||
|
def _safe_logo_url(url: str | None) -> str | None:
|
||||||
|
"""Валидирует logo_url бренда: только https:// + allowlist домен + не data:/file:.
|
||||||
|
|
||||||
|
data: URI допускаются для инлайн-логотипов (нет сетевого запроса).
|
||||||
|
Всё остальное отбрасывается. Если передан file:// — отбрасывается.
|
||||||
|
"""
|
||||||
|
if not url:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
p = urlparse(url)
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
if p.scheme == "data":
|
||||||
|
# data: URI безопасны — нет SSRF, только инлайн
|
||||||
|
return url
|
||||||
|
if p.scheme != "https":
|
||||||
|
# file://, http://, javascript: — все отбрасываем
|
||||||
|
return None
|
||||||
|
if p.netloc not in _ALLOWED_LOGO_CDN:
|
||||||
|
logger.warning("logo_url domain not in allowlist: %s", p.netloc)
|
||||||
|
return None
|
||||||
|
return url
|
||||||
|
|
||||||
|
|
||||||
|
def _safe_color(color: str | None, fallback: str) -> str:
|
||||||
|
"""Принимает только #RRGGBB hex. Всё остальное → fallback.
|
||||||
|
|
||||||
|
Защита от CSS-injection через primary_color/accent_color бренда
|
||||||
|
(WeasyPrint вставляет значение напрямую в style=).
|
||||||
|
"""
|
||||||
|
if color and _COLOR_RE.match(color):
|
||||||
|
return color
|
||||||
|
return fallback
|
||||||
|
|
||||||
|
|
||||||
|
# ── WeasyPrint custom url_fetcher (SSRF guard) ───────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _make_safe_url_fetcher(): # type: ignore[no-untyped-def]
|
||||||
|
"""Возвращает url_fetcher для WeasyPrint, блокирующий file:// и нелоговые схемы.
|
||||||
|
|
||||||
|
Разрешены: https:// (логотипы + шрифты CDN) и data: (инлайн SVG/PNG).
|
||||||
|
Блокированы: file://, http://, ftp:// и любые другие.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def fetcher(url: str, timeout: int = 10, ssl_context=None): # type: ignore[no-untyped-def]
|
||||||
|
# Импорт отложен до момента первого реального fetch, чтобы тесты могли
|
||||||
|
# подменять weasyprint.urls через sys.modules без ModuleNotFoundError.
|
||||||
|
from weasyprint.urls import default_url_fetcher
|
||||||
|
|
||||||
|
try:
|
||||||
|
p = urlparse(url)
|
||||||
|
except Exception:
|
||||||
|
raise ValueError(f"Invalid URL in PDF resource: {url!r}") from None
|
||||||
|
if p.scheme in ("data", "https"):
|
||||||
|
return default_url_fetcher(url, timeout=timeout, ssl_context=ssl_context)
|
||||||
|
raise ValueError(
|
||||||
|
f"PDF resource scheme '{p.scheme}' is not allowed (only data: and https:). "
|
||||||
|
f"URL: {url!r}"
|
||||||
|
)
|
||||||
|
|
||||||
|
return fetcher
|
||||||
|
|
||||||
|
|
||||||
# ── Source pseudo-logos (текстовые pill-badges как у Брусники с логотипами) ──
|
# ── Source pseudo-logos (текстовые pill-badges как у Брусники с логотипами) ──
|
||||||
_SOURCE_LOGO_COLORS: dict[str, tuple[str, str]] = {
|
_SOURCE_LOGO_COLORS: dict[str, tuple[str, str]] = {
|
||||||
"avito": ("#00aaff", "#fff"), # Avito brand blue (упрощённо)
|
"avito": ("#00aaff", "#fff"), # Avito brand blue (упрощённо)
|
||||||
|
|
@ -433,13 +513,28 @@ def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> s
|
||||||
sub_label="Диапазон цен по фактическим сделкам",
|
sub_label="Диапазон цен по фактическим сделкам",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
logo_url = _safe_logo_url(brand.logo_url)
|
||||||
|
logo_html = (
|
||||||
|
f"<img src='{_html.escape(logo_url)}' alt='{_html.escape(brand.name)}' "
|
||||||
|
f"style='max-height:28pt;max-width:120pt;object-fit:contain;vertical-align:middle;'/>"
|
||||||
|
if logo_url
|
||||||
|
else f"<span style='font-size:12pt;font-weight:900;letter-spacing:0.06em;'>"
|
||||||
|
f"{_html.escape(brand.name).upper()}</span>"
|
||||||
|
)
|
||||||
|
disclaimer_html = ""
|
||||||
|
if brand.pdf_disclaimer:
|
||||||
|
disclaimer_html = (
|
||||||
|
f"<p style='margin-top:10pt;font-size:7.5pt;color:#6b7280;line-height:1.4;"
|
||||||
|
f"border-top:1pt solid #e6e8ec;padding-top:6pt;'>"
|
||||||
|
f"{_html.escape(brand.pdf_disclaimer)}</p>"
|
||||||
|
)
|
||||||
|
|
||||||
return f"""
|
return f"""
|
||||||
<div style="page-break-after:always;">
|
<div style="page-break-after:always;">
|
||||||
|
|
||||||
<!-- Top brand bar — компактный -->
|
<!-- Top brand bar — компактный -->
|
||||||
<div style="background:{brand.primary_color};color:#fff;padding:8pt 14pt;margin-bottom:8pt;
|
<div style="background:{brand.primary_color};color:#fff;padding:8pt 14pt;margin-bottom:8pt;">
|
||||||
font-size:12pt;font-weight:900;letter-spacing:0.06em;">
|
{logo_html}
|
||||||
{_html.escape(brand.name).upper()}
|
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<h1 style="font-size:12pt;font-weight:700;margin:4pt 0 8pt 0;text-transform:uppercase;
|
<h1 style="font-size:12pt;font-weight:700;margin:4pt 0 8pt 0;text-transform:uppercase;
|
||||||
|
|
@ -502,6 +597,7 @@ def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> s
|
||||||
padding-top:8pt;">
|
padding-top:8pt;">
|
||||||
<strong>Этот отчёт онлайн:</strong> {settings.public_url}?id={estimate.estimate_id}
|
<strong>Этот отчёт онлайн:</strong> {settings.public_url}?id={estimate.estimate_id}
|
||||||
</p>
|
</p>
|
||||||
|
{disclaimer_html}
|
||||||
|
|
||||||
</div>
|
</div>
|
||||||
"""
|
"""
|
||||||
|
|
@ -528,10 +624,12 @@ def _deals_range(deals: list[AnalogLot], fallback: tuple[int, int]) -> tuple[int
|
||||||
|
|
||||||
def _build_listings_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
|
def _build_listings_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
|
||||||
n_total = estimate.n_analogs
|
n_total = estimate.n_analogs
|
||||||
n_with_repair = max(3, int(n_total * 0.4)) # упрощение: ~40% с указанным ремонтом
|
# Используем n_analogs — estimator уже фильтрует по repair_state при поиске.
|
||||||
|
# Fabricated "~40%" удалён (#33): показываем тот же count, что прошёл фильтр.
|
||||||
|
n_with_repair = n_total
|
||||||
|
|
||||||
# Source logos (pseudo) — 5 максимальных
|
# Source logos (pseudo) — берём из estimate.sources_used (не захардкоженный список).
|
||||||
sources_to_show = estimate.sources_used or ["avito", "cian", "domklik", "yandex", "rosreestr"]
|
sources_to_show = estimate.sources_used or []
|
||||||
sources_html = "".join(_source_logo_pill(s) for s in sources_to_show[:5])
|
sources_html = "".join(_source_logo_pill(s) for s in sources_to_show[:5])
|
||||||
|
|
||||||
# Params правой колонки — параметры поиска (НЕ конкретной квартиры)
|
# Params правой колонки — параметры поиска (НЕ конкретной квартиры)
|
||||||
|
|
@ -674,15 +772,17 @@ def _examples_rows(lots: list[AnalogLot]) -> str:
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
addr_cell = addr
|
addr_cell = addr
|
||||||
|
ppm2_cell = _fmt_ppm2(lot.price_per_m2) if lot.price_per_m2 is not None else "—"
|
||||||
|
price_cell = _fmt_ppm2(lot.price_rub) if lot.price_rub is not None else "—"
|
||||||
rows.append(
|
rows.append(
|
||||||
"<tr>"
|
"<tr>"
|
||||||
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;'>{addr_cell}</td>"
|
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;'>{addr_cell}</td>"
|
||||||
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;'>"
|
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;'>"
|
||||||
f"{_source_badge_inline(lot.source)}</td>"
|
f"{_source_badge_inline(lot.source)}</td>"
|
||||||
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
||||||
f"font-variant-numeric:tabular-nums;'>{_fmt_ppm2(lot.price_per_m2)}</td>"
|
f"font-variant-numeric:tabular-nums;'>{ppm2_cell}</td>"
|
||||||
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
||||||
f"font-variant-numeric:tabular-nums;'>{_fmt_ppm2(lot.price_rub)}</td>"
|
f"font-variant-numeric:tabular-nums;'>{price_cell}</td>"
|
||||||
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
f"<td style='padding:7pt 4pt;border-bottom:1px solid #f3f4f6;text-align:right;"
|
||||||
f"font-variant-numeric:tabular-nums;'>{lot.days_on_market or '—'}</td>"
|
f"font-variant-numeric:tabular-nums;'>{lot.days_on_market or '—'}</td>"
|
||||||
"</tr>"
|
"</tr>"
|
||||||
|
|
@ -698,9 +798,13 @@ def _build_deals_page(estimate: AggregatedEstimate, input_snapshot: dict, brand)
|
||||||
today = dt.date.today()
|
today = dt.date.today()
|
||||||
period_start = today - dt.timedelta(days=estimate.period_months * 30)
|
period_start = today - dt.timedelta(days=estimate.period_months * 30)
|
||||||
|
|
||||||
# Источники для сделок — Этажи / Домклик / Росреестр
|
# Источники для сделок — берём из estimate.sources_used (не захардкоженный список).
|
||||||
deal_sources = ["etazhi", "domklik", "rosreestr"]
|
# Фильтруем по известным источникам сделок; fallback к пустому (не fabricate).
|
||||||
sources_html = "".join(_source_logo_pill(s) for s in deal_sources)
|
_deal_source_keys = {"etazhi", "domklik", "rosreestr"}
|
||||||
|
deal_sources = [s for s in (estimate.sources_used or []) if s in _deal_source_keys]
|
||||||
|
if not deal_sources:
|
||||||
|
deal_sources = [s for s in (estimate.sources_used or [])]
|
||||||
|
sources_html = "".join(_source_logo_pill(s) for s in deal_sources[:5])
|
||||||
|
|
||||||
area = float(input_snapshot.get("area_m2", 0) or 0)
|
area = float(input_snapshot.get("area_m2", 0) or 0)
|
||||||
area_min = round(area * 0.85, 1)
|
area_min = round(area * 0.85, 1)
|
||||||
|
|
@ -1146,15 +1250,51 @@ h1, h2, h3 {{ margin: 0; }}
|
||||||
# ── Public API ───────────────────────────────────────────────────────────────
|
# ── Public API ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _build_insufficient_data_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
|
||||||
|
"""Страница-заглушка для случаев, когда данных недостаточно (#697/#9).
|
||||||
|
|
||||||
|
Рендерится вместо числового отчёта при estimate.insufficient_data=True.
|
||||||
|
Предотвращает публикацию «0,0 млн» и fabricated таблиц потерь.
|
||||||
|
"""
|
||||||
|
address = _html.escape(input_snapshot.get("address", "—"))
|
||||||
|
report_num = _report_number(estimate.estimate_id)
|
||||||
|
today = dt.date.today()
|
||||||
|
return f"""
|
||||||
|
<div>
|
||||||
|
<div style="background:{brand.primary_color};color:#fff;padding:8pt 14pt;margin-bottom:8pt;
|
||||||
|
font-size:12pt;font-weight:900;letter-spacing:0.06em;">
|
||||||
|
{_html.escape(brand.name).upper()}
|
||||||
|
</div>
|
||||||
|
<p style="font-size:9pt;color:#6b7280;margin:0 0 18pt 0;">
|
||||||
|
№ {report_num} · {today.strftime("%d.%m.%Y")} · {address}
|
||||||
|
</p>
|
||||||
|
<div style="margin:40pt auto;text-align:center;max-width:320pt;">
|
||||||
|
<div style="font-size:32pt;margin-bottom:12pt;">—</div>
|
||||||
|
<h2 style="font-size:14pt;font-weight:700;color:#1a1d23;margin-bottom:8pt;">
|
||||||
|
Недостаточно данных для оценки
|
||||||
|
</h2>
|
||||||
|
<p style="font-size:10pt;color:#6b7280;line-height:1.5;">
|
||||||
|
По данному объекту не найдено достаточного количества аналогов
|
||||||
|
для формирования надёжной рыночной оценки.
|
||||||
|
Пожалуйста, уточните параметры или обратитесь к специалисту.
|
||||||
|
</p>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
def generate_trade_in_pdf(
|
def generate_trade_in_pdf(
|
||||||
estimate: AggregatedEstimate,
|
estimate: AggregatedEstimate,
|
||||||
input_snapshot: dict, # type: ignore[type-arg]
|
input_snapshot: dict, # type: ignore[type-arg]
|
||||||
*,
|
*,
|
||||||
brand=None, # type: ignore[no-untyped-def]
|
brand=None, # type: ignore[no-untyped-def]
|
||||||
) -> bytes:
|
) -> bytes:
|
||||||
"""Генерирует 4-страничный WeasyPrint PDF в стиле Брусника.Обмен.
|
"""Генерирует WeasyPrint PDF в стиле Брусника.Обмен.
|
||||||
|
|
||||||
Pages:
|
Если estimate.insufficient_data=True — рендерит одностраничный empty-state
|
||||||
|
вместо числового отчёта («Недостаточно данных», #697/#9).
|
||||||
|
|
||||||
|
Pages (normal):
|
||||||
1. Cover — № отчёта + параметры + 2 диапазона + 3 блока «Что важно»
|
1. Cover — № отчёта + параметры + 2 диапазона + 3 блока «Что важно»
|
||||||
2. Listings — рынок аналогов + источники + таблица примеров
|
2. Listings — рынок аналогов + источники + таблица примеров
|
||||||
3. Deals — фактические сделки + источники + таблица
|
3. Deals — фактические сделки + источники + таблица
|
||||||
|
|
@ -1186,25 +1326,45 @@ def generate_trade_in_pdf(
|
||||||
pdf_disclaimer=None,
|
pdf_disclaimer=None,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Sanitize brand colors before inserting into CSS/style attributes (#13).
|
||||||
|
brand = Brand(
|
||||||
|
slug=brand.slug,
|
||||||
|
name=brand.name,
|
||||||
|
logo_url=brand.logo_url,
|
||||||
|
primary_color=_safe_color(brand.primary_color, "#1d4ed8"),
|
||||||
|
accent_color=_safe_color(brand.accent_color, "#f59e0b"),
|
||||||
|
footer_text=brand.footer_text,
|
||||||
|
pdf_disclaimer=brand.pdf_disclaimer,
|
||||||
|
)
|
||||||
|
|
||||||
|
# #697/#9: insufficient data → empty-state page, NOT fabricated 0-median report.
|
||||||
|
if estimate.insufficient_data:
|
||||||
|
body_html = _build_insufficient_data_page(estimate, input_snapshot, brand)
|
||||||
|
else:
|
||||||
|
body_html = (
|
||||||
|
_build_cover(estimate, input_snapshot, brand)
|
||||||
|
+ _build_listings_page(estimate, input_snapshot, brand)
|
||||||
|
+ _build_deals_page(estimate, input_snapshot, brand)
|
||||||
|
+ _build_offer_page(estimate, brand)
|
||||||
|
)
|
||||||
|
|
||||||
html_str = (
|
html_str = (
|
||||||
'<!DOCTYPE html><html lang="ru"><head><meta charset="UTF-8">'
|
'<!DOCTYPE html><html lang="ru"><head><meta charset="UTF-8">'
|
||||||
f"<title>Trade-In — {_html.escape(input_snapshot.get('address', ''))}</title>"
|
f"<title>Trade-In — {_html.escape(input_snapshot.get('address', ''))}</title>"
|
||||||
"</head><body>"
|
f"</head><body>{body_html}</body></html>"
|
||||||
+ _build_cover(estimate, input_snapshot, brand)
|
|
||||||
+ _build_listings_page(estimate, input_snapshot, brand)
|
|
||||||
+ _build_deals_page(estimate, input_snapshot, brand)
|
|
||||||
+ _build_offer_page(estimate, brand)
|
|
||||||
+ "</body></html>"
|
|
||||||
)
|
)
|
||||||
css_str = _build_css(brand=brand)
|
css_str = _build_css(brand=brand)
|
||||||
# base_url=file:/// чтобы WeasyPrint мог resolve file:// URLs для PNG
|
# SSRF guard (#13): кастомный url_fetcher блокирует file:// и нелоговые схемы.
|
||||||
pdf_bytes = HTML(string=html_str, base_url="file:///").write_pdf(
|
# base_url=None — не resolve никакие относительные URL (все ресурсы data: или https:).
|
||||||
stylesheets=[CSS(string=css_str)]
|
pdf_bytes = HTML(string=html_str, base_url=None).write_pdf(
|
||||||
|
stylesheets=[CSS(string=css_str)],
|
||||||
|
url_fetcher=_make_safe_url_fetcher(),
|
||||||
)
|
)
|
||||||
logger.info(
|
logger.info(
|
||||||
"PDF generated estimate_id=%s brand=%s size=%d (Brusnika-style)",
|
"PDF generated estimate_id=%s brand=%s size=%d insufficient_data=%s",
|
||||||
estimate.estimate_id,
|
estimate.estimate_id,
|
||||||
brand.slug,
|
brand.slug,
|
||||||
len(pdf_bytes),
|
len(pdf_bytes),
|
||||||
|
estimate.insufficient_data,
|
||||||
)
|
)
|
||||||
return pdf_bytes
|
return pdf_bytes
|
||||||
|
|
|
||||||
437
tradein-mvp/backend/tests/test_pdf_security.py
Normal file
437
tradein-mvp/backend/tests/test_pdf_security.py
Normal file
|
|
@ -0,0 +1,437 @@
|
||||||
|
"""PDF security + correctness tests (#7/#9/#13/#33).
|
||||||
|
|
||||||
|
Tests cover:
|
||||||
|
- Part A (#7): brand derived from owner, not query param (tested at service level via
|
||||||
|
generate_trade_in_pdf — API-layer change verified separately via grep).
|
||||||
|
- Part B (#9): insufficient_data=True → empty-state page, NOT numeric report / loss table.
|
||||||
|
- Part C (#8): pdf_disclaimer / logo_url from brand object rendered on cover.
|
||||||
|
- Part D (#13): _safe_logo_url rejects file:// / non-allowlist; _safe_color rejects
|
||||||
|
non-hex; _make_safe_url_fetcher blocks file:// scheme.
|
||||||
|
- Part E (#33): n_with_repair real count; sources_used respected; None fields no crash.
|
||||||
|
|
||||||
|
WeasyPrint is stubbed — tests exercise HTML builders only (consistent with
|
||||||
|
tests/services/test_trade_in_pdf_dual_price.py pattern).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from datetime import UTC, datetime, timedelta
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||||
|
|
||||||
|
_wp_mock = MagicMock()
|
||||||
|
sys.modules.setdefault("weasyprint", _wp_mock)
|
||||||
|
|
||||||
|
import pytest # noqa: E402
|
||||||
|
|
||||||
|
from app.schemas.trade_in import AggregatedEstimate, AnalogLot # noqa: E402
|
||||||
|
from app.services.brand import Brand # noqa: E402
|
||||||
|
from app.services.exporters import trade_in_pdf as mod # noqa: E402
|
||||||
|
|
||||||
|
# ── Fixtures ─────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
_GENERIC = Brand(
|
||||||
|
slug="generic",
|
||||||
|
name="Trade-In",
|
||||||
|
logo_url=None,
|
||||||
|
primary_color="#1d4ed8",
|
||||||
|
accent_color="#f59e0b",
|
||||||
|
footer_text=None,
|
||||||
|
pdf_disclaimer=None,
|
||||||
|
)
|
||||||
|
|
||||||
|
_PRAKTIKA = Brand(
|
||||||
|
slug="praktika",
|
||||||
|
name="Практика",
|
||||||
|
logo_url="https://gk-praktika.ru/images/logo/logo-pink.svg",
|
||||||
|
primary_color="#235F49",
|
||||||
|
accent_color="#1FCECB",
|
||||||
|
footer_text="ГК «Практика» · Екатеринбург",
|
||||||
|
pdf_disclaimer="Отчёт носит ориентировочный характер. ГК Практика, 2026.",
|
||||||
|
)
|
||||||
|
|
||||||
|
_SNAPSHOT = {
|
||||||
|
"address": "Екатеринбург, ул. Ленина, 1",
|
||||||
|
"area_m2": 50.0,
|
||||||
|
"rooms": 2,
|
||||||
|
"floor": 3,
|
||||||
|
"total_floors": 9,
|
||||||
|
"year_built": 2010,
|
||||||
|
"house_type": "panel",
|
||||||
|
"repair_state": "standard",
|
||||||
|
"has_balcony": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _estimate(**overrides) -> AggregatedEstimate:
|
||||||
|
base = dict(
|
||||||
|
estimate_id=uuid4(),
|
||||||
|
median_price_rub=10_000_000,
|
||||||
|
range_low_rub=9_000_000,
|
||||||
|
range_high_rub=11_000_000,
|
||||||
|
median_price_per_m2=200_000,
|
||||||
|
confidence="high",
|
||||||
|
n_analogs=15,
|
||||||
|
period_months=24,
|
||||||
|
analogs=[],
|
||||||
|
actual_deals=[],
|
||||||
|
expires_at=datetime.now(UTC) + timedelta(days=30),
|
||||||
|
)
|
||||||
|
base.update(overrides)
|
||||||
|
return AggregatedEstimate(**base)
|
||||||
|
|
||||||
|
|
||||||
|
def _zero_estimate(**overrides) -> AggregatedEstimate:
|
||||||
|
"""Оценка с median=0 — insufficient_data=True."""
|
||||||
|
base = dict(
|
||||||
|
estimate_id=uuid4(),
|
||||||
|
median_price_rub=0,
|
||||||
|
range_low_rub=0,
|
||||||
|
range_high_rub=0,
|
||||||
|
median_price_per_m2=0,
|
||||||
|
confidence="low",
|
||||||
|
n_analogs=0,
|
||||||
|
period_months=24,
|
||||||
|
analogs=[],
|
||||||
|
actual_deals=[],
|
||||||
|
expires_at=datetime.now(UTC) + timedelta(days=30),
|
||||||
|
)
|
||||||
|
base.update(overrides)
|
||||||
|
return AggregatedEstimate(**base)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part B (#9): insufficient_data guard ─────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_insufficient_data_renders_empty_state_not_price() -> None:
|
||||||
|
"""median=0 → _build_insufficient_data_page called, no numeric headline."""
|
||||||
|
est = _zero_estimate()
|
||||||
|
assert est.insufficient_data is True
|
||||||
|
html = mod._build_insufficient_data_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "Недостаточно данных" in html
|
||||||
|
# no fabricated price / loss table
|
||||||
|
assert "млн. руб." not in html
|
||||||
|
assert "Общие финансовые потери" not in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_insufficient_data_page_contains_address_and_report_num() -> None:
|
||||||
|
est = _zero_estimate()
|
||||||
|
html = mod._build_insufficient_data_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "Ленина" in html # address
|
||||||
|
assert "EKБ-" in html # report number prefix
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_pdf_insufficient_skips_offer_page() -> None:
|
||||||
|
"""generate_trade_in_pdf с insufficient_data → offer page (loss table) не рендерится."""
|
||||||
|
est = _zero_estimate()
|
||||||
|
# Stub WeasyPrint HTML/CSS so we can inspect the html_str fed to it.
|
||||||
|
captured: list[str] = []
|
||||||
|
|
||||||
|
class _FakeHTML:
|
||||||
|
def __init__(self, string=None, base_url=None, **kw):
|
||||||
|
captured.append(string or "")
|
||||||
|
|
||||||
|
def write_pdf(self, stylesheets=None, url_fetcher=None):
|
||||||
|
return b"%PDF-fake"
|
||||||
|
|
||||||
|
class _FakeCSS:
|
||||||
|
def __init__(self, string=None):
|
||||||
|
pass
|
||||||
|
|
||||||
|
with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}):
|
||||||
|
import app.services.exporters.trade_in_pdf as _mod
|
||||||
|
|
||||||
|
_mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC)
|
||||||
|
|
||||||
|
assert captured, "HTML was never passed to WeasyPrint mock"
|
||||||
|
html = captured[0]
|
||||||
|
assert "Недостаточно данных" in html
|
||||||
|
assert "Общие финансовые потери" not in html
|
||||||
|
assert "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ" not in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_pdf_normal_estimate_has_offer_page() -> None:
|
||||||
|
"""Полная оценка → offer page присутствует."""
|
||||||
|
est = _estimate()
|
||||||
|
captured: list[str] = []
|
||||||
|
|
||||||
|
class _FakeHTML:
|
||||||
|
def __init__(self, string=None, base_url=None, **kw):
|
||||||
|
captured.append(string or "")
|
||||||
|
|
||||||
|
def write_pdf(self, stylesheets=None, url_fetcher=None):
|
||||||
|
return b"%PDF-fake"
|
||||||
|
|
||||||
|
class _FakeCSS:
|
||||||
|
def __init__(self, string=None):
|
||||||
|
pass
|
||||||
|
|
||||||
|
with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}):
|
||||||
|
import app.services.exporters.trade_in_pdf as _mod
|
||||||
|
|
||||||
|
_mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC)
|
||||||
|
|
||||||
|
assert captured
|
||||||
|
html = captured[0]
|
||||||
|
assert "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ" in html
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part C (#8): disclaimer + logo_url in cover ──────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_cover_renders_pdf_disclaimer() -> None:
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_cover(est, _SNAPSHOT, _PRAKTIKA)
|
||||||
|
assert "Отчёт носит ориентировочный характер" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_cover_disclaimer_absent_when_none() -> None:
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_cover(est, _SNAPSHOT, _GENERIC)
|
||||||
|
# No stray disclaimer block
|
||||||
|
assert "ориентировочный характер" not in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_cover_renders_logo_img_when_valid_url() -> None:
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_cover(est, _SNAPSHOT, _PRAKTIKA)
|
||||||
|
assert "<img" in html
|
||||||
|
assert "gk-praktika.ru" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_cover_falls_back_to_name_when_logo_none() -> None:
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_cover(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "<img" not in html
|
||||||
|
assert "TRADE-IN" in html.upper()
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part D (#13): _safe_logo_url ─────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_accepts_allowlisted_https() -> None:
|
||||||
|
url = "https://gk-praktika.ru/images/logo.svg"
|
||||||
|
assert mod._safe_logo_url(url) == url
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_rejects_file_scheme() -> None:
|
||||||
|
assert mod._safe_logo_url("file:///etc/passwd") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_rejects_http() -> None:
|
||||||
|
assert mod._safe_logo_url("http://gk-praktika.ru/logo.svg") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_rejects_unknown_domain() -> None:
|
||||||
|
assert mod._safe_logo_url("https://evil.com/logo.svg") is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_accepts_data_uri() -> None:
|
||||||
|
data = "data:image/svg+xml;base64,PHN2Zy8+"
|
||||||
|
assert mod._safe_logo_url(data) == data
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_logo_url_none_input() -> None:
|
||||||
|
assert mod._safe_logo_url(None) is None
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part D (#13): _safe_color ─────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_color_accepts_valid_hex() -> None:
|
||||||
|
assert mod._safe_color("#1d4ed8", "#000000") == "#1d4ed8"
|
||||||
|
assert mod._safe_color("#AbCdEf", "#000000") == "#AbCdEf"
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_color_rejects_css_expression() -> None:
|
||||||
|
assert mod._safe_color("expression(alert(1))", "#1d4ed8") == "#1d4ed8"
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_color_rejects_short_hex() -> None:
|
||||||
|
# 3-digit hex not accepted (require strict 6-digit)
|
||||||
|
assert mod._safe_color("#abc", "#1d4ed8") == "#1d4ed8"
|
||||||
|
|
||||||
|
|
||||||
|
def test_safe_color_rejects_none() -> None:
|
||||||
|
assert mod._safe_color(None, "#1d4ed8") == "#1d4ed8"
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part D (#13): url_fetcher blocks file:// ─────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_url_fetcher_rejects_file_scheme() -> None:
|
||||||
|
"""_make_safe_url_fetcher returns a fetcher that raises on file:// URLs."""
|
||||||
|
fake_default = MagicMock(return_value={"content": b"fake"})
|
||||||
|
urls_mod = MagicMock()
|
||||||
|
urls_mod.default_url_fetcher = fake_default
|
||||||
|
fetcher = mod._make_safe_url_fetcher()
|
||||||
|
with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}):
|
||||||
|
with pytest.raises(ValueError, match="not allowed"):
|
||||||
|
fetcher("file:///etc/passwd")
|
||||||
|
|
||||||
|
|
||||||
|
def test_url_fetcher_allows_data_uri() -> None:
|
||||||
|
fake_default = MagicMock(return_value={"content": b"fake"})
|
||||||
|
urls_mod = MagicMock()
|
||||||
|
urls_mod.default_url_fetcher = fake_default
|
||||||
|
fetcher = mod._make_safe_url_fetcher()
|
||||||
|
with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}):
|
||||||
|
result = fetcher("data:image/svg+xml;base64,PHN2Zy8+")
|
||||||
|
fake_default.assert_called_once()
|
||||||
|
assert result == {"content": b"fake"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_url_fetcher_allows_https() -> None:
|
||||||
|
fake_default = MagicMock(return_value={"content": b"fake"})
|
||||||
|
urls_mod = MagicMock()
|
||||||
|
urls_mod.default_url_fetcher = fake_default
|
||||||
|
fetcher = mod._make_safe_url_fetcher()
|
||||||
|
with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}):
|
||||||
|
fetcher("https://gk-praktika.ru/logo.svg")
|
||||||
|
fake_default.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_url_fetcher_rejects_http() -> None:
|
||||||
|
fake_default = MagicMock(return_value={"content": b"fake"})
|
||||||
|
urls_mod = MagicMock()
|
||||||
|
urls_mod.default_url_fetcher = fake_default
|
||||||
|
fetcher = mod._make_safe_url_fetcher()
|
||||||
|
with patch.dict(sys.modules, {"weasyprint.urls": urls_mod}):
|
||||||
|
with pytest.raises(ValueError, match="not allowed"):
|
||||||
|
fetcher("http://example.com/style.css")
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part D (#13): base_url no longer "file:///" ───────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_pdf_no_file_base_url() -> None:
|
||||||
|
"""generate_trade_in_pdf must NOT pass base_url='file:///' to WeasyPrint."""
|
||||||
|
est = _estimate()
|
||||||
|
html_calls: list[dict] = []
|
||||||
|
|
||||||
|
class _FakeHTML:
|
||||||
|
def __init__(self, string=None, base_url=None, **kw):
|
||||||
|
html_calls.append({"base_url": base_url})
|
||||||
|
|
||||||
|
def write_pdf(self, stylesheets=None, url_fetcher=None):
|
||||||
|
return b"%PDF-fake"
|
||||||
|
|
||||||
|
class _FakeCSS:
|
||||||
|
def __init__(self, string=None):
|
||||||
|
pass
|
||||||
|
|
||||||
|
with patch.dict(sys.modules, {"weasyprint": MagicMock(HTML=_FakeHTML, CSS=_FakeCSS)}):
|
||||||
|
import app.services.exporters.trade_in_pdf as _mod
|
||||||
|
|
||||||
|
_mod.generate_trade_in_pdf(est, _SNAPSHOT, brand=_GENERIC)
|
||||||
|
|
||||||
|
assert html_calls, "HTML() was never called"
|
||||||
|
base_url = html_calls[0]["base_url"]
|
||||||
|
assert base_url != "file:///", f"base_url should NOT be 'file:///' (got {base_url!r})"
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part E (#33): n_with_repair + sources + None-guards ──────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _analog(**overrides) -> AnalogLot:
|
||||||
|
base = dict(
|
||||||
|
address="ул. Тест, 1",
|
||||||
|
source="avito",
|
||||||
|
source_url="https://avito.ru/1",
|
||||||
|
price_rub=5_000_000,
|
||||||
|
price_per_m2=100_000,
|
||||||
|
area_m2=50.0,
|
||||||
|
rooms=2,
|
||||||
|
floor=3,
|
||||||
|
total_floors=9,
|
||||||
|
listing_date=None,
|
||||||
|
days_on_market=None,
|
||||||
|
distance_m=200,
|
||||||
|
)
|
||||||
|
base.update(overrides)
|
||||||
|
return AnalogLot(**base)
|
||||||
|
|
||||||
|
|
||||||
|
def test_n_with_repair_equals_n_analogs_not_fabricated() -> None:
|
||||||
|
"""n_with_repair должен совпадать с n_analogs (не fabricated ~40%)."""
|
||||||
|
est = _estimate(n_analogs=12, sources_used=["avito"])
|
||||||
|
html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
# both counts should show 12 (no fabricated sub-count)
|
||||||
|
assert "12 шт." in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_listings_page_no_crash_with_no_sources() -> None:
|
||||||
|
est = _estimate(sources_used=[])
|
||||||
|
html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "РЫНОК КВАРТИР" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_listings_sources_from_estimate_not_hardcoded() -> None:
|
||||||
|
"""sources_used=['yandex'] → только yandex pill, не avito/cian/domklik/..."""
|
||||||
|
est = _estimate(sources_used=["yandex"])
|
||||||
|
html = mod._build_listings_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "Я.Недвижимость" in html
|
||||||
|
# avito and cian should NOT appear if not in sources_used
|
||||||
|
assert "Avito" not in html
|
||||||
|
assert "Циан" not in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_deals_sources_from_estimate_not_hardcoded() -> None:
|
||||||
|
"""sources_used=['rosreestr'] → только rosreestr badges на deals-странице."""
|
||||||
|
est = _estimate(sources_used=["rosreestr"])
|
||||||
|
html = mod._build_deals_page(est, _SNAPSHOT, _GENERIC)
|
||||||
|
assert "Росреестр" in html
|
||||||
|
# Этажи/Домклик — не должны появиться
|
||||||
|
assert "Этажи" not in html
|
||||||
|
assert "Домклик" not in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_examples_rows_none_price_per_m2_no_crash() -> None:
|
||||||
|
"""AnalogLot с price_per_m2=None — _examples_rows не падает, рендерит «—»."""
|
||||||
|
# price_per_m2 is typed int (not Optional) in AnalogLot, but defensively
|
||||||
|
# the renderer must guard it. We pass a valid lot and verify no crash.
|
||||||
|
lot = _analog(price_per_m2=100_000, days_on_market=None)
|
||||||
|
html = mod._examples_rows([lot])
|
||||||
|
assert "100 000" in html
|
||||||
|
assert "5 000 000" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_examples_rows_empty_list_no_crash() -> None:
|
||||||
|
html = mod._examples_rows([])
|
||||||
|
assert "Нет данных" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_listings_page_none_year_built_no_crash() -> None:
|
||||||
|
snap = dict(_SNAPSHOT)
|
||||||
|
snap["year_built"] = None
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_listings_page(est, snap, _GENERIC)
|
||||||
|
assert "РЫНОК КВАРТИР" in html
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_deals_page_none_year_built_no_crash() -> None:
|
||||||
|
snap = dict(_SNAPSHOT)
|
||||||
|
snap["year_built"] = None
|
||||||
|
est = _estimate()
|
||||||
|
html = mod._build_deals_page(est, snap, _GENERIC)
|
||||||
|
assert "СДЕЛКИ" in html.upper()
|
||||||
|
|
||||||
|
|
||||||
|
# ── Part A (#7): brand derived from owner (service-level check) ───────────────
|
||||||
|
|
||||||
|
|
||||||
|
def test_brand_not_taken_from_query_param_docstring() -> None:
|
||||||
|
"""Verify the PDF endpoint no longer has a 'brand' query parameter."""
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
from app.api.v1.trade_in import estimate_pdf
|
||||||
|
|
||||||
|
sig = inspect.signature(estimate_pdf)
|
||||||
|
param_names = list(sig.parameters.keys())
|
||||||
|
assert (
|
||||||
|
"brand" not in param_names
|
||||||
|
), "estimate_pdf should NOT have a 'brand' query param after #7 fix"
|
||||||
Loading…
Add table
Reference in a new issue