diff --git a/REVIEW.md b/REVIEW.md new file mode 100644 index 0000000..656556e --- /dev/null +++ b/REVIEW.md @@ -0,0 +1,157 @@ +# REVIEW.md — telegram-scraper + +Дата: 2026-09-07 +Скоуп: полный аудит кода — безопасности и корректности (webui_server.py, app_state.py, scraper_jobs.py, telegram_scraper_with_forwarding.py, main.py, health.py, webui/*.js, деплой, тесты). + +--- + +## Статус: что уже исправлено + +10 пунктов (критичные/высокие) исправлены 07.09.2026. Подробности — в истории коммита. + +| ID | Проблема | Файл | Статус | +|---|---|---|---| +| C-2 | SSE-поток не завершался при успешном job (`done` vs `completed`) — утечка потоков до 30 мин | webui_server.py | ✅ fixed | +| C-3 | Пути `Path("data")`/`Path("session")` относительно CWD расходились с BASE_DIR webui — тихая рассинхронизация данных | telegram_scraper_with_forwarding.py, main.py, scraper_jobs.py | ✅ fixed | +| H-1 | `TelegramAuthManager.lock` объявлен, но не использовался — гонки на `auth_data`/`clients` между event-loop и HTTP-потоками | webui_server.py | ✅ fixed (RLock) | +| H-2 | Коллизия `job_id` из миллисекундного timestamp | webui_server.py | ✅ fixed (uuid4) | +| H-3 | Экспорт секретов через `include_secrets=1` (api_hash/api_id) | webui_server.py | ✅ fixed (всегда redact) | +| H-4 | CSRF: любой Content-Type, нет Origin/Sec-Fetch-Site проверки | webui_server.py | ✅ fixed (415 + same-origin 403) | +| H-5 | Нет rate limiting на брутфорс phone-code/2FA | webui_server.py | ✅ fixed (5 попыток → 60s lockout, 429) | +| H-6 | `StateStore.load()` возвращал мелкую копию — расшаренная мутация вложенных dict между потоками | app_state.py | ✅ fixed (deepcopy) | +| H-7 | Бесконечная рекурсия `forward_message` при FloodWaitError | telegram_scraper_with_forwarding.py | ✅ fixed (cap 3 retry) | +| H-8 | Повторная регистрация forward-хендлера → сообщения форвардились N раз | telegram_scraper_with_forwarding.py | ✅ fixed (remove_event_handler) | + +Бонус при фиксах: `migrate_database` больше не глотает исключения (логирует), миграция покрывает все колонки MessageData; фронтенд `app.js` корректно распознаёт `'done'` как терминальный статус. + +### Follow-up (второй проход по итогам ревью фиксов) + +| ID | Проблема | Статус | +|---|---|---| +| H-6 | deepcopy теперь на **всех** путях `load()` (cache-hit + cache-miss + fallback-ветки) | ✅ fixed | +| H-5 | cooldown 30s на успешный запрос кода (анти-SMS-флуд) + `_auth_attempts` ограничен (sweep при >10k записей) | ✅ fixed | +| — | Origin-проверка на GET `/api/jobs/{id}/events` (403 до открытия SSE) | ✅ fixed | +| — | SSE-потоки на аккаунт ограничены (`MAX_EVENT_STREAMS=10`, revoke самого старого) — закрыт thread-exhaustion | ✅ fixed | +| — | `read_json_body`: кап тела 1 MB → 413 (sentinel), malformed Content-Length → не 500, не-UTF-8 → 400 | ✅ fixed (M-2 закрыт) | +| — | +6 регрессионных тестов (H-4/H-5/H-6/C-2, sweep, oversize body) | ✅ fixed | + +Тесты: `29 passed` (все зелёные после фиксов). + +### Follow-up (третий проход) + +| ID | Что исправлено | Статус | +|---|---|---| +| M-17 | `clean_continuous_channels()` — валидация каналов на приёме в обоих POST-эндпоинтах (per-account + legacy); невалидные отбрасываются и возвращаются как `dropped_invalid` (drop, не 400 — фронтенд api() бросает на non-OK) | ✅ fixed | +| M-5 | `PerAccountContinuousScrapeManager.join(timeout=20)` + `remove_account` делает stop()+join перед rmtree; `stop()` идемпотентен; loop проверяет stop event на границах итераций | ✅ fixed | +| M-6 | `self.config` присваивается под локом; `refresh_config` больше не врёт о `running` (только stop_event); `running=False` ставится в `finally` потока при реальном выходе | ✅ fixed | +| — | +4 теста (33 passed) | ✅ fixed | + +--- + +## ОСТАВШИЕСЯ НАХОДКИ + +### 🔴 КРИТИЧНО — C-1. Нет аутентификации на веб-панели, bind 0.0.0.0 + публичный ingress + +Файл/строки: `webui_server.py:47-48` (DEFAULT_HOST=0.0.0.0), весь роутинг без auth-check, `compose.yaml:11-12` (порт 7887 на всех интерфейсах), `k8s/telegram-scraper.yaml:76-89` (IngressRoute `tg.workstation.internal` без middleware/basicAuth). + +Любой, кто достаёт порт/домен, может: +- прочитать `api_hash`/`api_id` (через export — теперь redact, но есть и другие пути, см. M-1: `/media/accounts//state.json`), +- прочитать QR-токен авторизации и **угнать Telegram-сессию** владельца, +- подменить креды, удалить аккаунт (`DELETE /api/accounts/{id}` → `shutil.rmtree`), +- читать все чаты, медиа, логи, continuous-scrape состояние. + +**Не исправлено** (сознательно — требует архитектурного решения). Рекомендуемый порядок: +1. `BasicAuth`/`ForwardAuth`/OIDC на Traefik IngressRoute (быстро, закрывает сетевой доступ). +2. App-level сессионная авторизация (cookie + random token), проверка в `do_GET`/`do_POST`/`do_DELETE` до диспатча. +3. Дефолт bind `127.0.0.1` + не публиковать 7887 на всех интерфейсах. +4. После ввода auth — пересмотреть M-1 (см. ниже), который сейчас маскируется отсутствием auth. + +--- + +### 🟠 СРЕДНИЕ + +| # | Файл:строка (актуально) | Проблема | Предложение | +|---|---|---|---| +| M-1 | webui_server.py:2829+ (`serve_media`) | `/media/` рутится в `DATA_DIR` целиком: `GET /media/accounts//state.json` отдаёт api_hash (plaintext), `/media/accounts///*.db` — базы. Conтент-проверки нет, только containment | Требовать сегмент `media/` в пути после account/channel; запретить `state.json`, `*.db`, `*.session` | +| ~~M-2~~ | ~~`read_json_body`~~ | ~~Нет капа тела, malformed Content-Length → 500~~ | ✅ закрыт follow-up: кап 1 MB → 413, try/except, не-UTF-8 → 400 | +| M-3 | webui_server.py (много: 2168-2176, 2188, 2195, 2484, 2709, 2716) | `bool(body.get("value"/"enabled"/"run_all_tracked"))` — строка `"false"`/`"0"` приходит как `True`. Фиксы H-4 не тронули эти места | Общий хелпер `parse_bool()`: `True` для `true/1/yes/on` | +| M-4 | webui_server.py `do_HEAD` (2120) + `stream_job_events` (2064) | HEAD на `/api/jobs/{id}/events` для несуществующего job → 200 вместо 404 | Валидировать job до HEAD | +| ~~M-5~~ | ~~webui_server.py:1102+ (`ContinuousScrapeOrchestrator.remove_account`)~~ | ~~Удаление аккаунта не джойнит поток continuous scrape: `stop()` только ставит event → `shutil.rmtree` может удалить DB/media, которые поток ещё пишет~~ | ✅ fixed | +| ~~M-6~~ | ~~webui_server.py:872+ (`refresh_config`)~~ | ~~Ставит `status["running"]=False`, пока поток ещё крутится (status врёт); `update()` присваивает `self.config` вне лока~~ | ✅ fixed | +| M-7 | webui_server.py:404-426 (`JobRunner.shutdown`) | Очередные jobs остаются `"queued"` навсегда (worker выходит, не дрена́я очередь) | Дрена́ж + пометить `"failed"/"cancelled"` на shutdown | +| M-8 | webui_server.py:2879+ (`_parse_range`) | Мульти-диапазоны `bytes=0-1,5-6` → 416; `bytes=0-0` на пустом файле → 416 | Обработать single-range случаи корректно | +| M-9 | webui_server.py legacy endpoints + `webui_server.py:116-120` (`load_state`/`save_state` через `STATE_STORE`) vs `app_state.py:123-131` (`_GLOBAL_STORE`) | Два независимых StateStore на один файл `data/state.json` — расхождение TTL-кэшей до 1s, конфликтные `.tmp`. Legacy GET `/api/channels`/`/api/dashboard` после миграции читают пустой глобальный state (не делегируют в migrated account) | Свести к единому store; legacy GET — делегировать в `legacy_account_id` | +| M-10 | app_state.py:72-81 (`save`) | Нет `fsync` перед rename (потеря питания → пустой/битый файл); фиксированное имя `.tmp` (два писателя в файл клообьют друг друга) | `flush()+os.fsync()` перед replace; уникальные tmp-имена (tempfile) | +| M-11 | scraper_jobs.py:14-25 | `asyncio.run()` на каждый job — `RuntimeError` при вызове из потока с существующим loop (e.g. auth loop thread); `set_scrape_media` пишет в глобальный `STATE_STORE` вместо per-account | Выделенный поток с `new_event_loop()` / per-account клиент-пул | +| M-12 | telegram_scraper_with_forwarding.py (scrape_channel) | Держит все media-объекты в памяти за весь проход (100k+ сообщений в большом канале) | Пакетная обработка media (как batch_insert) | +| M-13 | telegram_scraper_with_forwarding.py:127-131 (`save_state`) | Перезапись всего per-account JSON каждые 50 сообщений — сотни сериализаций на длинный канал | Throttle до 5s / писать только в конце | +| M-14 | telegram_scraper_with_forwarding.py (existing_files glob) | Первое произвольное совпадение `{id}-*` может быть stale/частичным файлом | Матчить точное имя / проверять non-empty | +| M-15 | health.py:73-83 | `/health` читает глобальный state: в multi-account режиме всегда `has_api_credentials: false, tracked_channels: 0` — вводит в заблуждение | Агрегировать per-account проверки | +| M-16 | webui_server.py(s) + k8s | Контейнер в k8s без `securityContext` (root, r/w FS, нет limits); в Dockerfile нет `USER` (compose задаёт 1000:1000, k8s — нет) | `runAsNonRoot: true, readOnlyRootFilesystem: true` + `resources.limits` | +| ~~M-17~~ | ~~webui_server.py continuous endpoints (обе версии `/api/continuous` и `/api/accounts/{id}/continuous`)~~ | ~~Список каналов сохраняется сырым `str().strip()` без `normalize_channel_id` — безопасно только пока фильтрует `_resolve_channels` по tracked~~ | ✅ fixed | +| M-18 | data/ и session/ (хост) | `root:root 755`, state.json пишется 644 — session-файлы Telethon (полные auth-ключи) и api_hash читаемы локальными юзерами | chmod 700 на data/session; StateStore пишет 0600 | +| M-19 | webui_server.py:1860-1862, 2011-2013 и др. | `int(query...)` без try/except → ValueError убивает поток + traceback в stderr; многие хендлеры эхат `str(exc)` (абс-пути в ответах) | try/except → 400 JSON; ред.актировать пути из ответов | +| M-20 | webui_server.py (все POST) | CSRF-фикс (H-4) закрыл Origin/Content-Type, но CSRF-токенов per-session нет; при вводе реальной auth (C-1) нужны | CSRF-token + SameSite cookies после C-1 | + +--- + +### ⚪ НИЗКИЕ + +| # | Файл | Проблема | +|---|---|---| +| L-1 | webui_server.py:2776-2782 (access log) | Логируется весь `self.path` с query-параметрами (поисковые запросы и т.п.) | +| L-2 | webui/swagger.js:52 | `innerHTML` с ошибкой из /openapi.json (низкий риск — серверный контент) | +| L-3 | requirements.txt | Зависимости корректны (aiohttp 3.12.14 — патч CVE-2025-53643), но Telethon 1.40.0 (есть ~1.44.x); добавить `uv audit`/`pip-audit` в CI | +| L-4 | webui_server.py send_json/serve_file | Нет security-заголовков: CSP, X-Content-Type-Options, X-Frame-Options/frame-ancestors, Referrer-Policy (clickjacking актуален после ввода auth) | +| L-5 | webui_server.py auth snapshots (582-589) | QR-токен и его изображение висят в snapshot до сканирования — one-time + expiry ~60s | +| L-6 | webui_server.py:2147-2173 (legacy channels add/remove) | Паттерн load→save вместо `StateStore.update()` — lost-update race между потоками | +| L-7 | telegram_scraper_with_forwarding.py:838-841 | Прогресс-бар врут на инкрементальных прогонах (total vs only-new) — косметика | +| L-8 | webui_server.py `_check_same_origin` | DNS-rebinding: `Host == Origin.netloc` проходит, если оба — домен атакующего (при rebinding `Sec-Fetch-Site` = same-origin). Закрыть allowlist'ом (localhost/127.0.0.1) или дефолт-bind 127.0.0.1 | +| L-9 | webui/app.js:944-962, webui/settings.js:248-266 | Import/export round-trip молча теряет `api_id`/`api_hash` (H-3 redact): UI не предупреждает, что креды нужно ввести заново после импорта | Toast после импорта с redacted-флагами | + +--- + +## Follow-up findings (round 3) + +### 🟠 СРЕДНИЕ (from round-3 review) + +| # | Файл:строка | Проблема | Предложение | +|---|---|---|---| +| F-1 | webui_server.py:2592, app_state.py:263 | Импорт аккаунта и legacy-миграция пишут `continuous_scraping.channels` как есть, минуя валидацию M-17 (латентно, т.к. `_resolve_channels` фильтрует по normalized tracked) | Прогонять через `clean_continuous_channels` при импорте и миграции | +| F-2 | webui_server.py:2880 + webui/app.js:207-211 | `dropped_invalid` возвращается, но ни один JS его не читает — юзер не видит, что каналы отброшены | В app.js при сохранении: `if (resp.dropped_invalid?.length) toast(...)` | +| F-3 | webui_server.py:1048-1058 | Enable во время drain: `start()` early-return по `status["running"]`, потом `finally` ставит False — аккаунт enabled=True, но мёртв до ручного переключения | `start()` проверять `thread.is_alive()` или ждать drain через `join()` перед стартом | +| F-4 | webui_server.py:1246-1254 | `remove_account` удаляет менеджера даже при таймауте join — recreate того же id создаёт второй воркер поверх живого (дубли) | Pop только если `join()` вернул True; иначе tombstone | + +### ⚪ НИЗКИЕ (from round-3 review) + +| # | Файл | Проблема | +|---|---|---| +| F-5 | webui_server.py:237-238 | `channels: null/не-список` молча стирает весь список каналов (`([], [])`) — рассмотреть 400 на malformed payload | +| F-6 | webui_server.py:987-991 | `refresh_config`/`_save_config` стрипают, но не нормализуют — `@`-значения с диска (import/migration) никогда не матчатся с normalized tracked, молча не скрейпятся | +| F-7 | webui_server.py:1104-1171 | Нет верхнего `except` в `_run_loop`: исключение в refresh/auth-check убивает поток с последним_error нетронутым | +| F-8 | tests/test_integration.py:441-477 | Новые тесты не покрывают `join()`→False (таймаут) и start-during-drain; assert `running is True` после stop завязан на GIL-timing | + +--- + +### 🧪 Пробелы в тестах + +Покрыто новыми тестами (round 2, +6): deepcopy-изоляция `load()` (все пути), Content-Type/oversize в `read_json_body`, same-origin проверка, rate limiter (lockout + cooldown кода), sweep `_auth_attempts`, терминальные статусы SSE. + +Осталось: +- tests/test_integration.py:29-38 — **весь** scraper-движок замокан (`sys.modules["telegram_scraper_with_forwarding"] = MagicMock()`): реальный код (media naming, flood, forwarding, DB миграция, session) не покрыт вообще. Рекомендация: ставить telethon в CI и импортировать реальный модуль. +- tests/test_integration.py:176-201 — тест удаления аккаунта дублирует логику хендлера инлайн, не вызывает продакшн-путь → регрессии в `_handle_delete_account` не ловятся. +- Нет тестов на: `StateStore` TTL/atomic-write/конкурентный `update()`; `_parse_range` / Range-ответы; `normalize_media_url`/`guess_media_kind`; 404 SSE; auth-флоу (QR/phone/2FA state machine); H-1 (lock-дисциплина); H-7/H-8 (scraper engine — упирается в полный мок движка). +- tests/test_integration.py:48-50 — мутация `os.environ` на уровне импорта (leak между модулями). Лучше `monkeypatch`. +- tests/test_integration.py:337-397 — `START_CONTINUOUS = False` выставляется в setup и не восстанавливается. + +--- + +### ✅ Проверено — уязвимостей НЕТ + +- **Path traversal**: `serve_media`/`serve_static` — `resolve()` + `relative_to()` (корректно, включая symlink); `normalize_channel_id` отвергает `/`, `\`, control chars, `.`/`..`. +- **SQL injection**: все запросы параметризованы, `search` — через `LIKE ?`. +- **XSS**: viewer.js рендерит контент через `textContent`/`createTextNode`; media-URL всегда префиксуется `/media/` (нет `javascript:` схемы). +- **SSRF**: юзер-контролируемого фетча URL нет (только MTProto). +- **Десериализация**: только JSON, без pickle/yaml. +- **Command injection**: нет subprocess/os.system в продакшн-путях. +- **Secrets в image**: `.dockerignore` исключает `data/` и `session/`. \ No newline at end of file diff --git a/app_state.py b/app_state.py index 5a84e21..b5866c3 100644 --- a/app_state.py +++ b/app_state.py @@ -50,12 +50,12 @@ class StateStore: with self.lock: now = time.time() if self._cache is not None and (now - self._cache_time) < self._cache_ttl: - return dict(self._cache) + return deepcopy(self._cache) if not self.path.exists(): result = deepcopy(self.defaults) self._cache = result self._cache_time = now - return result + return deepcopy(result) try: with self.path.open("r", encoding="utf-8") as handle: state: Dict[str, Any] = json.load(handle) @@ -63,11 +63,11 @@ class StateStore: result = deepcopy(self.defaults) self._cache = result self._cache_time = now - return result + return deepcopy(result) result = self._merge_defaults(state) self._cache = result self._cache_time = now - return result + return deepcopy(result) def save(self, state: Dict[str, Any]) -> None: with self.lock: diff --git a/main.py b/main.py index 7bc8875..9dad28e 100644 --- a/main.py +++ b/main.py @@ -18,8 +18,9 @@ def main(): # Run legacy migration before starting the server from app_state import migrate_legacy_state - data_dir = Path("data") - session_dir = Path("session") + BASE_DIR = Path(__file__).resolve().parent + data_dir = BASE_DIR / "data" + session_dir = BASE_DIR / "session" try: if migrate_legacy_state(data_dir, session_dir): logger.info("Legacy migration completed successfully.") diff --git a/scraper_jobs.py b/scraper_jobs.py index 216d981..33e2443 100644 --- a/scraper_jobs.py +++ b/scraper_jobs.py @@ -1,11 +1,14 @@ import asyncio import logging +from pathlib import Path from typing import Any, Dict, List, Optional from app_state import StateStore logger = logging.getLogger(__name__) +BASE_DIR = Path(__file__).resolve().parent + class ScraperJobService: def __init__(self, state_store: StateStore): @@ -28,7 +31,7 @@ class ScraperJobService: # Extract account_id from payload, default to None (legacy) account_id: Optional[str] = payload.get("account_id") ScraperClass = self._import_scraper_class() - scraper = ScraperClass(account_id=account_id) + scraper = ScraperClass(account_id=account_id, base_dir=BASE_DIR) if account_id: from app_state import load_account diff --git a/telegram_scraper_with_forwarding.py b/telegram_scraper_with_forwarding.py index 8c4fbcf..d2d1d2f 100644 --- a/telegram_scraper_with_forwarding.py +++ b/telegram_scraper_with_forwarding.py @@ -2,6 +2,7 @@ import sqlite3 import json import csv import asyncio +import logging import time import sys import warnings @@ -27,6 +28,9 @@ from app_state import ( get_account_store, ) +BASE_DIR = Path(__file__).resolve().parent +logger = logging.getLogger(__name__) + warnings.filterwarnings( "ignore", message="Using async sessions support is an experimental feature" ) @@ -90,16 +94,18 @@ def _ensure_session_wal(session_path: str) -> None: class OptimizedTelegramScraper: - def __init__(self, account_id: Optional[str] = None): + def __init__(self, account_id: Optional[str] = None, base_dir: Optional[Path] = None): self.account_id = account_id - self.SESSION_DIR = Path("session") + base_dir = base_dir or BASE_DIR + self.BASE_DIR = base_dir + self.SESSION_DIR = base_dir / "session" self.SESSION_DIR.mkdir(exist_ok=True) if account_id: - self.DATA_DIR = Path("data") / "accounts" / account_id - self.state_store = get_account_store(Path("data"), account_id) + self.DATA_DIR = base_dir / "data" / "accounts" / account_id + self.state_store = get_account_store(base_dir / "data", account_id) else: - self.DATA_DIR = Path("data") + self.DATA_DIR = base_dir / "data" self.state_store = StateStore(self.DATA_DIR / "state.json") self.DATA_DIR.mkdir(parents=True, exist_ok=True) @@ -210,6 +216,22 @@ class OptimizedTelegramScraper: columns = {row[1] for row in cursor.fetchall()} migrations = [] + if "sender_id" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN sender_id INTEGER") + if "first_name" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN first_name TEXT") + if "last_name" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN last_name TEXT") + if "username" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN username TEXT") + if "message" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN message TEXT") + if "media_type" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN media_type TEXT") + if "media_path" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN media_path TEXT") + if "reply_to" not in columns: + migrations.append("ALTER TABLE messages ADD COLUMN reply_to INTEGER") if "post_author" not in columns: migrations.append("ALTER TABLE messages ADD COLUMN post_author TEXT") if "views" not in columns: @@ -222,8 +244,8 @@ class OptimizedTelegramScraper: for migration in migrations: try: conn.execute(migration) - except Exception: - pass + except Exception as e: + logger.warning("Migration failed for %s: %s", migration, e) if migrations: conn.commit() @@ -353,7 +375,7 @@ class OptimizedTelegramScraper: return False async def forward_message( - self, message, rule: ForwardingRule, source_channel_id: int = None + self, message, rule: ForwardingRule, source_channel_id: int = None, _retry: int = 0 ): try: dest_entity = await self._resolve_entity(rule.destination_channel) @@ -393,9 +415,12 @@ class OptimizedTelegramScraper: return True except FloodWaitError as e: + if _retry >= 3: + print(f" Failed to forward message {message.id}: FloodWait retry limit exceeded") + return False print(f" Rate limited, waiting {e.seconds}s...") await asyncio.sleep(e.seconds) - return await self.forward_message(message, rule, source_channel_id) + return await self.forward_message(message, rule, source_channel_id, _retry=_retry + 1) except Exception as e: print(f" Failed to forward message {message.id}: {e}") return False @@ -429,6 +454,11 @@ class OptimizedTelegramScraper: print("No valid source channels") return False + # Unregister a previously installed handler so it is never registered twice. + if self.forwarding_handler is not None: + self.client.remove_event_handler(self.forwarding_handler) + self.forwarding_handler = None + @self.client.on( events.NewMessage(chats=source_channels, incoming=True, outgoing=True) ) diff --git a/tests/test_integration.py b/tests/test_integration.py index 17a7457..2163814 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -14,6 +14,7 @@ They verify: - load_messages() pagination with search filter """ +import io import os import shutil import sqlite3 @@ -124,6 +125,29 @@ def create_channel_db(data_dir: Path, account_id: Optional[str], channel_id: str conn.close() +def _make_ws_handler(headers=None, body=b""): + """Build a bare webui_server request handler for unit-level checks. + + Avoids touching sockets - read_json_body()/_check_same_origin() only use + headers/rfile/send_error_json, which are stubbed here. + """ + import webui_server as ws_module + + hdrs = dict(headers or {}) + hdrs.setdefault("Content-Length", str(len(body))) + handler = object.__new__(ws_module.TelegramScraperRequestHandler) + handler.headers = hdrs + handler.rfile = io.BytesIO(body) + handler.wfile = io.BytesIO() + handler.path = "/api/test" + handler.command = "POST" + handler.client_address = ("127.0.0.1", 4321) + handler.server = MagicMock() + handler.send_error_json = MagicMock() + handler.send_json = MagicMock() + return handler + + # ── Fixture setup / teardown ──────────────────────────────────────────────── @@ -305,11 +329,29 @@ class TestInputSafety: redacted = ws_module.export_account_state(state) assert redacted["api_hash"] is None assert redacted["api_hash_present"] is True + # api_id is sensitive too and must never be exported + assert redacted["api_id"] is None + assert redacted["api_id_present"] is True assert state["api_hash"] == "secret" + assert state["api_id"] == 123 - full = ws_module.export_account_state(state, include_secrets=True) - assert full["api_hash"] == "secret" - assert "api_hash_present" not in full + def test_clean_continuous_channels_drops_invalid_and_normalizes(self): + """Continuous config ingest must reject path-traversal entries.""" + import webui_server as ws_module + + cleaned, dropped = ws_module.clean_continuous_channels([ + "@valid_name", "123", "nested/channel", r"nested\channel", "../escape", ".", "..", "", + ]) + # Valid entries pass through normalized (leading '@' stripped, numbers kept) + assert cleaned == ["valid_name", "123"] + # Invalid entries are dropped (not persisted), preserving order of appearance + assert dropped == ["nested/channel", r"nested\channel", "../escape", ".", "..", ""] + + def test_clean_continuous_channels_non_list_input(self): + import webui_server as ws_module + + assert ws_module.clean_continuous_channels(None) == ([], []) + assert ws_module.clean_continuous_channels("not-a-list") == ([], []) class TestContinuousOrchestrator: @@ -396,6 +438,74 @@ class TestContinuousOrchestrator: finally: self._restore_ws_data_dir() + def test_stop_and_join_waits_for_thread_then_flips_running(self): + """stop()+join() must let status['running'] become False only after the + worker thread truly exits, and join must be safe/idempotent on a + short-lived thread.""" + PerAccountContinuousScrapeManager, _, ws = self._import_orch_classes() + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 60, "channels": [], "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + # Keep the loop from doing anything slow: skip config reload and + # make auth check see a non-authorized account so it just waits on + # the stop event briefly. + mgr.refresh_config = lambda: None + mgr.join # ensure attribute exists + mgr.start() + assert mgr.status["running"] is True + assert mgr.thread is not None and mgr.thread.is_alive() + + mgr.stop() + # stop() only requests; running stays True until the thread exits. + assert mgr.status["running"] is True + + # Second stop() must be safe (idempotent). + mgr.stop() + + assert mgr.join(timeout=5.0) is True, "worker thread did not exit" + # After join, the thread's finally has flipped running to False. + assert mgr.status["running"] is False + + # join() on a dead/never-started thread is a no-op success. + fresh = PerAccountContinuousScrapeManager(aid) + assert fresh.join(timeout=1.0) is True + finally: + self._restore_ws_data_dir() + + def test_refresh_config_disable_requests_stop_without_lying_about_running(self): + """refresh_config() disabling the account must request a stop but must + NOT set status['running']=False (the worker thread owns that).""" + PerAccountContinuousScrapeManager, _, ws = self._import_orch_classes() + self._setup_ws_data_dir() + try: + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 60, "channels": [], "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + # Simulate a live worker by faking the running state and an enabled + # in-memory config, then flip the on-disk config to disabled. + with mgr.lock: + mgr.status["running"] = True + mgr.config["enabled"] = True + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": False, "interval_minutes": 60, "channels": [], "run_all_tracked": True, + }) + mgr.refresh_config() + assert mgr.stop_event.is_set(), "refresh_config should request stop" + # running is NOT touched by refresh_config — the loop's finally sets it. + assert mgr.status["running"] is True + # After the worker would exit, running flips to False (simulated here). + with mgr.lock: + mgr.status["running"] = False + assert mgr.status["running"] is False + finally: + self._restore_ws_data_dir() + class TestMessageSearch: """load_messages() SQL correctness with pagination and search.""" @@ -576,7 +686,6 @@ class TestJobDeduplicationSchema: runner = JobRunner() runner.queue.put = lambda job: None j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) - time.sleep(0.002) # ensure different timestamp -> different job_id j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc2"}) assert j1.job_id != j2.job_id runner.shutdown(timeout=0) @@ -587,7 +696,6 @@ class TestJobDeduplicationSchema: runner.queue.put = lambda job: None j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) j1.status = "done" - time.sleep(0.002) # ensure different timestamp -> different job_id j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"}) assert j1.job_id != j2.job_id runner.shutdown(timeout=0) @@ -626,6 +734,179 @@ class TestAccountHealthSummary: assert "media_count" in health +class TestSecurityHardening: + """Regression tests for the security / hardening fixes. + + - StateStore.load() must never hand callers a reference to its internal + cache (H-6) + - read_json_body() size cap + content-type rejection (H-4) + - same-origin enforcement on state-mutating / SSE endpoints (H-4) + - auth rate limiter: lockout on failures + cooldown on code requests (H-5) + - terminal job statuses include "done" so SSE streams terminate (C-2) + """ + + def _ws_module(self): + import webui_server as ws_module + return ws_module + + def _clear_auth_attempts(self): + ws = self._ws_module() + with ws._auth_attempts_lock: + ws._auth_attempts.clear() + + def test_state_store_load_returns_independent_copies(self): + from app_state import StateStore + + defaults = {"accounts": [], "nested": {"x": 1}} + state_path = TEST_DATA / "indep-copy" / "state.json" + state_path.parent.mkdir(parents=True, exist_ok=True) + if state_path.exists(): + state_path.unlink() + + # File-not-exists path -> defaults; mutating the first result must not + # poison the second load() within the TTL window. + store = StateStore(state_path, defaults=defaults) + first = store.load() + first["accounts"].append("mutated") + first["nested"]["x"] = 999 + assert store.load()["accounts"] == [] + assert store.load()["nested"]["x"] == 1 + + # Merge path (real file); use a fresh instance to bypass the TTL cache. + state_path.write_text( + '{"accounts": ["a"], "nested": {"x": 2, "y": 3}}', encoding="utf-8" + ) + store = StateStore(state_path, defaults=defaults) + third = store.load() + assert third["accounts"] == ["a"] + third["nested"]["y"] = 999 + reloaded = store.load() + assert reloaded["accounts"] == ["a"] + assert reloaded["nested"] == {"x": 2, "y": 3} + + # Invalid JSON -> defaults fallback must also return an independent copy. + state_path.write_text("{not valid json", encoding="utf-8") + store = StateStore(state_path, defaults=defaults) + fallback = store.load() + fallback["nested"]["x"] = 500 + reloaded = store.load() + assert reloaded["nested"]["x"] == 1 + + def test_read_json_body_rejects_non_json_and_oversized(self): + ws = self._ws_module() + + # non-JSON Content-Type -> sentinel (do_POST maps it to 415) + handler = _make_ws_handler( + headers={"Content-Type": "text/plain"}, body=b"hello" + ) + assert handler.read_json_body() is ws._JSON_CONTENT_TYPE_REJECTED + + # declared Content-Length over the cap -> sentinel (do_POST maps to 413) + oversized = str(ws.MAX_JSON_BODY_BYTES + 1) + handler = _make_ws_handler( + headers={"Content-Length": oversized, "Content-Type": "application/json"}, + body=b"{}", + ) + assert handler.read_json_body() is ws._JSON_BODY_TOO_LARGE + + # valid application/json body parses normally + payload = b'{"a": 1}' + handler = _make_ws_handler( + headers={"Content-Type": "application/json; charset=utf-8"}, + body=payload, + ) + assert handler.read_json_body() == {"a": 1} + + # unparseable Content-Length -> treated as no body (no 500) + handler = _make_ws_handler( + headers={"Content-Length": "garbage", "Content-Type": "application/json"}, + body=b"{}", + ) + assert handler.read_json_body() == {} + + def test_check_same_origin_rejects_cross_origin(self): + ws = self._ws_module() + + # cross-origin Origin header -> rejected (403 response sent) + h = _make_ws_handler( + headers={"Host": "localhost:8080", "Origin": "http://evil.example"} + ) + assert h._check_same_origin() is False + h.send_error_json.assert_called_once() + + # Sec-Fetch-Site: cross-site -> rejected + h2 = _make_ws_handler( + headers={"Host": "localhost:8080", "Sec-Fetch-Site": "cross-site"} + ) + assert h2._check_same_origin() is False + h2.send_error_json.assert_called_once() + + # same-origin Origin + Sec-Fetch-Site -> allowed + h3 = _make_ws_handler( + headers={ + "Host": "localhost:8080", + "Origin": "http://localhost:8080", + "Sec-Fetch-Site": "same-origin", + } + ) + assert h3._check_same_origin() is True + + # no Origin / Sec-Fetch-Site headers (curl, same-origin) -> allowed + h4 = _make_ws_handler(headers={"Host": "localhost:8080"}) + assert h4._check_same_origin() is True + + def test_auth_rate_limiter_lockout_and_code_cooldown(self): + ws = self._ws_module() + self._clear_auth_attempts() + try: + # Lockout after AUTH_MAX_FAILED_ATTEMPTS failures. + ip, acc = "10.0.0.5", "acc1" + for _ in range(ws.AUTH_MAX_FAILED_ATTEMPTS): + ws._record_auth_failure(ip, acc) + assert ws._check_auth_throttle(ip, acc) is False + + # Lockout expires -> attempts allowed again. + with ws._auth_attempts_lock: + ws._auth_attempts[(ip, acc)]["locked_until"] = time.time() - 1 + assert ws._check_auth_throttle(ip, acc) is True + + # A successful code request starts a cooldown for (ip, account). + ip2 = "10.0.0.6" + ws._record_auth_code_request(ip2, acc) + assert ws._check_auth_code_cooldown(ip2, acc) is False + # Different client IP is unaffected. + assert ws._check_auth_code_cooldown("10.0.0.7", acc) is True + + # Cooldown expires -> allowed again. + with ws._auth_attempts_lock: + ws._auth_attempts[(ip2, acc)]["cooldown_until"] = time.time() - 1 + assert ws._check_auth_code_cooldown(ip2, acc) is True + finally: + self._clear_auth_attempts() + + def test_auth_attempts_sweep_evicts_expired_entries(self): + ws = self._ws_module() + self._clear_auth_attempts() + try: + with ws._auth_attempts_lock: + # Grow past the cap with only already-expired lockouts. + for i in range(ws.AUTH_ATTEMPTS_MAX_ENTRIES + 2): + ws._auth_attempts[("10.99.0.1", str(i))] = { + "failures": 0, + "locked_until": time.time() - 30, + "cooldown_until": None, + } + assert len(ws._auth_attempts) > ws.AUTH_ATTEMPTS_MAX_ENTRIES + ws._check_auth_throttle("10.99.0.1", "0") # triggers the sweep + assert len(ws._auth_attempts) < ws.AUTH_ATTEMPTS_MAX_ENTRIES + finally: + self._clear_auth_attempts() + + def test_terminal_job_statuses_include_done(self): + ws = self._ws_module() + assert "done" in ws.TERMINAL_JOB_STATUSES + + # ── Cleanup all temp data ────────────────────────────────────────────────── diff --git a/webui/app.js b/webui/app.js index cb82894..9081196 100644 --- a/webui/app.js +++ b/webui/app.js @@ -411,7 +411,7 @@ function pollJobFallback(accountId, jobId) { try { const job = await api(`/api/jobs/${encodeURIComponent(jobId)}`); updateRenderedJob(accountId, job); - if (['completed', 'failed'].includes(job.status)) { + if (['done', 'completed', 'failed'].includes(job.status)) { clearInterval(pollTimer); jobStreams.delete(jobId); refreshAccount(accountId); @@ -455,7 +455,7 @@ function subscribeJobStream(accountId, jobId, status) { retryCount = 0; // reset backoff on successful message const job = JSON.parse(event.data); updateRenderedJob(accountId, job); - if (['completed', 'failed'].includes(job.status)) { + if (['done', 'completed', 'failed'].includes(job.status)) { newStream.close(); jobStreams.delete(jobId); refreshAccount(accountId); diff --git a/webui_server.py b/webui_server.py index d3b64aa..4d4e17b 100644 --- a/webui_server.py +++ b/webui_server.py @@ -14,12 +14,13 @@ import threading import time import traceback import urllib.parse +import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple import qrcode import qrcode.image.svg @@ -54,6 +55,116 @@ SCRAPER_JOBS = ScraperJobService(STATE_STORE) logger = logging.getLogger(__name__) +# Job statuses that are terminal: once a job reaches one of these states the +# job is considered finished and event streams / polling should stop. +TERMINAL_JOB_STATUSES = {"done", "failed", "completed"} + +# Maximum number of concurrently open SSE event streams. Each stream pins a +# handler thread for up to 30 minutes, so cap them and close the oldest. +MAX_EVENT_STREAMS = 10 + +# How long ContinuousScrapeOrchestrator.remove_account waits for a scrape +# worker thread to exit before deleting account data best-effort (seconds). +REMOVE_ACCOUNT_JOIN_TIMEOUT = 20.0 + +# Sentinel returned by read_json_body() when a request carries a body with a +# non-JSON Content-Type - maps to a 415 Unsupported Media Type response. +_JSON_CONTENT_TYPE_REJECTED = object() + +# Sentinel returned by read_json_body() when the declared/actual body size +# exceeds MAX_JSON_BODY_BYTES - maps to a 413 Payload Too Large response. +_JSON_BODY_TOO_LARGE = object() +MAX_JSON_BODY_BYTES = 1_048_576 # 1 MB + +# ── Auth throttling ───────────────────────────────────────────────────── +# Repeated failed submissions on the phone-code / 2FA-password endpoints are +# throttled per (client IP, account): AUTH_MAX_FAILED_ATTEMPTS failures lock +# the pair out for AUTH_LOCKOUT_SECONDS. +AUTH_MAX_FAILED_ATTEMPTS = 5 +AUTH_LOCKOUT_SECONDS = 60 +# Minimum interval (seconds) between successful phone-code REQUEST calls for +# the same (client IP, account) pair. Covers the success path so a LAN client +# cannot loop request_phone_code and flood the victim's phone with SMS. +AUTH_CODE_REQUEST_COOLDOWN_SECONDS = 30 +# Cap the in-memory auth tracking dict; beyond this we sweep expired entries. +AUTH_ATTEMPTS_MAX_ENTRIES = 10_000 +_auth_attempts: Dict[Tuple[str, str], Dict[str, Any]] = {} +_auth_attempts_lock = threading.Lock() + + +def _sweep_auth_attempts(now: float) -> None: + """Evict expired lockout/cooldown entries when the dict grows too large.""" + global _auth_attempts + if len(_auth_attempts) <= AUTH_ATTEMPTS_MAX_ENTRIES: + return + _auth_attempts = { + k: v + for k, v in _auth_attempts.items() + if (v.get("locked_until") or 0) > now or (v.get("cooldown_until") or 0) > now + } + + +def _check_auth_throttle(ip: str, account_id: str) -> bool: + """Return True if the attempt is allowed, False if currently locked out.""" + now = time.time() + with _auth_attempts_lock: + _sweep_auth_attempts(now) + entry = _auth_attempts.get((ip, account_id)) + if not entry: + return True + locked_until = entry.get("locked_until") + if locked_until and now < locked_until: + return False + if locked_until and now >= locked_until: + # Lockout window expired - reset the counter. + _auth_attempts.pop((ip, account_id), None) + return True + + +def _check_auth_code_cooldown(ip: str, account_id: str) -> bool: + """Return True if a code REQUEST is allowed, False if in cooldown.""" + now = time.time() + with _auth_attempts_lock: + _sweep_auth_attempts(now) + entry = _auth_attempts.get((ip, account_id)) + if not entry: + return True + cooldown_until = entry.get("cooldown_until") + if cooldown_until and now < cooldown_until: + return False + return True + + +def _record_auth_failure(ip: str, account_id: str) -> None: + now = time.time() + with _auth_attempts_lock: + _sweep_auth_attempts(now) + entry = _auth_attempts.setdefault( + (ip, account_id), {"failures": 0, "locked_until": None} + ) + locked_until = entry.get("locked_until") + if locked_until is not None and now < locked_until: + return + entry["failures"] = entry.get("failures", 0) + 1 + if entry["failures"] >= AUTH_MAX_FAILED_ATTEMPTS: + entry["locked_until"] = now + AUTH_LOCKOUT_SECONDS + entry["failures"] = 0 + + +def _record_auth_success(ip: str, account_id: str) -> None: + with _auth_attempts_lock: + _auth_attempts.pop((ip, account_id), None) + + +def _record_auth_code_request(ip: str, account_id: str) -> None: + """Record a successful phone-code request to start its cooldown.""" + now = time.time() + with _auth_attempts_lock: + entry = _auth_attempts.setdefault( + (ip, account_id), {"failures": 0, "locked_until": None} + ) + entry["cooldown_until"] = now + AUTH_CODE_REQUEST_COOLDOWN_SECONDS + def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -112,11 +223,41 @@ def normalize_channel_id(value: Any) -> str: return channel_id -def export_account_state(state: Dict[str, Any], include_secrets: bool = False) -> Dict[str, Any]: +def clean_continuous_channels(channels: Any) -> Tuple[List[str], List[str]]: + """Validate/normalize a continuous-scrape channel list at ingest. + + Returns ``(cleaned, dropped)`` where ``cleaned`` holds the normalized, + path-safe channel ids and ``dropped`` holds the raw values that were + rejected (invalid path characters, control chars, ``.``/``..``). The + dropped entries must NOT be persisted so they cannot become a path + traversal vector for any future direct read. + """ + cleaned: List[str] = [] + dropped: List[str] = [] + if not isinstance(channels, list): + return cleaned, dropped + for item in channels: + raw = str(item).strip() + try: + cleaned.append(normalize_channel_id(raw)) + except ValueError: + dropped.append(raw) + return cleaned, dropped + + +def export_account_state(state: Dict[str, Any]) -> Dict[str, Any]: + """Return a sanitized copy of an account state for export. + + Credentials (api_hash, api_id) are always redacted - they are never + included in exports, only a presence flag is exposed. + """ exported = dict(state) - if not include_secrets and "api_hash" in exported: + if "api_hash" in exported: exported["api_hash_present"] = bool(exported.get("api_hash")) exported["api_hash"] = None + if "api_id" in exported: + exported["api_id_present"] = bool(exported.get("api_id")) + exported["api_id"] = None return exported @@ -345,9 +486,33 @@ class JobRunner: self.queue: "queue.Queue[Job]" = queue.Queue() self.lock = threading.Lock() self._shutdown_flag = False + # Active SSE event streams: stream_id -> started_at, plus any stream + # ids that have been revoked (told to close) because the cap was hit. + self._stream_started: Dict[str, float] = {} + self._stream_revoked: set = set() self.worker = threading.Thread(target=self._run, daemon=True) self.worker.start() + def register_event_stream(self) -> str: + """Register an active SSE stream; revoke the oldest when over the cap.""" + stream_id = uuid.uuid4().hex + with self.lock: + self._stream_started[stream_id] = time.time() + if len(self._stream_started) > MAX_EVENT_STREAMS: + oldest = min(self._stream_started, key=lambda sid: self._stream_started[sid]) + self._stream_started.pop(oldest, None) + self._stream_revoked.add(oldest) + return stream_id + + def unregister_event_stream(self, stream_id: str) -> None: + with self.lock: + self._stream_started.pop(stream_id, None) + self._stream_revoked.discard(stream_id) + + def is_stream_revoked(self, stream_id: str) -> bool: + with self.lock: + return stream_id in self._stream_revoked + def create_job(self, job_type: str, title: str, payload: Dict[str, Any]) -> Job: if self._shutdown_flag: raise RuntimeError("Server is shutting down, cannot create new jobs") @@ -366,7 +531,7 @@ class JobRunner: + f"[{datetime.now().strftime('%H:%M:%S')}] Reused existing active job for this account." ).strip() return existing - job_id = f"job-{int(time.time() * 1000)}" + job_id = f"job-{uuid.uuid4().hex[:12]}" job = Job( job_id=job_id, job_type=job_type, @@ -459,7 +624,12 @@ class JobRunner: class TelegramAuthManager: def __init__(self) -> None: - self.lock = threading.Lock() + # Reentrant lock guarding self.auth_data / self.clients. These dicts + # are mutated from the async event-loop thread (coroutines) and read + # from HTTP handler threads, so every access must hold the lock. + # RLock allows nested acquisition from helpers (_get_auth_data / + # _set_state) called inside compound locked operations. + self.lock = threading.RLock() self.loop = asyncio.new_event_loop() self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() @@ -475,26 +645,28 @@ class TelegramAuthManager: return future.result() def _get_auth_data(self, account_id: str) -> Dict[str, Any]: - if account_id not in self.auth_data: - self.auth_data[account_id] = { - "phase": "idle", - "status": "unknown", - "details": "", - "qr_url": None, - "qr_image": None, - "phone": None, - "phone_code_hash": None, - "qr_login": None, - "qr_wait_task": None, - "user_id": None, - "updated_at": utc_now_iso(), - } - return self.auth_data[account_id] + with self.lock: + if account_id not in self.auth_data: + self.auth_data[account_id] = { + "phase": "idle", + "status": "unknown", + "details": "", + "qr_url": None, + "qr_image": None, + "phone": None, + "phone_code_hash": None, + "qr_login": None, + "qr_wait_task": None, + "user_id": None, + "updated_at": utc_now_iso(), + } + return self.auth_data[account_id] def _set_state(self, account_id: str, **updates: Any) -> None: - data = self._get_auth_data(account_id) - data.update(updates) - data["updated_at"] = utc_now_iso() + with self.lock: + data = self._get_auth_data(account_id) + data.update(updates) + data["updated_at"] = utc_now_iso() def _make_qr_image(self, qr_url: str) -> str: qr = qrcode.QRCode(border=1, box_size=8) @@ -512,32 +684,39 @@ class TelegramAuthManager: api_hash = acc_state.get("api_hash") if not api_id or not api_hash: raise RuntimeError("Save api_id and api_hash first for this account.") - if account_id not in self.clients or self.clients[account_id] is None: - _ensure_session_wal(account_session_path(SESSION_DIR, account_id)) - self.clients[account_id] = TelegramClient( - account_session_path(SESSION_DIR, account_id), - api_id, - api_hash, - ) - client = self.clients[account_id] + with self.lock: + if account_id not in self.clients or self.clients[account_id] is None: + _ensure_session_wal(account_session_path(SESSION_DIR, account_id)) + self.clients[account_id] = TelegramClient( + account_session_path(SESSION_DIR, account_id), + api_id, + api_hash, + ) + client = self.clients[account_id] if not client.is_connected(): await client.connect() - data = self._get_auth_data(account_id) - if data.get("user_id") is None and await client.is_user_authorized(): + with self.lock: + needs_user_id = self._get_auth_data(account_id).get("user_id") is None + if needs_user_id and await client.is_user_authorized(): try: me = await client.get_me() - data["user_id"] = me.id except Exception: - pass + me = None + if me is not None: + with self.lock: + data = self._get_auth_data(account_id) + if data.get("user_id") is None: + data["user_id"] = me.id return client def auth_state(self, account_id: str) -> Dict[str, Any]: - data = self._get_auth_data(account_id) - acc_state = load_account(DATA_DIR, account_id) - snapshot = dict(data) + with self.lock: + data = self._get_auth_data(account_id) + snapshot = dict(data) snapshot.pop("qr_login", None) snapshot.pop("qr_wait_task", None) snapshot.pop("phone_code_hash", None) + acc_state = load_account(DATA_DIR, account_id) snapshot["saved_credentials"] = { "api_id": acc_state.get("api_id"), "api_hash_present": bool(acc_state.get("api_hash")), @@ -568,7 +747,6 @@ class TelegramAuthManager: async def _start_qr_login(self, account_id: str) -> Dict[str, Any]: client = await self._get_client(account_id) - data = self._get_auth_data(account_id) if await client.is_user_authorized(): self._set_state( account_id, @@ -579,7 +757,10 @@ class TelegramAuthManager: return self.auth_state(account_id) qr_login = await client.qr_login() qr_url = qr_login.url - data["qr_login"] = qr_login + with self.lock: + data = self._get_auth_data(account_id) + data["qr_login"] = qr_login + data["updated_at"] = utc_now_iso() self._set_state( account_id, phase="qr_waiting", @@ -588,14 +769,18 @@ class TelegramAuthManager: qr_url=qr_url, qr_image=self._make_qr_image(qr_url), ) - data["qr_wait_task"] = self.loop.create_task( - self._wait_for_qr_login(account_id) - ) + with self.lock: + data = self._get_auth_data(account_id) + data["qr_wait_task"] = self.loop.create_task( + self._wait_for_qr_login(account_id) + ) + data["updated_at"] = utc_now_iso() return self.auth_state(account_id) async def _wait_for_qr_login(self, account_id: str) -> None: - data = self._get_auth_data(account_id) - qr_login = data.get("qr_login") + with self.lock: + data = self._get_auth_data(account_id) + qr_login = data.get("qr_login") if not qr_login: return try: @@ -631,7 +816,6 @@ class TelegramAuthManager: async def _request_phone_code(self, account_id: str, phone: str) -> Dict[str, Any]: client = await self._get_client(account_id) - data = self._get_auth_data(account_id) if await client.is_user_authorized(): self._set_state( account_id, @@ -641,14 +825,13 @@ class TelegramAuthManager: ) return self.auth_state(account_id) sent = await client.send_code_request(phone) - data["phone"] = phone - data["phone_code_hash"] = sent.phone_code_hash self._set_state( account_id, phase="code_required", status="code_required", details=f"Code sent to {phone}. Enter it below.", phone=phone, + phone_code_hash=sent.phone_code_hash, ) return self.auth_state(account_id) @@ -657,14 +840,17 @@ class TelegramAuthManager: async def _submit_phone_code(self, account_id: str, code: str) -> Dict[str, Any]: client = await self._get_client(account_id) - data = self._get_auth_data(account_id) - if not data.get("phone") or not data.get("phone_code_hash"): + with self.lock: + data = self._get_auth_data(account_id) + phone = data.get("phone") + phone_code_hash = data.get("phone_code_hash") + if not phone or not phone_code_hash: raise RuntimeError("Request a phone code first.") try: await client.sign_in( - phone=data["phone"], + phone=phone, code=code, - phone_code_hash=data["phone_code_hash"], + phone_code_hash=phone_code_hash, ) self._set_state( account_id, @@ -703,8 +889,9 @@ class TelegramAuthManager: return self._run(self._submit_password(account_id, password)) def delete_account(self, account_id: str) -> None: - if account_id in self.clients: - client = self.clients[account_id] + with self.lock: + client = self.clients.get(account_id) + if client is not None: try: future = asyncio.run_coroutine_threadsafe( self._disconnect_client(client), self.loop @@ -712,23 +899,27 @@ class TelegramAuthManager: future.result(timeout=5) except Exception: pass - del self.clients[account_id] - self.auth_data.pop(account_id, None) + with self.lock: + self.clients.pop(account_id, None) + self.auth_data.pop(account_id, None) async def _disconnect_client(self, client: TelegramClient) -> None: if client and client.is_connected(): await client.disconnect() def shutdown(self, timeout: float = 5.0) -> None: + with self.lock: + clients = list(self.clients.values()) + async def _disconnect_all(): - for client in self.clients.values(): + for client in clients: try: if client and client.is_connected(): await client.disconnect() except Exception: pass - if self.clients: + if clients: try: future = asyncio.run_coroutine_threadsafe(_disconnect_all(), self.loop) future.result(timeout=timeout) @@ -799,13 +990,15 @@ class PerAccountContinuousScrapeManager: if str(item).strip() ] self.config["run_all_tracked"] = bool(disk_cfg.get("run_all_tracked", True)) - # Sync running state with desired enabled state + # Sync running state with desired enabled state. We only request a + # stop here — the worker thread flips status["running"] to False in + # its finally block only once it actually exits, so the status flag + # never lies about a still-running scrape iteration. if self.config["enabled"] and not self.status["running"]: pass # don't auto-start — user must call start() elif not self.config["enabled"] and self.status["running"]: self._log("Continuous disabled via external state change, stopping.", "warn") self.stop_event.set() - self.status["running"] = False self._log("Config refreshed from disk.", "debug") def _log(self, message: str, level: str = "debug") -> None: @@ -848,7 +1041,10 @@ class PerAccountContinuousScrapeManager: "run_all_tracked": bool(run_all_tracked), } self._save_config() - self.config = self._load_config() + # Re-read from disk under the same lock so config assignment is + # atomic w.r.t. refresh_config() (which also mutates config under + # the lock), avoiding a torn read / write race. + self.config = self._load_config() if enabled: self.start() else: @@ -864,17 +1060,36 @@ class PerAccountContinuousScrapeManager: self.status["running"] = True self.status["last_started_at"] = utc_now_iso() self.status["last_error"] = None - self.thread = threading.Thread(target=self._run_loop, daemon=True) + self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() self._log("Continuous scraping started.", "info") def stop(self) -> None: + # Idempotent: only the first call logs. We deliberately do NOT set + # status["running"] = False here — the worker thread flips it once it + # has actually exited (see _run_loop's finally), so status reflects + # reality rather than intent. + already_set = self.stop_event.is_set() self.stop_event.set() - with self.lock: - was_running = self.status["running"] - self.status["running"] = False - if was_running: - self._log("Continuous scraping stop requested.", "warn") + if not already_set: + with self.lock: + was_running = self.status["running"] + if was_running: + self._log("Continuous scraping stop requested.", "warn") + + def join(self, timeout: float = 20.0) -> bool: + """Wait up to ``timeout`` seconds for the worker thread to exit. + + Safe to call when the thread was never started or is already dead. + Returns True if the thread finished within the timeout, False if it is + still running (e.g. mid-scrape) and the caller should proceed + best-effort. + """ + thread = self.thread + if thread is None or not thread.is_alive(): + return True + thread.join(timeout=timeout) + return not thread.is_alive() def _resolve_channels(self) -> List[str]: acc_state = load_account(DATA_DIR, self.account_id) @@ -887,66 +1102,73 @@ class PerAccountContinuousScrapeManager: return [channel for channel in configured if channel in tracked] def _run_loop(self) -> None: - while not self.stop_event.is_set(): - # Refresh config from disk so channel / setting changes take effect - self.refresh_config() + try: + while not self.stop_event.is_set(): + # Refresh config from disk so channel / setting changes take effect + self.refresh_config() + # Bail promptly if refresh/cancel requested the stop so join() + # usually returns quickly instead of waiting out a full scrape. + if self.stop_event.is_set(): + break - # Check auth — don't iterate if account isn't authorized - auth_info = auth_status_for(self.account_id) - if auth_info.get("status") not in ("ready", "authorized"): - self._log( - f"Account not authorized (status={auth_info.get('status')}), " - "skipping iteration", - "warn", - ) - sleep_seconds = max(5, 60) + # Check auth — don't iterate if account isn't authorized + auth_info = auth_status_for(self.account_id) + if auth_info.get("status") not in ("ready", "authorized"): + self._log( + f"Account not authorized (status={auth_info.get('status')}), " + "skipping iteration", + "warn", + ) + sleep_seconds = max(5, 60) + interrupted = self.stop_event.wait(timeout=sleep_seconds) + if interrupted: + break + continue + + channels = self._resolve_channels() + cfg = self.snapshot()["config"] + interval_minutes = cfg.get("interval_minutes", 1) + + if not channels: + self._log("No channels configured for continuous scraping.", "warn") + else: + self._log(f"Starting iteration for {len(channels)} channel(s).", "info") + buffer = io.StringIO() + try: + with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): + run_job("scrape_selected", { + "channels": channels, + "account_id": self.account_id, + }) + output = buffer.getvalue().strip() + if output: + for line in output.splitlines(): + self._log(line) + with self.lock: + self.status["last_iteration_at"] = utc_now_iso() + self.status["last_finished_at"] = utc_now_iso() + self.status["last_error"] = None + self._log("Iteration finished.", "success") + except Exception as exc: + output = buffer.getvalue().strip() + if output: + for line in output.splitlines(): + self._log(line) + self._log(f"Iteration failed: {exc}", "error") + with self.lock: + self.status["last_error"] = str(exc) + + sleep_seconds = max(5, interval_minutes * 60) + self._log(f"Sleeping for {interval_minutes} minute(s).", "debug") interrupted = self.stop_event.wait(timeout=sleep_seconds) if interrupted: break - continue - - channels = self._resolve_channels() - cfg = self.snapshot()["config"] - interval_minutes = cfg.get("interval_minutes", 1) - - if not channels: - self._log("No channels configured for continuous scraping.", "warn") - else: - self._log(f"Starting iteration for {len(channels)} channel(s).", "info") - buffer = io.StringIO() - try: - with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): - run_job("scrape_selected", { - "channels": channels, - "account_id": self.account_id, - }) - output = buffer.getvalue().strip() - if output: - for line in output.splitlines(): - self._log(line) - with self.lock: - self.status["last_iteration_at"] = utc_now_iso() - self.status["last_finished_at"] = utc_now_iso() - self.status["last_error"] = None - self._log("Iteration finished.", "success") - except Exception as exc: - output = buffer.getvalue().strip() - if output: - for line in output.splitlines(): - self._log(line) - self._log(f"Iteration failed: {exc}", "error") - with self.lock: - self.status["last_error"] = str(exc) - - sleep_seconds = max(5, interval_minutes * 60) - self._log(f"Sleeping for {interval_minutes} minute(s).", "debug") - interrupted = self.stop_event.wait(timeout=sleep_seconds) - if interrupted: - break - - with self.lock: - self.status["running"] = False - self._log("Continuous scraping stopped.", "warn") + finally: + # Only mark running=False once the thread has truly exited so the + # status reflects reality (a still-running scrape is not "stopped"). + with self.lock: + self.status["running"] = False + self._log("Continuous scraping stopped.", "warn") # ── ContinuousScrapeOrchestrator ───────────────────────────────────────── @@ -1012,7 +1234,22 @@ class ContinuousScrapeOrchestrator: self.start_account(account_id) def remove_account(self, account_id: str) -> None: - self.stop_account(account_id) + with self.lock: + mgr = self.managers.get(account_id) + if mgr: + mgr.stop() + # Wait for the scrape thread to actually stop before the caller + # deletes the account directory / session files, so rmtree does not + # race with a writer mid-iteration. If the thread is still running + # (e.g. mid-scrape) after the timeout we proceed best-effort and + # log a warning. + if not mgr.join(timeout=REMOVE_ACCOUNT_JOIN_TIMEOUT): + logger.warning( + "Continuous scrape thread for account %r still running after " + "%.1fs; removing account data best-effort", + account_id, + REMOVE_ACCOUNT_JOIN_TIMEOUT, + ) with self.lock: self.managers.pop(account_id, None) @@ -1841,6 +2078,13 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if path == "/api/jobs": return self.send_json(self.app.job_runner.recent_jobs()) if path.startswith("/api/jobs/") and path.endswith("/events"): + # SSE streams pin a worker thread up to 30 minutes, so reject + # cross-site / cross-origin clients (e.g. tags on other + # pages) before opening the stream. Requests without an Origin / + # Sec-Fetch-Site header (curl, same-origin EventSource that omits + # it) are allowed - see _check_same_origin(). + if not self._check_same_origin(): + return job_id = path.split("/")[-2] return self.stream_job_events(job_id) if path.startswith("/api/jobs/"): @@ -1936,13 +2180,11 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if sub == ["health"]: return self.send_json(account_health_summary(account_id, self.app.job_runner)) if sub == ["export"]: - include_secrets = query.get("include_secrets", ["0"])[0].lower() in {"1", "true", "yes"} - payload = { + return self.send_json({ "version": 1, "account_id": account_id, - "state": export_account_state(load_account(DATA_DIR, account_id), include_secrets=include_secrets), - } - return self.send_json(payload) + "state": export_account_state(load_account(DATA_DIR, account_id)), + }) if sub == ["channels"]: return self._handle_get_account_channels(account_id) if len(sub) >= 3 and sub[0] == "channels" and sub[-1] == "messages": @@ -1978,29 +2220,37 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): def stream_job_events(self, job_id: str) -> None: if not self.app.job_runner.get_job(job_id): return self.send_error_json(HTTPStatus.NOT_FOUND, "Job not found") - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", "text/event-stream") - self.send_header("Cache-Control", "no-cache") - self.send_header("Connection", "keep-alive") - self.end_headers() + runner = self.app.job_runner + stream_id = runner.register_event_stream() + try: + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "keep-alive") + self.end_headers() - last_payload = None - deadline = time.time() + 60 * 30 - while time.time() < deadline: - job = self.app.job_runner.get_job(job_id) - if not job: - break - payload = json.dumps(job, ensure_ascii=False) - if payload != last_payload: - try: - self.wfile.write(f"data: {payload}\n\n".encode("utf-8")) - self.wfile.flush() - except (BrokenPipeError, ConnectionResetError, OSError): + last_payload = None + deadline = time.time() + 60 * 30 + while time.time() < deadline: + if runner.is_stream_revoked(stream_id): + # Capped: this stream is the oldest and must close. break - last_payload = payload - if job.get("status") in {"completed", "failed"}: - break - time.sleep(1) + job = runner.get_job(job_id) + if not job: + break + payload = json.dumps(job, ensure_ascii=False) + if payload != last_payload: + try: + self.wfile.write(f"data: {payload}\n\n".encode("utf-8")) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError): + break + last_payload = payload + if job.get("status") in TERMINAL_JOB_STATUSES: + break + time.sleep(1) + finally: + runner.unregister_event_stream(stream_id) def _handle_get_account_channels(self, account_id: str) -> None: return self.send_json(list_channels_snapshot(account_id)) @@ -2075,9 +2325,15 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): # ── POST ───────────────────────────────────────────────────────────── def do_POST(self) -> None: + if not self._check_same_origin(): + return parsed = urllib.parse.urlparse(self.path) path = parsed.path body = self.read_json_body() + if body is _JSON_CONTENT_TYPE_REJECTED: + return self.send_error_json(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "Content-Type must be application/json") + if body is _JSON_BODY_TOO_LARGE: + return self.send_error_json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "Request body too large") if body is None: return self.send_error_json(HTTPStatus.BAD_REQUEST, "Expected JSON body") @@ -2097,11 +2353,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): try: enabled = bool(body.get("enabled")) interval_minutes = int(body.get("interval_minutes", 1)) - channels = [ - str(item).strip() - for item in body.get("channels", []) - if str(item).strip() - ] + channels, dropped = clean_continuous_channels(body.get("channels", [])) run_all_tracked = bool(body.get("run_all_tracked", True)) if self.app.legacy_account_id: payload = self.app.continuous_orchestrator.update_for( @@ -2115,6 +2367,8 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): payload = {"config": {}, "status": {"running": False, "logs": [], "log_entries": []}} except Exception as exc: return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + if isinstance(payload, dict): + payload["dropped_invalid"] = dropped return self.send_json(payload) if path == "/api/auth/credentials": @@ -2421,33 +2675,62 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): return self.send_json(payload) def _handle_post_account_auth_phone_request(self, account_id: str, body: Dict[str, Any]) -> None: + ip = self.client_address[0] if self.client_address else "" + if not _check_auth_throttle(ip, account_id): + return self.send_error_json( + HTTPStatus.TOO_MANY_REQUESTS, + "Too many failed attempts. Try again later.", + ) + if not _check_auth_code_cooldown(ip, account_id): + return self.send_error_json( + HTTPStatus.TOO_MANY_REQUESTS, + "Please wait before requesting another code.", + ) phone = str(body.get("phone", "")).strip() if not phone: return self.send_error_json(HTTPStatus.BAD_REQUEST, "phone is required") try: payload = self.app.auth_manager.request_phone_code(account_id, phone) except Exception as exc: + _record_auth_failure(ip, account_id) return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + _record_auth_code_request(ip, account_id) return self.send_json(payload) def _handle_post_account_auth_phone_submit(self, account_id: str, body: Dict[str, Any]) -> None: + ip = self.client_address[0] if self.client_address else "" + if not _check_auth_throttle(ip, account_id): + return self.send_error_json( + HTTPStatus.TOO_MANY_REQUESTS, + "Too many failed attempts. Try again later.", + ) code = str(body.get("code", "")).strip() if not code: return self.send_error_json(HTTPStatus.BAD_REQUEST, "code is required") try: payload = self.app.auth_manager.submit_phone_code(account_id, code) except Exception as exc: + _record_auth_failure(ip, account_id) return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + _record_auth_success(ip, account_id) return self.send_json(payload) def _handle_post_account_auth_password(self, account_id: str, body: Dict[str, Any]) -> None: + ip = self.client_address[0] if self.client_address else "" + if not _check_auth_throttle(ip, account_id): + return self.send_error_json( + HTTPStatus.TOO_MANY_REQUESTS, + "Too many failed attempts. Try again later.", + ) password = str(body.get("password", "")).strip() if not password: return self.send_error_json(HTTPStatus.BAD_REQUEST, "password is required") try: payload = self.app.auth_manager.submit_password(account_id, password) except Exception as exc: + _record_auth_failure(ip, account_id) return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + _record_auth_success(ip, account_id) return self.send_json(payload) def _clean_imported_channels(self, channels: Any) -> Dict[str, Any]: @@ -2594,11 +2877,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): try: enabled = bool(body.get("enabled")) interval_minutes = int(body.get("interval_minutes", 1)) - channels = [ - str(item).strip() - for item in body.get("channels", []) - if str(item).strip() - ] + channels, dropped = clean_continuous_channels(body.get("channels", [])) run_all_tracked = bool(body.get("run_all_tracked", True)) payload = self.app.continuous_orchestrator.update_for( account_id=account_id, @@ -2609,11 +2888,14 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): ) except Exception as exc: return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + payload["dropped_invalid"] = dropped return self.send_json(payload) # ── DELETE ─────────────────────────────────────────────────────────── def do_DELETE(self) -> None: + if not self._check_same_origin(): + return parsed = urllib.parse.urlparse(self.path) path = parsed.path @@ -2664,16 +2946,49 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): # ── Helpers ───────────────────────────────────────────────────────── - def read_json_body(self) -> Optional[Dict[str, Any]]: - length = int(self.headers.get("Content-Length", "0")) + def read_json_body(self) -> Any: + try: + length = int(self.headers.get("Content-Length", "0")) + except (TypeError, ValueError): + # Unparseable Content-Length - treat as no body rather than 500. + length = 0 if length <= 0: return {} + if length > MAX_JSON_BODY_BYTES: + return _JSON_BODY_TOO_LARGE + content_type = self.headers.get("Content-Type", "") + media_type = content_type.split(";", 1)[0].strip().lower() + if media_type != "application/json": + return _JSON_CONTENT_TYPE_REJECTED raw = self.rfile.read(length) + if len(raw) > MAX_JSON_BODY_BYTES: + return _JSON_BODY_TOO_LARGE try: return json.loads(raw.decode("utf-8")) - except json.JSONDecodeError: + except (json.JSONDecodeError, UnicodeDecodeError): return None + def _check_same_origin(self) -> bool: + """Reject state-mutating requests that are clearly cross-origin. + + Returns True when the request may proceed. Requests without an + Origin or Sec-Fetch-Site header are allowed (curl, same-origin + browsers that omit the header). When present, the headers must not + indicate a cross-origin request. + """ + host = self.headers.get("Host", "") + origin = self.headers.get("Origin") + if origin: + origin_host = urllib.parse.urlparse(origin).netloc + if not origin_host or (host and origin_host != host): + self.send_error_json(HTTPStatus.FORBIDDEN, "Cross-origin request rejected") + return False + sec_fetch_site = self.headers.get("Sec-Fetch-Site", "").strip().lower() + if sec_fetch_site and sec_fetch_site not in {"same-origin", "same-site", "none"}: + self.send_error_json(HTTPStatus.FORBIDDEN, "Cross-origin request rejected") + return False + return True + def serve_static(self, relative_path: str, head_only: bool = False) -> None: file_path = (WEBUI_DIR / relative_path).resolve() try: