bot-backend
14f3ef2019
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
231 lines
15 KiB
Python
231 lines
15 KiB
Python
"""§19 data-residency redaction layer (#960) — THE critical safety boundary.
|
||
|
||
Провайдер LLM ВНЕШНИЙ (OpenAI) → данные покидают РФ. Чувствительные данные не
|
||
должны попасть наружу НИКОГДА. Защита эшелонирована (defense-in-depth):
|
||
|
||
1. Allowlist / safe-payload (ПЕРВИЧНОЕ): публичный API заставляет вызывающего
|
||
явно собрать ``SafePayload`` из не-чувствительных полей/агрегатов/публичного
|
||
текста — а НЕ передавать сырые строки БД. См. ``SafePayload`` ниже и контракт
|
||
в ``app/services/llm/__init__.py``.
|
||
2. Hard-block (ENFORCED): ``SafePayload(is_confidential=True)`` → ``RedactionRefusedError``.
|
||
Контент, помеченный конфиденциальным (напр. insight.is_confidential), НИКОГДА
|
||
не уходит во внешнего провайдера — клиент поднимает типизированную ошибку, а не
|
||
«тихо» отправляет.
|
||
3. Regex-scrub (ВТОРИЧНОЕ, belt-and-suspenders): из свободного текста вырезаются
|
||
RU PII-паттерны (телефон / email / СНИЛС / ИНН / паспорт / ФИО-подобное) и
|
||
заменяются на ``[REDACTED:<kind>]``. Это НЕ основная гарантия — основная это (1).
|
||
|
||
``scrub_safe_payload`` — единственная точка, через которую ``client.complete``
|
||
прогоняет данные перед внешним вызовом (mandatory на external-пути; будущий
|
||
RU-hosted провайдер сможет это ослабить — см. шов ``LLMProvider.is_external``).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RedactionRefusedError(Exception):
|
||
"""Полезная нагрузка помечена конфиденциальной → отправка наружу запрещена.
|
||
|
||
Поднимается из ``scrub_safe_payload`` для external-провайдера, когда
|
||
``SafePayload.is_confidential`` is True. Клиент ловит это и превращает в
|
||
детерминированный fallback (система не падает из-за того что данные нельзя слать).
|
||
"""
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class SafePayload:
|
||
"""Явно собранная вызывающим НЕ-чувствительная нагрузка для внешней LLM.
|
||
|
||
Контракт (allowlist-first): консьюмеры (#956 граддок-extraction, #957 chat)
|
||
ОБЯЗАНЫ конструировать это вручную из проверенных полей — НЕ передавать сырые
|
||
DB-строки/insight'ы. Структура — единственное, что физически доходит до клиента.
|
||
|
||
Attributes:
|
||
text: Свободный текст для модели (публичный градрегламент, агрегаты, вопрос
|
||
пользователя). Прогоняется через regex-scrub перед отправкой.
|
||
fields: Структурированные не-чувствительные пары ключ→значение (числа/коды/
|
||
короткие строки). Значения-строки тоже скрабятся.
|
||
is_confidential: Если True — вся нагрузка под жёстким блоком: ``scrub_safe_payload``
|
||
для external-провайдера поднимет ``RedactionRefusedError``. Прокидывать сюда
|
||
ЛЮБОЙ confidential-флаг источника (напр. insight.is_confidential).
|
||
"""
|
||
|
||
text: str = ""
|
||
fields: dict[str, Any] = field(default_factory=dict)
|
||
is_confidential: bool = False
|
||
|
||
|
||
# ── RU PII regex (вторичная защита) ──────────────────────────────────────────
|
||
# Порядок применения важен: более специфичные/длинные паттерны раньше, чтобы они не
|
||
# «съедались» более общими (СНИЛС/ИНН/паспорт раньше «голых» групп цифр).
|
||
|
||
# СНИЛС: NNN-NNN-NNN NN (последняя пара — через пробел/дефис).
|
||
_SNILS_RE = re.compile(r"\b\d{3}-\d{3}-\d{3}[\s-]\d{2}\b")
|
||
# Паспорт РФ: NNNN NNNNNN (серия 4 + номер 6, разделитель пробел/дефис).
|
||
_PASSPORT_RE = re.compile(r"\b\d{4}[\s-]\d{6}\b")
|
||
# Телефон РФ: +7/8, опц. скобки/пробелы/дефисы, 10 значащих цифр.
|
||
_PHONE_RE = re.compile(r"(?:\+7|\b8)[\s\-(]*\d{3}[\s\-)]*\d{3}[\s-]*\d{2}[\s-]*\d{2}\b")
|
||
# Телефон РФ «голый»: 11 цифр без разделителей, начинаются с 7 или 8 (#1207).
|
||
# Покрывает copy-paste форму «79221234567» / «89221234567», которую _PHONE_RE
|
||
# пропускает из-за требования префикса «+7»/«\b8» + хотя бы одного разделителя.
|
||
# Ставится РАНЬШЕ _SNILS_BARE_RE (любые 11 цифр), чтобы не путать с СНИЛС.
|
||
_PHONE_BARE_RE = re.compile(r"(?<!\d)[78]\d{10}(?!\d)")
|
||
# Телефон РФ «локальный»: 10 значащих цифр БЕЗ кода страны, начинаются с мобильного
|
||
# «9», сгруппированы 9XX XXX XX XX через пробел/дефис (#1641). _PHONE_RE требует
|
||
# префикс +7/8, _PHONE_BARE_RE — ровно 11 слитных цифр → формат «922 123 45 67» /
|
||
# «922-123-45-67» проходил мимо. Якорь «9» + фиксированная группировка 3-3-2-2 не
|
||
# пересекается с паспортом (4+6) / СНИЛС (3-3-3 2). Ставится ПОСЛЕ префиксных форм,
|
||
# чтобы local-часть «+7 912 …» сначала ушла как полноценный phone.
|
||
_PHONE_LOCAL_RE = re.compile(r"(?<!\d)9\d{2}[\s-]\d{3}[\s-]\d{2}[\s-]\d{2}(?!\d)")
|
||
# Email.
|
||
_EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b")
|
||
# ИНН-кандидат: ровно 12 (физлицо) или 10 (юрлицо) цифр, не приклеенные к другим
|
||
# цифрам. ВНИМАНИЕ: regex здесь — лишь поиск кандидата. Голый «N{10}|N{12}» ловил бы и
|
||
# легитимные слитные суммы (1 млрд+ руб = 10 цифр, ~трлн = 12) и молча затирал бы их
|
||
# до отправки в LLM (#1640). Поэтому замена проходит ТОЛЬКО если кандидат проходит
|
||
# официальную проверку контрольных цифр ИНН (_inn_checksum_valid) — см. _inn_repl.
|
||
_INN_RE = re.compile(r"(?<!\d)(?:\d{12}|\d{10})(?!\d)")
|
||
|
||
|
||
def _inn_checksum_valid(digits: str) -> bool:
|
||
"""Проверить контрольные цифры ИНН (официальный алгоритм ФНС).
|
||
|
||
10-значный (юрлицо): одна контрольная цифра. 12-значный (физлицо/ИП): две.
|
||
Вес-коэффициенты фиксированы стандартом. Любая слитная группа 10/12 цифр, НЕ
|
||
проходящая checksum (напр. круглая сумма «1000000000»), считается НЕ-ИНН и не
|
||
редактируется — это и закрывает ложные срабатывания #1640.
|
||
|
||
NB: checksum резко снижает false-positive rate, но не доводит его до нуля —
|
||
~1/11 случайных 10-значных чисел совпадает с валидным ИНН по контрольной цифре.
|
||
Это приемлемо для вторичной (belt-and-suspenders) защиты.
|
||
"""
|
||
if len(digits) == 10:
|
||
weights = (2, 4, 10, 3, 5, 9, 4, 6, 8)
|
||
control = sum(int(digits[i]) * weights[i] for i in range(9)) % 11 % 10
|
||
return control == int(digits[9])
|
||
if len(digits) == 12:
|
||
w1 = (7, 2, 4, 10, 3, 5, 9, 4, 6, 8)
|
||
w2 = (3, 7, 2, 4, 10, 3, 5, 9, 4, 6, 8)
|
||
c1 = sum(int(digits[i]) * w1[i] for i in range(10)) % 11 % 10
|
||
c2 = sum(int(digits[i]) * w2[i] for i in range(11)) % 11 % 10
|
||
return c1 == int(digits[10]) and c2 == int(digits[11])
|
||
return False
|
||
|
||
|
||
def _inn_repl(match: re.Match[str]) -> str:
|
||
"""re.sub-callback: редактировать кандидат ТОЛЬКО если checksum валиден."""
|
||
token = match.group(0)
|
||
return "[REDACTED:inn]" if _inn_checksum_valid(token) else token
|
||
|
||
|
||
# СНИЛС «голый»: ровно 11 цифр без разделителей (#1207). _SNILS_RE требует
|
||
# формат «NNN-NNN-NNN NN»; raw «12345678901» проходит мимо. По длине не пересекается
|
||
# с ИНН (10/12); пересекается с _PHONE_BARE_RE (тоже 11 цифр), поэтому идёт ПОСЛЕ
|
||
# него — phone-bare съест 7/8-префиксы, СНИЛС-bare добьёт остаток.
|
||
_SNILS_BARE_RE = re.compile(r"(?<!\d)\d{11}(?!\d)")
|
||
# ФИО-подобное: три слова с заглавной кириллической буквы подряд (Фамилия Имя Отчество).
|
||
# Каждое слово — либо «Иванов» (Titlecase), либо «ИВАНОВ» (CAPS, мин 2 буквы) — последнее
|
||
# покрывает стандарт ЕГРН/паспортных выписок «ИВАНОВ ИВАН ИВАНОВИЧ» (#1207).
|
||
# Грубая эвристика — может задеть «Министерство Строительства России» и подобное; это
|
||
# приемлемо для вторичной защиты (лучше лишний REDACTED, чем утечка ФИО).
|
||
_FULLNAME_RE = re.compile(
|
||
r"\b(?:[А-ЯЁ][а-яё]+|[А-ЯЁ]{2,})"
|
||
r"\s+(?:[А-ЯЁ][а-яё]+|[А-ЯЁ]{2,})"
|
||
r"\s+(?:[А-ЯЁ][а-яё]+|[А-ЯЁ]{2,})\b"
|
||
)
|
||
|
||
# (regex, kind, repl). ``repl`` — строка-плейсхолдер ИЛИ callback для re.subn
|
||
# (используется ИНН: редактирует только checksum-валидные кандидаты — #1640).
|
||
# Применяются последовательно в этом порядке.
|
||
# Порядок критичен: _PHONE_BARE_RE раньше _SNILS_BARE_RE, чтобы 11-значные
|
||
# с префиксом 7/8 ушли как phone (телефон семантически точнее СНИЛС'а).
|
||
_Repl = str | Callable[[re.Match[str]], str]
|
||
_PII_PATTERNS: tuple[tuple[re.Pattern[str], str, _Repl], ...] = (
|
||
(_SNILS_RE, "snils", "[REDACTED:snils]"),
|
||
(_PASSPORT_RE, "passport", "[REDACTED:passport]"),
|
||
(_PHONE_RE, "phone", "[REDACTED:phone]"),
|
||
(_PHONE_BARE_RE, "phone", "[REDACTED:phone]"),
|
||
(_PHONE_LOCAL_RE, "phone", "[REDACTED:phone]"),
|
||
(_EMAIL_RE, "email", "[REDACTED:email]"),
|
||
(_INN_RE, "inn", _inn_repl),
|
||
(_SNILS_BARE_RE, "snils", "[REDACTED:snils]"),
|
||
(_FULLNAME_RE, "name", "[REDACTED:name]"),
|
||
)
|
||
|
||
|
||
def scrub_text(value: str) -> str:
|
||
"""Вырезать RU PII из свободного текста → ``[REDACTED:<kind>]``.
|
||
|
||
Вторичная защита (см. модульный docstring). НИКОГДА не логирует ни исходный, ни
|
||
результирующий текст — только факт срабатывания и kind (без значения).
|
||
"""
|
||
if not value:
|
||
return value
|
||
redacted = value
|
||
placeholder = "[REDACTED:%s]"
|
||
for pattern, kind, repl in _PII_PATTERNS:
|
||
# n из subn для callback-repl (ИНН) считает ВСЕ совпадения, включая кандидаты,
|
||
# которые callback вернул без изменений (не прошли checksum). Поэтому реальное
|
||
# число замен берём по приросту числа плейсхолдеров — корректно и для str, и
|
||
# для callback, без утечки самого PII-значения в лог.
|
||
before_count = redacted.count(placeholder % kind)
|
||
redacted = pattern.sub(repl, redacted)
|
||
n = redacted.count(placeholder % kind) - before_count
|
||
if n > 0:
|
||
# Логируем ТОЛЬКО kind и количество — без самого PII-значения.
|
||
logger.info("redaction: scrubbed %d %s token(s) from free text", n, kind)
|
||
return redacted
|
||
|
||
|
||
def _scrub_value(value: Any) -> Any:
|
||
"""Рекурсивно проскрабить строковые значения в структуре fields."""
|
||
if isinstance(value, str):
|
||
return scrub_text(value)
|
||
if isinstance(value, dict):
|
||
return {k: _scrub_value(v) for k, v in value.items()}
|
||
if isinstance(value, (list, tuple)):
|
||
scrubbed = [_scrub_value(v) for v in value]
|
||
return type(value)(scrubbed)
|
||
return value
|
||
|
||
|
||
def scrub_safe_payload(payload: SafePayload, *, is_external: bool) -> SafePayload:
|
||
"""Mandatory-шаг перед внешним вызовом: hard-block + regex-scrub.
|
||
|
||
Args:
|
||
payload: явно собранная вызывающим нагрузка.
|
||
is_external: True для провайдера, выгружающего данные за пределы РФ
|
||
(OpenAI). Для будущего RU-hosted провайдера можно передать False, чтобы
|
||
ослабить scrub — это и есть задокументированный шов (data остаётся в РФ).
|
||
|
||
Returns:
|
||
Новый ``SafePayload`` с проскрабленными ``text``/``fields``.
|
||
|
||
Raises:
|
||
RedactionRefusedError: если ``is_external`` и ``payload.is_confidential``.
|
||
"""
|
||
if is_external and payload.is_confidential:
|
||
# НЕ логируем содержимое — только факт отказа.
|
||
logger.warning("redaction: refusing confidential payload for external LLM provider")
|
||
raise RedactionRefusedError(
|
||
"payload marked is_confidential=True cannot be sent to an external LLM provider"
|
||
)
|
||
if not is_external:
|
||
# RU-hosted провайдер: данные не покидают РФ — scrub не обязателен.
|
||
return payload
|
||
return SafePayload(
|
||
text=scrub_text(payload.text),
|
||
fields=_scrub_value(payload.fields),
|
||
is_confidential=payload.is_confidential,
|
||
)
|
||
|
||
|
||
__all__ = ["RedactionRefusedError", "SafePayload", "scrub_safe_payload", "scrub_text"]
|