Close four security gaps: shared rate limit, proxy auth, subprocess cleanup, log redaction

bot.py

@@ -65,6 +65,29 @@ _LOG_QUEUE: queue.SimpleQueue[logging.LogRecord] = queue.SimpleQueue()
 _LOG_QUEUE_HANDLER = logging.handlers.QueueHandler(_LOG_QUEUE)
 _LOG_LISTENER: logging.handlers.QueueListener | None = None
 
+_SECRET_NAMES = (
+    "BOT_TOKEN", "TELEGRAM_TOKEN", "TELEGRAM_BOT_TOKEN", "GEMINI_API_KEY",
+    "OPENROUTER_API_KEY", "OPENROUTER_KEY", "ADMIN_SECRET_SEED",
+    "_ADMIN_SECRET_SEED", "WEBHOOK_SECRET", "ADMIN_PANEL_KEY",
+    "UPSTASH_REDIS_REST_TOKEN", "LUMEN_PROXY_SECRET",
+)
+
+
+def _current_log_secrets() -> tuple[str, ...]:
+    values = [value for name in _SECRET_NAMES
+              for value in (globals().get(name), os.getenv(name, ""))
+              if isinstance(value, str) and value]
+    return tuple(sorted(set(values), key=len, reverse=True))
+
+
+class _SecretLogFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        text = super().format(record)
+        for secret in _current_log_secrets():
+            text = text.replace(secret, "<REDACTED>")
+        return text
+
+
 def _setup_logging() -> logging.Logger:
     # LOG_LEVEL — раньше был захардкожен INFO везде (root+оба handler'а), из-за
     # чего оба существующих log.debug(...) в проекте не печатались никогда, ни в
@@ -87,6 +110,8 @@ def _setup_logging() -> logging.Logger:
     console_handler.setFormatter(fmt)
     file_handler.setFormatter(fmt)
 
+    _LOG_QUEUE_HANDLER.setFormatter(_SecretLogFormatter())
+
     global _LOG_LISTENER
     # НАЙДЕНО ПРИ АУДИТЕ (4 сентября 2026, реальный воспроизведённый инцидент — см.
     # py-spy дамп стеков зависшего процесса): _setup_logging() вызывается больше
@@ -198,6 +223,7 @@ def _normalize_telegram_base_url(url: str) -> str:
         url = "https://" + url
     return url
 
+LUMEN_PROXY_SECRET = os.getenv("LUMEN_PROXY_SECRET", "")
 TELEGRAM_API_BASE_URL = _normalize_telegram_base_url(os.getenv("TELEGRAM_API_BASE_URL", "https://api.telegram.org"))
 log.info("[setup] Using Telegram API Base URL: %s", TELEGRAM_API_BASE_URL)
 
@@ -290,10 +316,7 @@ def _redactable_secrets() -> tuple[str, ...]:
     сохранённым историям чатов. honey: имена читаются по значению на момент
     вызова — WEBHOOK_SECRET/ADMIN_PANEL_KEY/UPSTASH_REDIS_REST_TOKEN объявлены
     ниже по файлу, это безопасно для module-level globals в теле функции."""
-    return tuple(s for s in (
-        BOT_TOKEN, GEMINI_API_KEY, OPENROUTER_API_KEY, _ADMIN_SECRET_SEED,
-        WEBHOOK_SECRET, ADMIN_PANEL_KEY, UPSTASH_REDIS_REST_TOKEN,
-    ) if s)
+    return _current_log_secrets()
 
 def _sentry_scrub_secrets(event: dict, hint: dict) -> dict | None:
     """before_send-хук Sentry — вычищает секреты из события ПЕРЕД отправкой.
@@ -489,6 +512,7 @@ from lumen_telegram_transport import (
     _TelegramProxyCircuitBreaker,
     _looks_like_proxy_garbage,
     IPv4AiohttpSession,
+    proxy_auth_middlewares,
     get_telegram_session as _lumen_get_telegram_session,
     close_telegram_session as _lumen_close_telegram_session,
 )
@@ -497,7 +521,10 @@ TG_PROXY_TRIP_THRESHOLD = int(os.getenv("TG_PROXY_TRIP_THRESHOLD", "3"))
 _tg_proxy_breaker = _TelegramProxyCircuitBreaker(cooldown_sec=TG_PROXY_COOLDOWN_SEC, trip_threshold=TG_PROXY_TRIP_THRESHOLD)
 
 async def _get_telegram_session() -> aiohttp.ClientSession:
-    return await _lumen_get_telegram_session(TELEGRAM_REQUEST_TIMEOUT)
+    return await _lumen_get_telegram_session(
+        TELEGRAM_REQUEST_TIMEOUT,
+        proxy_secret=LUMEN_PROXY_SECRET, proxy_base_urls=_TELEGRAM_PROXY_CANDIDATES,
+    )
 
 async def _rotate_telegram_proxy() -> bool:
     """Переключается на следующий кандидат из _TELEGRAM_PROXY_CANDIDATES по кругу —
@@ -523,7 +550,10 @@ async def _rotate_telegram_proxy() -> bool:
     log.warning('[telegram] Switching to fallback proxy: %s -> %s', old_url, new_url)
     if bot is not None:
         old_session = bot.session
-        bot.session = IPv4AiohttpSession(api=TelegramAPIServer.from_base(new_url))
+        bot.session = IPv4AiohttpSession(
+            api=TelegramAPIServer.from_base(new_url),
+            proxy_secret=LUMEN_PROXY_SECRET, proxy_base_urls=_TELEGRAM_PROXY_CANDIDATES,
+        )
         with contextlib.suppress(Exception):
             await old_session.close()
     return _telegram_proxy_idx != 0
@@ -542,6 +572,10 @@ async def _get_http_session() -> aiohttp.ClientSession:
             # используют на порядок меньше одновременных соединений, так что
             # повышение лимита их не затрагивает.
             connector=aiohttp.TCPConnector(family=socket.AF_INET, limit=40, ttl_dns_cache=300),
+            middlewares=proxy_auth_middlewares(
+                proxy_secret=LUMEN_PROXY_SECRET,
+                proxy_base_urls=(*_TELEGRAM_PROXY_CANDIDATES, *_tikwm_proxy_candidates()),
+            ),
         )
     return _http_session
 
@@ -2421,6 +2455,7 @@ from lumen_tiktok import (
     _download_url_bin,
     _probe_video_dimensions,
     _generate_video_thumbnail,
+    _communicate_process,
     _probe_and_thumbnail_from_bytes,
     _resolve_tiktok_short,
     _fetch_tikwm_media_data,
@@ -3974,6 +4009,8 @@ async def cmd_draw(message: Message) -> None:
     if not prompt:
         await _safe_reply(message, "Укажите текст после команды /draw. Пример: /draw космическая станция")
         return
+    if await _reject_rate_limited_message(message):
+        return
     await inline_draw(message, prompt)
 
 # ─────────────────── TTS-пайплайн (Fish Audio + Gemini TTS) ───────────────────
@@ -4066,7 +4103,7 @@ async def inline_tts(message: Message, text: str) -> None:
                     stdout=asyncio.subprocess.DEVNULL,
                     stderr=asyncio.subprocess.DEVNULL,
                 )
-                await asyncio.wait_for(proc.communicate(), timeout=30)
+                await _communicate_process(proc, timeout=30)
                 if os.path.exists(dst_path) and os.path.getsize(dst_path) > 0:
                     with open(dst_path, "rb") as fh:
                         final_audio = fh.read()
@@ -4080,7 +4117,7 @@ async def inline_tts(message: Message, text: str) -> None:
                             stdout=asyncio.subprocess.PIPE,
                             stderr=asyncio.subprocess.DEVNULL,
                         )
-                        probe_out, _ = await asyncio.wait_for(probe.communicate(), timeout=10)
+                        probe_out, _ = await _communicate_process(probe, timeout=10)
                         raw_dur = probe_out.decode().strip()
                         voice_duration = max(1, round(float(raw_dur))) if raw_dur else 0
                     except Exception as probe_exc:
@@ -4119,6 +4156,8 @@ async def cmd_tts(message: Message) -> None:
     if not text:
         await _safe_reply(message, "Укажите текст после команды /tts. Пример: /tts Добрый день")
         return
+    if await _reject_rate_limited_message(message):
+        return
     await inline_tts(message, text)
 
 @dp.message(Command("reset"))
@@ -4537,6 +4576,13 @@ def _check_and_register_rate_limit(user_id: int | None) -> bool:
     return False
 
 
+async def _reject_rate_limited_message(message: Message) -> bool:
+    if not _check_and_register_rate_limit(_rate_limit_key_for_message(message)):
+        return False
+    await _tg_call(message.reply, "Вы отправляете слишком много запросов. Подождите немного.")
+    return True
+
+
 async def _resolve_incoming_media(
     message: Message, state: dict[str, Any], clean_prompt: str, *, is_private: bool,
 ) -> tuple[str | None, str, str, tuple[bytes, str] | None]:
@@ -4623,8 +4669,7 @@ async def _handle_message_core(message: Message, extra_media: list[tuple[bytes,
         return
 
     # Начинаем обработку активного запроса с проверкой rate limit
-    if _check_and_register_rate_limit(_rate_limit_key_for_message(message)):
-        await _tg_call(message.reply, "Вы отправляете слишком много запросов. Подождите немного.")
+    if await _reject_rate_limited_message(message):
         return
 
     # Проверка на ссылки загрузки (TikTok — сразу всегда, даже в группах без упоминания)
@@ -4981,9 +5026,14 @@ async def main() -> None:
     # Мы настраиваем Bot сессию с принудительным IPv4 и таймаутами для hg space
     if TELEGRAM_API_BASE_URL != "https://api.telegram.org":
         api_server = TelegramAPIServer.from_base(TELEGRAM_API_BASE_URL)
-        sess = IPv4AiohttpSession(api=api_server)
+        sess = IPv4AiohttpSession(
+            api=api_server,
+            proxy_secret=LUMEN_PROXY_SECRET, proxy_base_urls=_TELEGRAM_PROXY_CANDIDATES,
+        )
     else:
-        sess = IPv4AiohttpSession()
+        sess = IPv4AiohttpSession(
+            proxy_secret=LUMEN_PROXY_SECRET, proxy_base_urls=_TELEGRAM_PROXY_CANDIDATES,
+        )
     bot = Bot(token=BOT_TOKEN, session=sess)
     client = genai.Client(api_key=GEMINI_API_KEY)
 

docs/ENVIRONMENT.md

@@ -1,6 +1,6 @@
 # Environment variables
 
-Only `BOT_TOKEN` and `GEMINI_API_KEY` are required. Everything else has a working default; most deployments never need to touch the rest of this file.
+`BOT_TOKEN` and `GEMINI_API_KEY` are required. Deployments using the authenticated proxy also require `LUMEN_PROXY_SECRET` on both the bot and every proxy instance.
 
 ## Required
 
@@ -15,6 +15,7 @@ Hugging Face Spaces' outbound IPs are blocked by Telegram's API and rejected (`4
 
 | Variable | Default | Purpose |
 |---|---|---|
+| `LUMEN_PROXY_SECRET` | — | Required on every Deno proxy and the bot when using proxies. Use the same independently generated random secret (at least 32 random bytes encoded as hex) for primary and fallback instances. Sent only in `X-Lumen-Proxy-Secret`, never in URLs or `Authorization`. |
 | `TELEGRAM_API_BASE_URL` | `https://api.telegram.org` | Base URL for the Telegram Bot API. In production this points at the proxy. |
 | `TELEGRAM_API_BASE_URL_FALLBACKS` | — | Comma-separated backup proxy addresses. On a circuit-breaker trip the bot rotates through these before pausing. |
 | `TG_PROXY_COOLDOWN_SEC` | `20` | Pause (seconds) after the circuit breaker trips, i.e. the proxy is judged unavailable. |
@@ -22,6 +23,12 @@ Hugging Face Spaces' outbound IPs are blocked by Telegram's API and rejected (`4
 | `TIKWM_API_BASE_URL` | — (direct requests) | Base URL for a TikWM proxy. Empty means the bot talks to both `tikwm.com` mirrors directly. |
 | `TIKWM_API_BASE_URL_FALLBACKS` | — | Comma-separated backup TikWM proxies, tried in order if the primary one fails. |
 
+The proxy denies requests before forwarding: missing/wrong credentials return `401`; an unset, empty or invalid server secret returns `503`. Secrets must be printable ASCII without spaces. The header is removed before upstream requests; Telegram bot-token paths and upstream `Authorization` are unchanged. Redirects are rejected to keep requests within the host allowlist. Upstream failures return a generic `502` without exception details. Local serving requires permission to read `LUMEN_PROXY_SECRET` (`--allow-env=LUMEN_PROXY_SECRET`) as well as network permission; isolated tests require neither.
+
+The bot applies proxy authentication to aiogram, raw Telegram/file requests and TikWM metadata requests, including configured fallbacks. The middleware limits credentials to trusted HTTPS proxy origins and base-path boundaries, never direct Telegram/TikWM or media/CDN hosts. Do not put the secret in shared session headers or the TikTok media headers.
+
+Before publishing, configure the same `LUMEN_PROXY_SECRET` on HF and every Deno proxy. Deploy the updated bot client first, then the protected proxy: old clients cannot use the protected proxy. The updated bot refuses configured proxies without a valid secret. Proxy deployment is separate from the HF auto-deploy. The secret is included in log and Sentry redaction.
+
 ## Admin access & secrets
 
 | Variable | Default | Purpose |

lumen_telegram_transport.py

@@ -26,6 +26,8 @@ from __future__ import annotations
 import logging
 import socket
 import time
+from collections.abc import Sequence
+from urllib.parse import urlsplit
 
 import aiohttp
 from aiogram.client.session.aiohttp import AiohttpSession
@@ -160,32 +162,112 @@ def _build_telegram_connector(limit: int) -> aiohttp.TCPConnector:
     )
 
 
+PROXY_AUTH_HEADER = "X-Lumen-Proxy-Secret"
+
+
+def proxy_auth_middlewares(
+    *, proxy_secret: str = "", proxy_base_urls: Sequence[str] = (),
+) -> tuple:
+    scopes = []
+    for base_url in proxy_base_urls:
+        if not base_url:
+            continue
+        try:
+            parsed = urlsplit(base_url)
+            port = parsed.port or 443
+        except ValueError:
+            raise ValueError("Invalid proxy base URL") from None
+        if (
+            parsed.scheme != "https" or not parsed.hostname
+            or parsed.username is not None or parsed.password is not None
+            or parsed.query or parsed.fragment
+        ):
+            raise ValueError("Proxy base URL must use HTTPS without credentials, query or fragment")
+        if parsed.hostname in {"api.telegram.org", "www.tikwm.com", "tikwm.com"}:
+            continue
+        scopes.append((parsed.hostname, port, parsed.path.rstrip("/")))
+    if scopes and (not proxy_secret or any(not 33 <= ord(c) <= 126 for c in proxy_secret)):
+        raise ValueError("LUMEN_PROXY_SECRET is required for configured proxies and must be printable ASCII without spaces")
+
+    async def authenticate(request: aiohttp.ClientRequest, handler):
+        request.headers.popall(PROXY_AUTH_HEADER, None)
+        url = request.url
+        authenticated = url.scheme == "https" and any(
+            url.host == host and url.port == port
+            and (url.path == path or url.path.startswith(path + "/"))
+            for host, port, path in scopes
+        )
+        if authenticated:
+            request.headers[PROXY_AUTH_HEADER] = proxy_secret
+        try:
+            response = await handler(request)
+        except Exception:
+            if authenticated:
+                raise RuntimeError("Authenticated proxy request failed") from None
+            raise
+        finally:
+            request.headers.popall(PROXY_AUTH_HEADER, None)
+        if authenticated:
+            safe_headers = response.request_info.headers.copy()
+            safe_headers.popall(PROXY_AUTH_HEADER, None)
+            response._request_info = aiohttp.RequestInfo(
+                url=response.request_info.url, method=response.request_info.method,
+                headers=safe_headers, real_url=response.request_info.real_url,
+            )
+        if authenticated and 300 <= response.status < 400:
+            response.close()
+            raise RuntimeError("Authenticated proxy redirects are disabled")
+        return response
+
+    return (authenticate,)
+
+
 class IPv4AiohttpSession(AiohttpSession):
+    def __init__(
+        self, *, proxy_secret: str = "", proxy_base_urls: Sequence[str] = (), **kwargs,
+    ) -> None:
+        self._proxy_middlewares = proxy_auth_middlewares(
+            proxy_secret=proxy_secret, proxy_base_urls=proxy_base_urls,
+        )
+        super().__init__(**kwargs)
+
     async def create_session(self) -> aiohttp.ClientSession:
         if self._session is None or self._session.closed:
             self._session = aiohttp.ClientSession(
                 connector=_build_telegram_connector(limit=30),
                 timeout=aiohttp.ClientTimeout(total=30.0, connect=10.0, sock_read=20.0),
                 json_serialize=self.json_dumps,
+                middlewares=self._proxy_middlewares,
             )
         return self._session
 
 
 _telegram_session: aiohttp.ClientSession | None = None
+_telegram_session_auth: tuple[str, tuple[str, ...]] | None = None
 
 
-async def get_telegram_session(request_timeout: float) -> aiohttp.ClientSession:
+async def get_telegram_session(
+    request_timeout: float, *, proxy_secret: str = "", proxy_base_urls: Sequence[str] = (),
+) -> aiohttp.ClientSession:
     """Кэширующий геттер общей aiohttp-сессии для прямых HTTP-вызовов к Telegram Bot
     API (используется telegram_api_call в bot.py). `request_timeout` — значение
     TELEGRAM_REQUEST_TIMEOUT из bot.py, передаётся параметром на каждый вызов (а не
     импортируется статически), т.к. это часть публичной, потенциально настраиваемой
     через env конфигурации bot.py, а не константа этого модуля."""
-    global _telegram_session
+    global _telegram_session, _telegram_session_auth
+    auth = (proxy_secret, tuple(proxy_base_urls))
+    middlewares = proxy_auth_middlewares(
+        proxy_secret=proxy_secret, proxy_base_urls=auth[1],
+    )
+    if _telegram_session is not None and not _telegram_session.closed and _telegram_session_auth != auth:
+        await _telegram_session.close()
     if _telegram_session is None or _telegram_session.closed:
         _telegram_session = aiohttp.ClientSession(
             connector=_build_telegram_connector(limit=10),
             timeout=aiohttp.ClientTimeout(total=request_timeout + 10.0, connect=10.0),
+            middlewares=middlewares,
         )
+        _telegram_session_auth = auth
     return _telegram_session
 
 

lumen_tiktok.py

@@ -522,6 +522,32 @@ async def _download_url_bin(session: aiohttp.ClientSession, url: str, headers: d
     return None
 
 
+async def _communicate_process(proc: asyncio.subprocess.Process, *, timeout: float):
+    async def finish():
+        try:
+            return await proc.communicate()
+        finally:
+            await proc.wait()
+
+    completion = asyncio.create_task(finish())
+    try:
+        return await asyncio.wait_for(asyncio.shield(completion), timeout=timeout)
+    except (asyncio.TimeoutError, asyncio.CancelledError):
+        if proc.returncode is None:
+            with contextlib.suppress(ProcessLookupError):
+                proc.kill()
+        while not completion.done():
+            try:
+                await asyncio.shield(completion)
+            except asyncio.CancelledError:
+                continue
+            except Exception:
+                break
+        with contextlib.suppress(Exception, asyncio.CancelledError):
+            completion.result()
+        raise
+
+
 async def _probe_video_dimensions(path: str) -> tuple[int, int, int]:
     """Возвращает (duration_seconds, width, height). Без этих полей Telegram иногда
     не может сам распознать видео и показывает его как "сырой файл" с 0:00 вместо
@@ -536,7 +562,7 @@ async def _probe_video_dimensions(path: str) -> tuple[int, int, int]:
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
         )
-        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=15)
+        stdout, _ = await _communicate_process(proc, timeout=15)
         width = height = 0
         duration = 0
         for line in stdout.decode(errors="replace").splitlines():
@@ -567,7 +593,7 @@ async def _generate_video_thumbnail(path: str, duration: int) -> bytes | None:
                 stdout=asyncio.subprocess.DEVNULL,
                 stderr=asyncio.subprocess.DEVNULL,
             )
-            await asyncio.wait_for(proc.wait(), timeout=15)
+            await _communicate_process(proc, timeout=15)
             if os.path.exists(thumb_path) and os.path.getsize(thumb_path) > 0:
                 with open(thumb_path, "rb") as f:
                     return f.read()

proxy.ts

@@ -41,6 +41,8 @@
  * изменений в Python-коде для смены адреса прокси не требуется.
  */
 
+export const PROXY_AUTH_HEADER = "X-Lumen-Proxy-Secret";
+
 export const ALLOWED_HOSTS = new Set([
   "api.telegram.org",
   "www.tikwm.com",
@@ -94,6 +96,7 @@ export function resolveTarget(pathname: string, search: string): TargetResolutio
 export function buildForwardHeaders(reqHeaders: Headers): Headers {
   const headers = new Headers(reqHeaders);
   for (const name of HOP_BY_HOP_REQUEST_HEADERS) headers.delete(name);
+  headers.delete(PROXY_AUTH_HEADER);
   return headers;
 }
 
@@ -105,7 +108,30 @@ export function buildResponseHeaders(upstreamHeaders: Headers): Headers {
 
 // fetchImpl — точка подмены для тестов (тот же приём, что bot._get_http_session
 // и т.п. в Python-части проекта) — реальная сеть не нужна ни одному юнит-тесту.
-export async function handleRequest(req: Request, fetchImpl: typeof fetch = fetch): Promise<Response> {
+export async function handleRequest(
+  req: Request,
+  fetchImpl: typeof fetch = fetch,
+  proxySecret: string | undefined = undefined,
+): Promise<Response> {
+  if (!proxySecret || !/^[\x21-\x7e]+$/.test(proxySecret)) {
+    return new Response("Proxy authentication unavailable", { status: 503 });
+  }
+  const suppliedSecret = req.headers.get(PROXY_AUTH_HEADER) ?? "";
+  const encoder = new TextEncoder();
+  const [expected, supplied] = await Promise.all([
+    crypto.subtle.digest("SHA-256", encoder.encode(proxySecret)),
+    crypto.subtle.digest("SHA-256", encoder.encode(suppliedSecret)),
+  ]);
+  const expectedBytes = new Uint8Array(expected);
+  const suppliedBytes = new Uint8Array(supplied);
+  let difference = 0;
+  for (let i = 0; i < expectedBytes.length; i++) {
+    difference |= expectedBytes[i] ^ suppliedBytes[i];
+  }
+  if (!suppliedSecret || difference !== 0) {
+    return new Response("Unauthorized", { status: 401 });
+  }
+
   const url = new URL(req.url);
   const target = resolveTarget(url.pathname, url.search);
   if (!target.ok) {
@@ -118,6 +144,7 @@ export async function handleRequest(req: Request, fetchImpl: typeof fetch = fetc
     upstreamResp = await fetchImpl(target.url, {
       method: req.method,
       headers: forwardHeaders,
+      redirect: "error",
       // GET/HEAD не могут иметь тело запроса (fetch бросит исключение, если
       // передать body для них) — для остальных методов пробрасываем тело
       // напрямую как поток, не буферизуя целиком в памяти (важно для
@@ -127,8 +154,8 @@ export async function handleRequest(req: Request, fetchImpl: typeof fetch = fetc
       // (часть стандарта WHATWG fetch для body типа ReadableStream).
       duplex: "half",
     });
-  } catch (e) {
-    return new Response(`Upstream fetch failed: ${e}`, { status: 502 });
+  } catch {
+    return new Response("Upstream fetch failed", { status: 502 });
   }
 
   const respHeaders = buildResponseHeaders(upstreamResp.headers);
@@ -141,5 +168,6 @@ export async function handleRequest(req: Request, fetchImpl: typeof fetch = fetc
 // Реальный сервер стартует только при прямом запуске файла (deno run/deploy),
 // не при импорте из proxy_test.ts — иначе тесты пытались бы забиндить порт.
 if (import.meta.main) {
-  Deno.serve((req) => handleRequest(req));
+  const proxySecret = Deno.env.get("LUMEN_PROXY_SECRET");
+  Deno.serve((req) => handleRequest(req, fetch, proxySecret));
 }

proxy_test.ts

@@ -7,12 +7,16 @@
 
 import {
   ALLOWED_HOSTS,
+  PROXY_AUTH_HEADER,
   buildForwardHeaders,
   buildResponseHeaders,
   handleRequest,
   resolveTarget,
 } from "./proxy.ts";
 
+const TEST_SECRET = "isolated-test-secret";
+const AUTH_HEADERS = { [PROXY_AUTH_HEADER]: TEST_SECRET };
+
 function assert(condition: boolean, message: string): void {
   if (!condition) throw new Error(message);
 }
@@ -132,8 +136,8 @@ Deno.test("handleRequest пробрасывает GET без тела и воз
       headers: { "Content-Type": "application/json" },
     }));
   };
-  const req = new Request("https://proxy.example/fetch/www.tikwm.com/api/?url=x", { method: "GET" });
-  const resp = await handleRequest(req, fakeFetch);
+  const req = new Request("https://proxy.example/fetch/www.tikwm.com/api/?url=x", { method: "GET", headers: AUTH_HEADERS });
+  const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
   assertEquals(resp.status, 200);
   const body = await resp.json();
   assertEquals(body, { code: 0 });
@@ -145,16 +149,16 @@ Deno.test("handleRequest возвращает 403 для неразрешённ
     fetchCalled = true;
     return Promise.resolve(new Response("не должно случиться"));
   };
-  const req = new Request("https://proxy.example/fetch/evil.example.com/x", { method: "GET" });
-  const resp = await handleRequest(req, fakeFetch);
+  const req = new Request("https://proxy.example/fetch/evil.example.com/x", { method: "GET", headers: AUTH_HEADERS });
+  const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
   assertEquals(resp.status, 403);
   assertEquals(fetchCalled, false, "fetch не должен был вызываться для неразрешённого хоста");
 });
 
 Deno.test("handleRequest возвращает 404 для пути без /fetch/ префикса", async () => {
-  const req = new Request("https://proxy.example/something-else", { method: "GET" });
+  const req = new Request("https://proxy.example/something-else", { method: "GET", headers: AUTH_HEADERS });
   const unusedFetch: typeof fetch = () => Promise.resolve(new Response("unused"));
-  const resp = await handleRequest(req, unusedFetch);
+  const resp = await handleRequest(req, unusedFetch, TEST_SECRET);
   assertEquals(resp.status, 404);
 });
 
@@ -169,9 +173,9 @@ Deno.test("handleRequest пробрасывает метод и тело для
   const req = new Request("https://proxy.example/fetch/api.telegram.org/bot123/sendMessage", {
     method: "POST",
     body: JSON.stringify({ chat_id: 1, text: "hi" }),
-    headers: { "Content-Type": "application/json" },
+    headers: { "Content-Type": "application/json", ...AUTH_HEADERS },
   });
-  const resp = await handleRequest(req, fakeFetch);
+  const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
   assertEquals(resp.status, 200);
   assertEquals(capturedMethod, "POST");
   assert(capturedBodyIsStream, "тело POST-запроса должно передаваться апстриму как поток (без буферизации целиком)");
@@ -179,9 +183,72 @@ Deno.test("handleRequest пробрасывает метод и тело для
 
 Deno.test("handleRequest возвращает 502, если апстрим-fetch упал с исключением", async () => {
   const fakeFetch: typeof fetch = () => {
-    throw new Error("network unreachable");
+    throw new Error(`network unreachable ${TEST_SECRET}`);
   };
-  const req = new Request("https://proxy.example/fetch/api.telegram.org/getMe", { method: "GET" });
-  const resp = await handleRequest(req, fakeFetch);
+  const req = new Request("https://proxy.example/fetch/api.telegram.org/getMe", { method: "GET", headers: AUTH_HEADERS });
+  const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
   assertEquals(resp.status, 502);
+  assertEquals(await resp.text(), "Upstream fetch failed");
+});
+
+for (const suppliedSecret of [undefined, "", "incorrect-test-secret"]) {
+  Deno.test(`handleRequest rejects missing or wrong secret: ${String(suppliedSecret)}`, async () => {
+    let fetchCalled = false;
+    const fakeFetch: typeof fetch = () => {
+      fetchCalled = true;
+      return Promise.resolve(new Response("unexpected"));
+    };
+    const headers = new Headers();
+    if (suppliedSecret !== undefined) headers.set(PROXY_AUTH_HEADER, suppliedSecret);
+    const req = new Request("https://proxy.example/fetch/api.telegram.org/bot123/getMe", { headers });
+    const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
+    assertEquals(resp.status, 401);
+    assertEquals(await resp.text(), "Unauthorized");
+    assertEquals(fetchCalled, false);
+  });
+}
+
+for (const configuredSecret of [undefined, "", "   ", "invalid secret"]) {
+  Deno.test(`handleRequest fails closed for invalid configuration: ${String(configuredSecret)}`, async () => {
+    let fetchCalled = false;
+    const fakeFetch: typeof fetch = () => {
+      fetchCalled = true;
+      return Promise.resolve(new Response("unexpected"));
+    };
+    const req = new Request("https://proxy.example/fetch/www.tikwm.com/api/", { headers: AUTH_HEADERS });
+    const resp = await handleRequest(req, fakeFetch, configuredSecret);
+    assertEquals(resp.status, 503);
+    assertEquals(await resp.text(), "Proxy authentication unavailable");
+    assertEquals(fetchCalled, false);
+  });
+}
+
+Deno.test("correct secret is stripped without changing Telegram authentication", async () => {
+  let fetchCalled = false;
+  const fakeFetch: typeof fetch = (url, init) => {
+    fetchCalled = true;
+    assertEquals(String(url), "https://api.telegram.org/bot123/getMe");
+    const headers = new Headers(init?.headers);
+    assertEquals(headers.has(PROXY_AUTH_HEADER), false);
+    assertEquals(headers.get("authorization"), "Bearer upstream-test-token");
+    assertEquals(init?.redirect, "error");
+    return Promise.resolve(new Response('{"ok":true}', { status: 200 }));
+  };
+  const req = new Request("https://proxy.example/fetch/api.telegram.org/bot123/getMe", {
+    headers: {
+      "x-lumen-proxy-secret": TEST_SECRET,
+      "Authorization": "Bearer upstream-test-token",
+    },
+  });
+  const resp = await handleRequest(req, fakeFetch, TEST_SECRET);
+  assertEquals(resp.status, 200);
+  assertEquals(fetchCalled, true);
+  assertEquals(await resp.json(), { ok: true });
+  assertEquals(req.headers.get(PROXY_AUTH_HEADER), TEST_SECRET);
+});
+
+Deno.test("buildForwardHeaders removes mixed-case proxy secret without mutating input", () => {
+  const original = new Headers({ "X-LuMeN-PrOxY-SeCrEt": TEST_SECRET });
+  assertEquals(buildForwardHeaders(original).has(PROXY_AUTH_HEADER), false);
+  assertEquals(original.get(PROXY_AUTH_HEADER), TEST_SECRET);
 });

test_bot.py

@@ -26,15 +26,18 @@ BOT_LOG_PATH) ДО импорта bot.py, так что реальные сек
 import asyncio
 import base64
 import json
+import logging
 import os
+import sys
 import time
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 import sentry_sdk
 
 import bot
+import lumen_telegram_transport
 import lumen_tiktok
 
 
@@ -4501,8 +4504,276 @@ def test_download_url_bin_ignores_malformed_content_length_header():
     assert result == b"ok"
 
 
+@pytest.fixture
+def rate_guard_setup(monkeypatch):
+    def _setup():
+        monkeypatch.setattr(bot, "user_rate_limits", {})
+        monkeypatch.setattr(bot, "RATE_LIMIT_MAX_REQUESTS", 2)
+        monkeypatch.setattr(bot, "get_state", lambda chat_id: {})
+        monkeypatch.setattr(bot, "is_guest_message", lambda message: False)
+        monkeypatch.setattr(bot, "message_mentions_bot", lambda message: False)
+        monkeypatch.setattr(bot, "_safe_reply", AsyncMock())
+        monkeypatch.setattr(bot, "_tg_call", AsyncMock())
+        monkeypatch.setattr(bot, "inline_draw", AsyncMock())
+        monkeypatch.setattr(bot, "inline_tts", AsyncMock())
+        return SimpleNamespace(
+            text="", caption=None, chat=SimpleNamespace(id=123, type=bot.ChatType.PRIVATE),
+            from_user=SimpleNamespace(id=456), reply=AsyncMock(), reply_to_message=None,
+        )
+    return _setup
+
+
+def test_draw_command_rejects_exhausted_quota(rate_guard_setup, monkeypatch):
+    message = rate_guard_setup()
+    message.text = "/draw кот"
+    monkeypatch.setattr(bot, "_check_and_register_rate_limit", lambda user_id: True)
+
+    asyncio.run(bot.cmd_draw(message))
+
+    bot.inline_draw.assert_not_awaited()
+    bot._tg_call.assert_awaited_once()
+
+
+def test_mixed_commands_share_rate_quota_and_reject_before_work(rate_guard_setup, monkeypatch):
+    message = rate_guard_setup()
+    media = AsyncMock()
+    monkeypatch.setattr(bot, "_resolve_incoming_media", media)
+
+    async def run():
+        message.text = "/draw кот"
+        await bot.cmd_draw(message)
+        message.text = "/tts привет"
+        await bot.cmd_tts(message)
+        await bot.cmd_tts(message)
+        message.text = "/draw собака"
+        await bot.cmd_draw(message)
+        message.text = "обычный вопрос"
+        await bot._handle_message_core(message)
+
+    asyncio.run(run())
+    bot.inline_draw.assert_awaited_once_with(message, "кот")
+    bot.inline_tts.assert_awaited_once_with(message, "привет")
+    media.assert_not_awaited()
+    assert bot._tg_call.await_count == 3
+    assert len(bot.user_rate_limits[456]) == 2
+
+
+@pytest.mark.parametrize("command", ["draw", "tts"])
+def test_blank_commands_do_not_consume_quota(rate_guard_setup, command):
+    message = rate_guard_setup()
+    message.text = f"/{command}   "
+    asyncio.run(getattr(bot, f"cmd_{command}")(message))
+    assert bot.user_rate_limits == {}
+    bot._safe_reply.assert_awaited_once()
+    bot.inline_draw.assert_not_awaited()
+    bot.inline_tts.assert_not_awaited()
+
+
+@pytest.mark.parametrize("text, handler, content", [
+    ("нарисуй кота", "inline_draw", "кота"),
+    ("озвучь привет", "inline_tts", "привет"),
+])
+def test_natural_language_trigger_consumes_one_slot(rate_guard_setup, text, handler, content):
+    message = rate_guard_setup()
+    message.text = text
+    asyncio.run(bot._handle_message_core(message))
+    getattr(bot, handler).assert_awaited_once_with(message, content)
+    assert len(bot.user_rate_limits[456]) == 1
+
+
+def test_passive_group_message_does_not_consume_quota(rate_guard_setup, monkeypatch):
+    message = rate_guard_setup()
+    message.chat.type = bot.ChatType.GROUP
+    message.text = "обычный разговор"
+    record = MagicMock()
+    monkeypatch.setattr(bot, "_record_passive_group_context", record)
+    asyncio.run(bot._handle_message_core(message))
+    record.assert_called_once()
+    assert bot.user_rate_limits == {}
+    bot._tg_call.assert_not_awaited()
+
+
 def test_download_url_bin_returns_none_on_non_200_status():
     resp = _FakeDownloadResponse([b"error page"], status=404)
     session = _FakeDownloadSession(resp)
     result = asyncio.run(lumen_tiktok._download_url_bin(session, "https://tikwm.com/missing.jpg"))
     assert result is None
+
+
+class _FakeProc:
+    def __init__(self, *, hang: bool = False, fail_communicate: bool = False):
+        self._hang = hang
+        self._fail_communicate = fail_communicate
+        self.killed = False
+        self.kill_count = 0
+        self.wait_count = 0
+        self.returncode = None
+
+    async def communicate(self):
+        if self._fail_communicate:
+            raise RuntimeError("pipe broken")
+        while self._hang and not self.killed:
+            await asyncio.sleep(0.01)
+        return b"out", b"err"
+
+    def kill(self):
+        self.kill_count += 1
+        self.killed = True
+        self.returncode = -9
+        self._hang = False
+
+    async def wait(self):
+        self.wait_count += 1
+        return self.returncode
+
+
+def test_communicate_process_returns_output_on_success():
+    proc = _FakeProc()
+    out, err = asyncio.run(lumen_tiktok._communicate_process(proc, timeout=5))
+    assert (out, err) == (b"out", b"err")
+    assert not proc.killed
+
+
+def test_communicate_process_kills_hung_process_on_timeout():
+    async def run():
+        proc = _FakeProc(hang=True)
+        started = time.monotonic()
+        with pytest.raises(asyncio.TimeoutError):
+            await lumen_tiktok._communicate_process(proc, timeout=0.05)
+        elapsed = time.monotonic() - started
+        assert proc.killed
+        assert proc.wait_count >= 1
+        return elapsed
+
+    elapsed = asyncio.run(run())
+    assert elapsed < 30
+
+
+def test_communicate_process_cleans_up_when_outer_task_cancelled():
+    async def run():
+        proc = _FakeProc(hang=True)
+        task = asyncio.create_task(lumen_tiktok._communicate_process(proc, timeout=5))
+        await asyncio.sleep(0.01)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        for _ in range(5):
+            if proc.killed and task.done():
+                break
+            await asyncio.sleep(0.01)
+        assert proc.killed
+        assert task.cancelled()
+
+    asyncio.run(run())
+
+
+def test_communicate_process_propagates_communicate_failure():
+    async def run():
+        proc = _FakeProc(fail_communicate=True)
+        with pytest.raises(RuntimeError):
+            await lumen_tiktok._communicate_process(proc, timeout=5)
+        assert proc.wait_count >= 1
+
+    asyncio.run(run())
+
+
+def test_log_queue_handler_masks_secrets_in_args_and_traceback(monkeypatch):
+    monkeypatch.setattr(bot, "LUMEN_PROXY_SECRET", "fake-proxy-secret-123", raising=False)
+    try:
+        raise RuntimeError("leak fake-proxy-secret-123")
+    except RuntimeError:
+        exc_info = sys.exc_info()
+    record = logging.LogRecord(
+        "bot", logging.ERROR, "path", 1,
+        "failed with token %s", ("fake-proxy-secret-123",), exc_info=exc_info,
+    )
+    prepared = bot._LOG_QUEUE_HANDLER.prepare(record)
+    rendered = bot._LOG_QUEUE_HANDLER.format(prepared)
+    assert "fake-proxy-secret-123" not in rendered
+    assert "<REDACTED>" in rendered
+
+
+def test_secret_log_formatter_masks_message_before_queue(monkeypatch):
+    formatter = bot._SecretLogFormatter("%(message)s")
+    monkeypatch.setattr(bot, "LUMEN_PROXY_SECRET", "fake-formatter-secret-456", raising=False)
+    record = logging.LogRecord(
+        "bot", logging.INFO, "path", 1,
+        "key is %s", ("fake-formatter-secret-456",), None,
+    )
+    assert "fake-formatter-secret-456" not in formatter.format(record)
+    assert "<REDACTED>" in formatter.format(record)
+
+
+def test_current_log_secrets_reads_env_before_module_globals(monkeypatch):
+    monkeypatch.delenv("LUMEN_PROXY_SECRET", raising=False)
+    monkeypatch.setattr(bot, "LUMEN_PROXY_SECRET", "", raising=False)
+    monkeypatch.setenv("LUMEN_PROXY_SECRET", "fake-env-only-secret-789")
+    secrets = bot._current_log_secrets()
+    assert "fake-env-only-secret-789" in secrets
+    monkeypatch.delenv("LUMEN_PROXY_SECRET", raising=False)
+
+
+def _run_proxy_middleware(url, *, secret="proxy-secret-abc", bases=("https://proxy.example/fetch/api.telegram.org",), headers=None):
+    from multidict import CIMultiDict
+    from yarl import URL
+
+    (authenticate,) = lumen_telegram_transport.proxy_auth_middlewares(
+        proxy_secret=secret, proxy_base_urls=bases,
+    )
+    seen = {}
+
+    class _FakeRequest:
+        def __init__(self):
+            self.url = URL(url)
+            self.headers = CIMultiDict(headers or {})
+
+    async def _handler(request):
+        seen["sent"] = dict(request.headers)
+        info_headers = CIMultiDict(request.headers)
+
+        class _FakeResponse:
+            status = 200
+
+            def __init__(self):
+                self.request_info = SimpleNamespace(
+                    url=request.url, method="GET", headers=info_headers, real_url=request.url,
+                )
+                self.closed = False
+
+            def close(self):
+                self.closed = True
+
+        return _FakeResponse()
+
+    request = _FakeRequest()
+    response = asyncio.run(authenticate(request, _handler))
+    return request, response, seen
+
+
+def test_proxy_middleware_sends_secret_only_to_configured_proxy():
+    _, _, seen = _run_proxy_middleware("https://proxy.example/fetch/api.telegram.org/bot123/sendMessage")
+    assert seen["sent"].get("X-Lumen-Proxy-Secret") == "proxy-secret-abc"
+
+
+def test_proxy_middleware_never_sends_secret_to_direct_or_unrelated_hosts():
+    for url in (
+        "https://api.telegram.org/bot123/sendMessage",
+        "https://www.tikwm.com/api/?url=x",
+        "https://cdn.example.com/file.jpg",
+        "https://proxy.example/other-path",
+        "http://proxy.example/fetch/api.telegram.org/bot123/sendMessage",
+    ):
+        _, _, seen = _run_proxy_middleware(url)
+        assert "X-Lumen-Proxy-Secret" not in seen["sent"], url
+
+
+def test_proxy_middleware_strips_stale_secret_and_requires_secret():
+    _, _, seen = _run_proxy_middleware(
+        "https://api.telegram.org/bot123/sendMessage",
+        headers={"X-Lumen-Proxy-Secret": "stale"},
+    )
+    assert "X-Lumen-Proxy-Secret" not in seen["sent"]
+    with pytest.raises(ValueError):
+        lumen_telegram_transport.proxy_auth_middlewares(
+            proxy_secret="", proxy_base_urls=("https://proxy.example/fetch/api.telegram.org",),
+        )