gendesign/tradein-mvp/backend/app/api/v1/lead.py

"""Trade-in lead capture endpoint (issue #2376, sub-issue родителя #1971).

POST /api/v1/trade-in/lead — контактная заявка с результата оценки:
телефон + явное согласие на обработку персональных данных. Persist в
trade_in_leads. Notification (Telegram/email) — вне scope: нет существующей
SMTP/Telegram интеграции в коде (подтверждено при разборе issue), только
persist + log; `notified_at` в таблице зарезервирован под будущую доставку.

IDOR-фикс (security-audit): `estimate_id` раньше только проверялся на
СУЩЕСТВОВАНИЕ (`SELECT 1 ... WHERE id = ...`), без проверки владельца — любой
аутентифицированный пилот мог привязать свою заявку к чужой оценке (утечка через
последующий просмотр лида: чужой адрес/телефон/оценка в заявке, которую видит не
её владелец). Гвард переиспользует `_assert_estimate_access` из
`app.api.v1.trade_in` — тот же owner-or-admin подход, что и `GET /estimate/{id}`
(#690, `tests/test_estimate_idor.py`): 401 без `X-Authenticated-User`, 403 —
неизвестная роль, 404 — оценка не найдена ИЛИ принадлежит не этому пользователю
(существование чужой оценки не подтверждаем).
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime, timedelta
from typing import Annotated, Any, Literal
from uuid import UUID

from fastapi import APIRouter, Depends, Header, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.api.v1.trade_in import _assert_estimate_access
from app.core.config import settings
from app.core.db import get_db

logger = logging.getLogger(__name__)

router = APIRouter()

# Простая маска телефона (RU/международная): опциональный "+", цифры/пробелы/
# скобки/дефисы/точки, 5-32 символа. Полная нормализация в E.164 — вне scope MVP,
# см. #2376 DoD ("простая regex, не EmailStr-подобное").
_PHONE_PATTERN = r"^[+]?[\d\s().-]{5,32}$"
# Маска выше делает цифры ОПЦИОНАЛЬНЫМИ: "(()) -- .." её проходит (0 цифр).
# Поэтому дополнительно требуем правдоподобное число реальных цифр. RU-мобильный =
# 11 цифр; берём лениентный диапазон 10-15 (нац. номер без/с кодом страны).
_PHONE_MIN_DIGITS = 10
_PHONE_MAX_DIGITS = 15

# Версия политики обработки ПДн (152-ФЗ), под которую собрано согласие. Персистится
# per-row в trade_in_leads.consent_policy_version (migration 182) — до неё писалась
# только в audit-лог (#2497 TODO, теперь закрыт).
_CONSENT_POLICY_VERSION = "2026-07"

# Снимок точного текста согласия, который видит пользователь при отправке лида.
# Должен ДОСЛОВНО совпадать с чекбоксом в LeadForm.tsx (frontend/src/components/
# trade-in/v2/LeadForm.tsx) — если текст политики меняется, здесь нужно поднять
# _CONSENT_POLICY_VERSION И обновить этот снимок в одном PR, иначе новые строки
# будут нести устаревший snapshot под новой version-меткой.
_CONSENT_TEXT_SNAPSHOT = (
    "Согласен(-на) на обработку персональных данных в соответствии с "
    "Федеральным законом «О персональных данных» № 152-ФЗ"
)


class TradeInLeadInput(BaseModel):
    phone: str = Field(min_length=5, max_length=32, pattern=_PHONE_PATTERN)
    estimate_id: UUID | None = None
    consent: Literal[True]
    # "landing"-воронка недостижима: /api/v1/trade-in/lead закрыт rbac_guard
    # (main.py — путь не в _PUBLIC_PATHS => 401 без X-Authenticated-User), а
    # публичного лендинг-роута нет. Убрали мёртвый литерал, чтобы контракт не
    # обещал невозможную воронку (#2376). Вернуть, если появится public-роут.
    source: Literal["result"] = "result"

    @field_validator("phone")
    @classmethod
    def _phone_has_enough_digits(cls, value: str) -> str:
        digits = len(re.sub(r"\D", "", value))
        if not (_PHONE_MIN_DIGITS <= digits <= _PHONE_MAX_DIGITS):
            raise ValueError(f"phone must contain {_PHONE_MIN_DIGITS}-{_PHONE_MAX_DIGITS} digits")
        return value


@router.post("/lead")
async def create_trade_in_lead(
    payload: TradeInLeadInput,
    request: Request,
    db: Annotated[Session, Depends(get_db)],
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> dict[str, Any]:
    """Сохраняет лид (телефон + согласие) в trade_in_leads."""
    if payload.estimate_id is not None:
        estimate_row = db.execute(
            text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
            {"id": str(payload.estimate_id)},
        ).fetchone()
        if estimate_row is None:
            raise HTTPException(status_code=404, detail="estimate not found")
        # IDOR guard (security-audit, зеркалит #690): нельзя привязать лид к
        # чужой оценке. 404 и на "не найдено", и на "чужая" — не подтверждаем
        # существование чужого estimate_id.
        _assert_estimate_access(estimate_row.created_by, x_authenticated_user)

    user_agent = request.headers.get("user-agent")
    # 152-ФЗ audit trail: реальный клиентский IP из X-Forwarded-For (его ставит
    # фронтящий Caddy), fallback — прямой peer. Персистится per-row в
    # trade_in_leads.client_ip (migration 182), а не только в лог.
    client_ip = request.headers.get("x-forwarded-for")
    if client_ip:
        client_ip = client_ip.split(",")[0].strip()
    elif request.client is not None:
        client_ip = request.client.host

    # 152-ФЗ proof-of-consent: client_ip / consent_policy_version /
    # consent_text_snapshot теперь durable-колонки на trade_in_leads (migration 182,
    # ранее — только audit-лог, #2497 TODO). client_ip может быть None (нет
    # X-Forwarded-For и request.client) — колонка nullable, CAST(NULL AS inet) валиден.
    #
    # ЭТАП 4 B2C: expires_at (migration 193) — раньше лид хранился бессрочно
    # (никакого TTL вообще не было, в отличие от trade_in_estimates.expires_at).
    # Считаем на insert-time тем же паттерном, что estimator.py делает для
    # trade_in_estimates — retention-период вынесен в settings, не хардкод.
    expires_at = datetime.now(tz=UTC) + timedelta(days=settings.trade_in_lead_retention_days)

    row = (
        db.execute(
            text(
                """
            INSERT INTO trade_in_leads (
                estimate_id, phone, consent, source, user_agent,
                client_ip, consent_policy_version, consent_text_snapshot,
                expires_at
            )
            VALUES (
                CAST(:estimate_id AS uuid), :phone, :consent, :source, :user_agent,
                CAST(:client_ip AS inet), :consent_policy_version, :consent_text_snapshot,
                :expires_at
            )
            RETURNING CAST(id AS text), created_at
            """
            ),
            {
                "estimate_id": str(payload.estimate_id) if payload.estimate_id else None,
                "phone": payload.phone,
                "consent": payload.consent,
                "source": payload.source,
                "user_agent": user_agent,
                "client_ip": client_ip,
                "consent_policy_version": _CONSENT_POLICY_VERSION,
                "consent_text_snapshot": _CONSENT_TEXT_SNAPSHOT,
                "expires_at": expires_at,
            },
        )
        .mappings()
        .one()
    )

    db.commit()

    logger.info(
        "trade_in_lead saved id=%s estimate_id=%s source=%s ip=%s policy=%s",
        row["id"],
        payload.estimate_id,
        payload.source,
        client_ip,
        _CONSENT_POLICY_VERSION,
    )

    return {
        "id": row["id"],
        "created_at": row["created_at"].isoformat(),
        "status": "received",
    }