fix(tradein): security hotfix — C-3 config default + C-4 empty estimate persist + C-6 PDF URL allowlist (#435)
All checks were successful
Deploy Trade-In / changes (push) Successful in 5s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-backend (push) Successful in 38s
Deploy Trade-In / deploy (push) Successful in 25s

This commit is contained in:
lekss361 2026-05-23 09:38:23 +00:00
parent 3295808b41
commit 5e3ba351a0
3 changed files with 115 additions and 10 deletions

View file

@ -6,14 +6,17 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
database_url: str = "postgresql+psycopg://tradein:tradein@postgres:5432/tradein"
# required — задаётся через env DATABASE_URL. Нет дефолта: fail-fast при старте
# если переменная не задана (C-3 security audit).
database_url: str
cors_origins: list[str] = ["http://localhost", "http://localhost:3000", "http://localhost:8080"]
environment: str = "dev"
# Geocoder
yandex_geocoder_key: str | None = None # 25K req/day free после регистрации
yandex_suggest_key: str | None = None # для frontend autocomplete (proxy через backend)
contact_email: str = "erginrajpopxbe@outlook.com" # для User-Agent в Nominatim (Nominatim Usage Policy)
# для User-Agent в Nominatim (Nominatim Usage Policy)
contact_email: str = "erginrajpopxbe@outlook.com"
# Public URL — для QR-кода в PDF, shareable links, etc.
public_url: str = "http://127.0.0.1:8080"

View file

@ -23,7 +23,7 @@ import json
import logging
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID, uuid4
from uuid import uuid4
from sqlalchemy import text
from sqlalchemy.orm import Session
@ -83,7 +83,7 @@ async def estimate_quality(
if geo is None:
# Без координат не можем искать через PostGIS. Возвращаем low confidence.
logger.warning("geocode failed for %s — returning low-confidence estimate", payload.address)
return _empty_estimate(payload, reason="address_not_geocoded")
return _empty_estimate(payload, db, reason="address_not_geocoded")
# 2. #392: обогащаем год / тип дома из картографии (OSM Overpass), если
# пользователь их не указал — это улучшает house-match аналогов (#6).
@ -562,19 +562,79 @@ def _deal_to_analog(row: dict[str, Any]) -> AnalogLot:
)
def _empty_estimate(payload: TradeInEstimateInput, *, reason: str) -> AggregatedEstimate:
"""Fallback когда нет данных для оценки."""
def _empty_estimate(
payload: TradeInEstimateInput, db: Session, *, reason: str
) -> AggregatedEstimate:
"""Fallback когда нет данных для оценки.
Сохраняет запись в БД (confidence='low', пустые analogs/deals), чтобы GET /estimate/{id}
не возвращал 404. C-4 security audit.
"""
estimate_id = uuid4()
now = datetime.now(tz=UTC)
expires_at = now + timedelta(hours=24)
db.execute(
text(
"""
INSERT INTO trade_in_estimates (
id, address,
area_m2, rooms, floor, total_floors,
year_built, house_type, repair_state, has_balcony,
ownership_type, has_mortgage, client_name, client_phone,
median_price, range_low, range_high, median_price_per_m2,
confidence, confidence_explanation, n_analogs,
analogs, actual_deals,
sources_used,
expires_at
) VALUES (
CAST(:id AS uuid), :address,
:area, :rooms, :floor, :total_floors,
:year_built, :house_type, :repair_state, :has_balcony,
:ownership_type, :has_mortgage, :client_name, :client_phone,
0, 0, 0, 0,
'low', :explanation, 0,
'[]'::jsonb, '[]'::jsonb,
'[]'::jsonb,
:expires_at
)
"""
),
{
"id": str(estimate_id),
"address": payload.address,
"area": payload.area_m2,
"rooms": payload.rooms,
"floor": payload.floor,
"total_floors": payload.total_floors,
"year_built": payload.year_built,
"house_type": payload.house_type,
"repair_state": payload.repair_state,
"has_balcony": payload.has_balcony,
"ownership_type": payload.ownership_type,
"has_mortgage": payload.has_mortgage,
"client_name": payload.client_name,
"client_phone": payload.client_phone,
"explanation": reason,
"expires_at": expires_at,
},
)
db.commit()
logger.info(
"empty_estimate: id=%s reason=%s addr=%s", estimate_id, reason, payload.address[:60]
)
return AggregatedEstimate(
estimate_id=uuid4(),
estimate_id=estimate_id,
median_price_rub=0,
range_low_rub=0,
range_high_rub=0,
median_price_per_m2=0,
confidence="low",
confidence_explanation=reason,
n_analogs=0,
period_months=DEALS_PERIOD_MONTHS,
analogs=[],
actual_deals=[],
expires_at=now + timedelta(hours=24),
expires_at=expires_at,
)

View file

@ -17,6 +17,7 @@ import datetime as dt
import html as _html
import io
import logging
from urllib.parse import urlparse
from uuid import UUID
import segno
@ -45,6 +46,46 @@ def _bar_svg_data_url(q_low: float = 0.20, q_high: float = 0.80) -> str:
logger = logging.getLogger(__name__)
# ── URL allowlist (C-6 security audit) ─────────────────────────────────────
# Только доверенные домены-источники объявлений попадают в <a href=> в PDF.
# Защита от javascript: / data: инъекций через source_url объявлений.
_ALLOWED_URL_DOMAINS: frozenset[str] = frozenset({
"www.avito.ru", "avito.ru", "m.avito.ru",
"www.cian.ru", "cian.ru",
"realty.yandex.ru",
"n1.ru", "ekaterinburg.n1.ru",
})
# CDN-домены для изображений аналогов (photo_url).
_ALLOWED_PHOTO_CDN: frozenset[str] = frozenset({
"images.avito.st",
"cdn-p.cian.site",
"avatars.mds.yandex.net",
"n1ru.cdn-cw.com",
})
def _safe_url(url: str | None, *, cdn: bool = False) -> str | None:
"""Проверяет URL по allowlist схемы и домена.
cdn=False allowlist для source_url объявлений (ссылки на листинги).
cdn=True allowlist для photo_url (CDN изображений).
Возвращает None если URL не прошёл проверку embed/ссылка пропускается.
"""
if not url:
return None
try:
p = urlparse(url)
except Exception:
return None
if p.scheme != "https":
return None
allowed = _ALLOWED_PHOTO_CDN if cdn else _ALLOWED_URL_DOMAINS
if p.netloc not in allowed:
return None
return url
# ── Source pseudo-logos (текстовые pill-badges как у Брусники с логотипами) ──
_SOURCE_LOGO_COLORS: dict[str, tuple[str, str]] = {
"avito": ("#00aaff", "#fff"), # Avito brand blue (упрощённо)
@ -484,8 +525,9 @@ def _examples_rows(lots: list[AnalogLot]) -> str:
rows = []
for lot in lots:
addr = _html.escape(lot.address)
if lot.source_url:
addr_cell = f"<a href='{_html.escape(lot.source_url)}' style='color:#1d4ed8;text-decoration:none;'>{addr} ↗</a>"
safe = _safe_url(lot.source_url)
if safe:
addr_cell = f"<a href='{_html.escape(safe)}' style='color:#1d4ed8;text-decoration:none;'>{addr} ↗</a>"
else:
addr_cell = addr
rows.append(