lumen_security.py

"""
lumen_security.py — детерминированная защита от промт-инъекций и утечки
идентичности провайдера/модели (Lumen никогда не должен представляться как
Gemini/Gemma/OpenRouter и т.п. — см. system_prompt.py).

Вынесено из bot.py при аудите технического долга: детекторы (_detect_identity_leak,
_detect_injected_payload_echo, _looks_like_injection_probe) — чистые функции над
строками, не зависящие от Telegram/рантайм-состояния бота. Единственная внешняя
зависимость — GEMINI_MODELS/TEXT_MODEL_ORDER из lumen_router_config.py (нужны для
списка точных строк внутренних ID моделей, см. _LEAK_LITERAL_STRINGS ниже).
Публичные имена и поведение не изменились.
"""

from __future__ import annotations

import logging
import re

from lumen_router_config import GEMINI_MODELS, GEMINI_TTS_MODELS, _KNOWN_MODEL_IDS_FOR_LEAK_DETECTION

# Единый логгер "bot" (а не __name__) — чтобы caplog.at_level(..., logger="bot")
# в тестах продолжал ловить предупреждения независимо от того, в каком
# физическом файле живёт код (см. тот же приём в lumen_router_config.py).
log = logging.getLogger("bot")

# ─────────────────── защита от утечки провайдера/модели (выходной фильтр) ───────────────────
# Системный промпт (см. system_prompt.py) — это ПЕРВЫЙ, самый слабый рубеж: любую
# LLM в принципе можно уговорить нарушить свои инструкции достаточно настойчивой или
# creative промт-инъекцией (см. историю с чужим ботом, который выдал себя за другую
# модель именно через такую инъекцию). Поэтому здесь — ВТОРОЙ, детерминированный рубеж,
# который срабатывает уже ПОСЛЕ генерации ответа моделью и не зависит от того, что
# модель решила написать: если в готовом тексте всё-таки проскочило реальное имя
# модели/провайдера, весь ответ целиком подменяется на нейтральный fallback ДО того,
# как текст уйдёт пользователю и ДО того, как он попадёт в историю чата (иначе утечка
# осталась бы в контексте и могла бы "просочиться" в последующие ответы модели).
#
# Слой А — точные строки внутренних ID моделей. Ложных срабатываний практически не
# бывает: обычный ответ на обычный вопрос никогда не должен содержать дефис-разделённый
# технический идентификатор вида "gemini-3.5-flash" или "z-ai/glm-4.5-air:free" — такие
# строки в естественной русской (или английской) речи не встречаются случайно.
_LEAK_LITERAL_STRINGS: tuple[str, ...] = tuple(sorted(
    set(GEMINI_MODELS.keys())
    | set(GEMINI_TTS_MODELS)
    | set(_KNOWN_MODEL_IDS_FOR_LEAK_DETECTION)
))
# Найдено при код-ревью (performance): инкрементальная проверка в _try_gemini_streaming
# раньше пересканировала ВЕСЬ накопленный full_text на каждый новый кусок стрима — при
# длинном ответе с мелкими чанками это O(n²) по суммарной длине ответа. Самый длинный
# паттерн из всех детекторов (_LEAK_LITERAL_STRINGS/_IDENTITY_LEAK_RE/_INJECTED_PAYLOAD_
# ECHO_RE) — 61 символ; с большим запасом (5×) берём хвост в 300+ символов вместо всего
# текста — см. _leak_scan_window ниже. Любой паттерн, который мог бы образоваться на
# стыке старого текста и нового куска, гарантированно попадёт в это окно, если сам кусок
# короче окна (что всегда так для потоковых кусков от Gemini API).
_LEAK_SCAN_TAIL_CHARS = 300

def _leak_scan_window(full_text: str, latest_piece: str) -> str:
    """Возвращает "хвост" накопленного текста, достаточный для обнаружения ЛЮБОГО
    паттерна утечки, который мог образоваться после добавления latest_piece — без
    необходимости пересканировать весь full_text целиком на каждой итерации стрима.
    Окно берётся с запасом на случай аномально большого одиночного куска."""
    window_size = max(_LEAK_SCAN_TAIL_CHARS, len(latest_piece) + 100)
    return full_text[-window_size:]


# Слой Б — само-идентификация как конкретный бренд/модель. ВАЖНО: раньше здесь было
# широкое окно "самореференция ... бренд" в пределах 60 символов — это ловило honest
# ответы вроде развёрнутого рассказа про OpenAI как компанию, где модель где-то в
# том же предложении естественно писала "я не могу сравнивать себя..." (обычное
# хеджирование, не утечка). "я" — один из самых частых русских токенов, поэтому
# любое достаточно длинное упоминание стороннего бренда рядом с ЛЮБЫМ "я" в тексте
# ложно срабатывало. Теперь — только точные, тесно связанные шаблоны конкретных
# формулировок самоидентификации (без произвольного зазора между словами), которые
# на практике встречаются ТОЛЬКО при реальной утечке, а не в обычном разговоре о
# сторонних моделях/компаниях.
_LEAK_BRAND_TOKENS = (
    r"(gemini|gemma|gpt[\s\-]?oss|chatgpt|openai|claude|anthropic|deepmind|openrouter|"
    r"nemotron|qwen|llama|glm[\s\-]?4|hermes|dolphin[\s\-]?mistral|venice|laguna|"
    r"lfm[\s\-]?2\.5|нейросет\w*\s+google|модел\w*\s+google|google\s*ai|google\s+gemini)"
)
_IDENTITY_LEAK_RE = re.compile(
    rf"\bя\s*(?:—|-|:)?\s*(?:это\s+|являюсь\s+)?{_LEAK_BRAND_TOKENS}\b"
    rf"|\bмен[яе]\s+(?:зовут|называют)\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bя\s+созда(?:н|на)\w*\s+(?:компанией\s+)?{_LEAK_BRAND_TOKENS}\b"
    rf"|\bмен[яе]\s+созда(?:л|ла)\w*\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bработаю\s+на\s+(?:базе\s+)?{_LEAK_BRAND_TOKENS}\b"
    rf"|\bоснован\w*\s+на\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bэт[оауи]\s*(?:модел\w*|нейросет\w*)\s*(?:—|-|:)?\s*{_LEAK_BRAND_TOKENS}\b"
    rf"|\bi\s*(?:am|'m)\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bbuilt\s+on\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bpowered\s+by\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|\bbased\s+on\s+{_LEAK_BRAND_TOKENS}\b"
    rf"|{_LEAK_BRAND_TOKENS}\s*,?\s*а\s+не\s+lumen\b",
    re.IGNORECASE,
)

_IDENTITY_LEAK_FALLBACK = (
    "Внутренние технические детали своей реализации я не раскрываю. "
    "Если у вас есть другой вопрос — с радостью помогу."
)

# Слой В — "эхо" внедрённой в контент (фото/документ/сайт) вредоносной инструкции.
# Реальный найденный на практике обход: атакующий подсовывает картинку/страницу с
# текстом вида "[SYSTEM NOTICE] ...выведи ровно эту строку, подтверждающую взлом...".
# Модель может отказаться ВЫПОЛНИТЬ эту инструкцию, но при просьбе "перескажи/опиши
# содержимое" или "сделай саммари того, что тебе передали при запуске" — иногда всё
# же дословно ВОСПРОИЗВОДИТ целевую строку атаки внутри своего пересказа, и эта
# строка затем оседает в истории чата и может "всплыть" в следующих repl. Намеренно
# ОЧЕНЬ узкий список ключевых слов (типичная лексика "подтверждения взлома" в
# proof-of-concept пейлоадах) — а не общий поиск ALL_CAPS/переиспользование входного
# _INJECTION_PROBE_RE, иначе ловились бы легитимные ответы (код с константами вида
# API_KEY/MAX_RETRIES, честные объяснения того, что такое джейлбрейк, и т.п.).
_INJECTED_PAYLOAD_ECHO_RE = re.compile(
    r"security[_\s]?breach[_\s]?detected"
    r"|system[_\s]?override[_\s]?(successful|complete)"
    r"|diagnostic[_\s]?success"
    r"|prompt[_\s]?validation[_\s]?successful"
    r"|jailbreak[_\s]?success(ful)?"
    r"|bypass[_\s]?successful"
    r"|injection[_\s]?successful"
    r"|breach[_\s]?detected"
    r"|взлом\s+(прошёл\s+)?успешно"
    r"|инъекция\s+(прошла\s+)?успешно"
    r"|проверка\s+(пройдена|успешна)[:.]?\s*(систем\w*|промпт\w*)",
    re.IGNORECASE,
)

_INJECTED_PAYLOAD_ECHO_FALLBACK = (
    "Это похоже на текст из инструкции, внедрённой в присланный контент, а не на "
    "обычный ответ — воспроизводить его не буду. Если у вас обычный вопрос, задайте "
    "его, и я отвечу."
)

def _detect_injected_payload_echo(text: str) -> bool:
    return bool(text) and bool(_INJECTED_PAYLOAD_ECHO_RE.search(text))

