backend/app/api/v1/chat.py

@@ -63,7 +63,10 @@ async def ask(
         оркестратор сам отдаёт детерминированный ответ (llm_used=False + fallback_reason).
     """
     try:
-        report, run_id = get_report_for_chat(db, payload.cad_num)
+        # Sync SQLAlchemy чтение (тянет крупный JSONB §22-отчёт) — мостим через
+        # run_in_threadpool, чтобы НЕ блокировать event loop (тот же приём, что и
+        # синхронный LLM `complete` в _answer_via_llm ниже).
+        report, run_id = await run_in_threadpool(get_report_for_chat, db, payload.cad_num)
     except Exception:
         # Read-only сбой БД — не валим клиента 500-кой, отдаём pending (как
         # get_parcel_forecast). Клиент может повторить.

backend/app/services/exporters/excel.py

@@ -53,6 +53,11 @@ _ADVISORY_MARKER: str = (
 )
 # Округление чисел по умолчанию (скоры/индексы ∈ [0,1] и доли — 3 знака читаемо).
 _ROUND_DIGITS: int = 3
+# OWASP CSV/formula-injection: лидирующие символы, с которых Excel/LibreOffice трактуют
+# строку как ФОРМУЛУ/команду при открытии. Внешние/скрейпинг-данные (comm_name из DOM.РФ,
+# headline/usp_text и пр.) с таким префиксом нейтрализуем (см. `_fmt`).
+# https://owasp.org/www-community/attacks/CSV_Injection
+_FORMULA_INJECTION_PREFIXES: tuple[str, ...] = ("=", "+", "-", "@", "\t", "\r")
 # Основной продуктовый горизонт (мес) — из него тянем сводный deficit_index сценария
 # (зеркало report_assembler._PRIMARY_HORIZON_MONTHS). ScenarioForecast.as_dict() несёт
 # список forecasts по горизонтам, скалярного «overall» у сценария НЕТ — берём дефицит
@@ -103,12 +108,27 @@ def _normalize(report: Any) -> dict[str, Any]:
     return {}
 
 
+def _sanitize_formula(text: str) -> str:
+    """Нейтрализовать formula/DDE-injection: префикс `'` строкам-«формулам». PURE.
+
+    openpyxl сохраняет строку с лидирующим `= + - @ \\t \\r` как ФОРМУЛУ (data_type='f'),
+    и Excel/LibreOffice исполнит её при открытии. Внешние/скрейпинг-данные (comm_name из
+    DOM.РФ, headline/usp_text/advisory и пр.) попадают в ячейки через `_fmt`, поэтому
+    экранируем по OWASP-набору: ведём строку апострофом — Excel показывает литеральный
+    текст и трактует ячейку как текст, формула не исполняется. Безопасные строки — как есть.
+    """
+    if text.startswith(_FORMULA_INJECTION_PREFIXES):
+        return "'" + text
+    return text
+
+
 def _fmt(value: Any) -> Any:
     """Привести значение к ячейке Excel: None → "—", float → округление. PURE.
 
     Числа округляем до `_ROUND_DIGITS` (но bool оставляем как есть — это не число для
-    округления). Строки/прочие скаляры — как есть. Контейнеры (dict/list) сворачиваем
-    в компактную строку (на случай неожиданно-вложенного значения — не падаем).
+    округления). Строки экранируем от formula/DDE-injection (`_sanitize_formula`) —
+    внешние/скрейпинг-данные с лидирующим `= + - @` не должны стать формулой. Контейнеры
+    (dict/list) сворачиваем в строку и тоже экранируем (на случай вложенного значения).
     """
     if value is None:
         return _DASH
@@ -116,10 +136,12 @@ def _fmt(value: Any) -> Any:
         return "да" if value else "нет"
     if isinstance(value, float):
         return round(value, _ROUND_DIGITS)
-    if isinstance(value, (str, int)):
+    if isinstance(value, str):
+        return _sanitize_formula(value)
+    if isinstance(value, int):
         return value
     if isinstance(value, (dict, list)):
-        return str(value)
+        return _sanitize_formula(str(value))
     return value
 
 
@@ -200,8 +222,12 @@ def _write_title(ws: Worksheet, row: int, text: str) -> int:
 
 
 def _write_kv(ws: Worksheet, row: int, label: str, value: Any) -> int:
-    """Строка «метка → значение» (label жирный, value через `_fmt`). PURE-side."""
-    label_cell = ws.cell(row=row, column=1, value=label)
+    """Строка «метка → значение» (label жирный, value через `_fmt`). PURE-side.
+
+    Метка тоже экранируется от formula-injection: часть меток — это КЛЮЧИ из секций
+    отчёта (`key_numbers`/`market_metrics`/`factors`…), потенциально внешнего происхождения.
+    """
+    label_cell = ws.cell(row=row, column=1, value=_sanitize_formula(label))
     label_cell.font = Font(bold=True)
     label_cell.alignment = _LEFT_TOP
     value_cell = ws.cell(row=row, column=2, value=_fmt(value))

backend/tests/api/v1/test_chat.py

@@ -227,6 +227,37 @@ def test_ask_no_run_returns_pending() -> None:
     assert "Запустите анализ участка" in body["answer"]
 
 
+def test_ask_report_read_bridged_through_threadpool(monkeypatch: pytest.MonkeyPatch) -> None:
+    """#1245: синхронное чтение §22-отчёта мостится через run_in_threadpool (не в loop).
+
+    `get_report_for_chat` — sync SQLAlchemy (тянет крупный JSONB) и НЕ должен блокировать
+    event loop. Оборачиваем `run_in_threadpool` в шпион: проверяем, что чтение отчёта
+    реально диспатчится через него (как и синхронный LLM `complete` в _answer_via_llm).
+    """
+    from app.services.chat import retrieval as retrieval_mod
+
+    dispatched: list[str] = []
+    real_run_in_threadpool = chat_router.run_in_threadpool
+
+    async def _spy(func: Any, *args: Any, **kwargs: Any) -> Any:
+        dispatched.append(getattr(func, "__name__", repr(func)))
+        return await real_run_in_threadpool(func, *args, **kwargs)
+
+    monkeypatch.setattr(chat_router, "run_in_threadpool", _spy)
+    client = _client_with_db()
+    with patch(
+        "app.services.chat.retrieval.latest_run_for",
+        return_value=_make_run(_report()),
+    ):
+        resp = client.post(
+            "/api/v1/chat/ask",
+            json={"cad_num": _CAD, "message": "резюме"},
+        )
+    assert resp.status_code == 200, resp.text
+    # Синхронное чтение отчёта прошло через threadpool (имя sync-функции зафиксировано).
+    assert retrieval_mod.get_report_for_chat.__name__ in dispatched
+
+
 def test_ask_db_error_returns_pending_not_500() -> None:
     """Сбой БД на read-only → pending (не 500), клиент может повторить."""
     client = _client_with_db()

backend/tests/services/exporters/test_excel.py

@@ -31,6 +31,7 @@ from app.services.exporters.excel import (
     _SHEET_SCORING,
     _SHEET_SUMMARY,
     _fmt,
+    _sanitize_formula,
     _scenario_deficit_index,
     export_report_xlsx,
 )
@@ -403,3 +404,72 @@ class TestGracefulPartialReport:
             assert len(payload) > 0
             wb = _reload(payload)
             assert _SHEET_SUMMARY in wb.sheetnames
+
+
+# ── #1244: formula/DDE-injection — внешние строки не должны стать формулой Excel ──
+# openpyxl сохраняет строку с лидирующим `= + - @ \t \r` как data_type='f' (формула),
+# Excel/LibreOffice исполнит её при открытии. Внешние/скрейпинг-поля (comm_name из
+# DOM.РФ, headline/usp_text/advisory) попадают в ячейки через `_fmt` → должны быть
+# нейтрализованы префиксом `'` (OWASP CSV-injection) и трактоваться как ТЕКСТ.
+
+
+class TestFormulaInjection:
+    def test_sanitize_prefixes_dangerous_strings(self) -> None:
+        # Каждый OWASP-префикс получает ведущий апостроф.
+        for payload in ("=SUM(A1:A2)", "+1+1", "-2+3", "@cmd", "\tTAB", "\rCR"):
+            assert _sanitize_formula(payload) == "'" + payload
+
+    def test_sanitize_leaves_safe_strings_untouched(self) -> None:
+        for safe in ("ЖК Альфа", "комфорт", "250 000", "0.34", ""):
+            assert _sanitize_formula(safe) == safe
+
+    def test_fmt_escapes_formula_string(self) -> None:
+        # Строка-«формула» через _fmt → текст с апострофом (не исполняемая формула).
+        assert _fmt("=1+1") == "'=1+1"
+        assert _fmt("@SUM(1)") == "'@SUM(1)"
+
+    def test_fmt_does_not_touch_numbers_or_safe_text(self) -> None:
+        # Числа/безопасный текст не трогаем: отрицательный float — это значение, не строка.
+        assert _fmt(-0.5) == -0.5
+        assert _fmt(-3) == -3
+        assert _fmt("ЖК Бета") == "ЖК Бета"
+
+    def test_injected_competitor_name_written_as_text_not_formula(self) -> None:
+        # comm_name из скрейпинга DOM.РФ с лидирующим `=` → ячейка text, не формула.
+        report = SiteFinderReport(
+            market_now=ReportMarketNow(
+                competitors=[
+                    {"obj_id": 1, "comm_name": "=cmd|'/c calc'!A1", "relevance_weight": 0.8}
+                ],
+            ),
+        )
+        ws = _reload(export_report_xlsx(report))[_SHEET_MARKET_NOW]
+        injected = [
+            c
+            for col in ws.iter_cols()
+            for c in col
+            if isinstance(c.value, str) and "calc" in c.value
+        ]
+        assert injected, "имя конкурента должно присутствовать в книге"
+        for cell in injected:
+            assert cell.data_type == "s", "ячейка должна быть текстом, не формулой ('f')"
+            assert cell.value.startswith("'="), "опасный префикс должен быть экранирован"
+
+    def test_injected_headline_and_usp_written_as_text(self) -> None:
+        # headline (Сводка) и usp_text (Продукт ТЗ) — тоже внешние → не формула.
+        report = SiteFinderReport(
+            exec_summary=ReportExecSummary(headline="=HYPERLINK('http://evil','x')"),
+            product_tz=ReportProductTz(
+                usp=[{"segment": "1-Студия", "usp_text": "+evil()"}],
+            ),
+        )
+        payload = export_report_xlsx(report)
+        wb = _reload(payload)
+        formula_cells = [
+            c
+            for ws in wb.worksheets
+            for col in ws.iter_cols()
+            for c in col
+            if c.data_type == "f"
+        ]
+        assert not formula_cells, f"ни одна ячейка не должна быть формулой: {formula_cells}"