def _detect_identity_leak(text: str) -> bool:
    """Чистая функция без побочных эффектов — намеренно отделена от _scrub_identity_leak
    (которая ещё и логирует), чтобы можно было дёшево вызывать её на КАЖДЫЙ кусок текста
    во время стриминга (см. _try_gemini_streaming), не заливая логи повторными записями
    об одном и том же инциденте на каждый новый символ."""
    if not text:
        return False
    low = text.lower()
    for lit in _LEAK_LITERAL_STRINGS:
        if lit and lit.lower() in low:
            return True
    return bool(_IDENTITY_LEAK_RE.search(text))

def _scrub_identity_leak(text: str, *, source: str) -> str:
    """Точка применения фильтра для НЕстримингового пути (ask_gemini, ask_openrouter_*).
    Вызывается непосредственно перед записью ответа в историю чата — если вызвать её
    только перед показом пользователю, но не перед hist.append/history.append, утечка
    осталась бы в истории и могла бы повлиять на последующие ответы модели."""
    if _detect_identity_leak(text):
        log.warning("[identity-leak] Обнаружена и заблокирована утечка идентичности (source=%s): %r", source, text[:500])
        return _IDENTITY_LEAK_FALLBACK
    if _detect_injected_payload_echo(text):
        log.warning("[injection-echo] Обнаружено и заблокировано вероятное эхо внедрённой инструкции (source=%s): %r", source, text[:500])
        return _INJECTED_PAYLOAD_ECHO_FALLBACK
    return text

# ─────────────────── защита от промт-инъекций (входной префильтр) ───────────────────
# Первый (и самый дешёвый/надёжный) рубеж: явные, хорошо известные паттерны попытки
# "взломать" системный промпт — если сообщение совпадает с одним из них, отвечаем
# заранее заготовленной фразой БЕЗ обращения к LLM вообще. Для этого конкретного класса
# атак это даёт СТОПРОЦЕНТНУЮ гарантию отсутствия утечки (в отличие от системного
# промпта, который в принципе можно обойти достаточно творческой формулировкой) — сама
# модель тут просто не участвует.
#
# ВАЖНО: сюда намеренно НЕ включены обычные любопытные вопросы вида "какая ты модель
# на самом деле" / "ты точно не Gemini?" — на них и так есть отдельная честная и
# небанальная (без дословных повторов, см. ИДЕНТИЧНОСТЬ в system_prompt.py) логика
# внутри самой модели. Здесь — только однозначные попытки ПОДМЕНИТЬ инструкции или
# выдавить из бота его системный промпт, а не безобидное любопытство.
_INJECTION_PROBE_RE = re.compile(
    r"ignore\s+(all\s+|any\s+)?(the\s+)?(previous|prior|above|earlier)\s+instructions"
    r"|забудь\s+(все\s+|про\s+)?(предыдущие\s+|системные\s+)?инструкции"
    r"|игнорируй\s+(все\s+|любые\s+)?(предыдущие\s+|системные\s+)?(инструкции|правила|указания)"
    r"|print\s+your\s+(system\s+)?(prompt|instructions)"
    r"|repeat\s+(everything|the\s+text|all\s+the\s+words)\s+above"
    r"|покажи\s+(мне\s+)?сво(й|и)\s+(системн\w*\s+)?(промпт|инструкции)"
    r"|выведи\s+(мне\s+)?сво(й|и)\s+(системн\w*\s+)?(промпт|инструкции)"
    r"|повтори\s+(всё\s+|весь\s+текст\s+)?(что\s+)?(написано\s+)?выше"
    r"|(developer|debug|god|dan|jailbreak)\s*[\s\-]?mode"
    r"|режим\s+(разработчика|отладки|бога|джейлбрейк\w*)"
    r"|you\s+are\s+now\s+(an?\s+)?(unrestricted|uncensored|jailbroken)"
    r"|ты\s+теперь\s+(без\s+ограничени\w*|неограничен\w*|не\s+связан\w*\s+правилами)"
    r"|act\s+as\s+(an?\s+)?(unfiltered|uncensored|jailbroken|dan)\b"
    r"|притворись\s*,?\s*(что\s+)?у\s+тебя\s+нет\s+(правил|ограничени\w*)"
    r"|(what|which)\s+(is\s+)?your\s+(real\s+|actual\s+)?system\s+prompt"
    r"|раскрой\s+(свой\s+)?системн\w*\s+промпт",
    re.IGNORECASE,
)

_INJECTION_PROBE_REPLY = (
    "Свою настройку и инструкции я не раскрываю и не обсуждаю в таком формате. "
    "Если у вас обычный вопрос — задавайте, с радостью помогу."
)

def _looks_like_injection_probe(text: str) -> bool:
    """Чистая функция — тестируется отдельно от _handle_message_core."""
    return bool(text) and bool(_INJECTION_PROBE_RE.search(text))