fix(server): harden auth, SSE, state, scraping

- Fix SSE streams not terminating on successful jobs (C-2)
- Anchor data/session paths to BASE_DIR instead of CWD (C-3)
- Guard TelegramAuthManager state with RLock (H-1)
- Replace millisecond job ids with uuid4 (H-2)
- Always redact api_id/api_hash on export, drop include_secrets (H-3)
- Enforce JSON content-type + same-origin on mutating requests (H-4)
- Rate-limit auth attempts and phone-code requests (H-5)
- Deep-copy StateStore.load() on all paths (H-6)
- Cap FloodWait retries in forward_message (H-7)
- De-duplicate forwarding handler registration (H-8)
- Validate continuous channels at ingest, join scrape thread on account
  removal, fix refresh_config status under lock, cap SSE streams and
  JSON body size (M-5, M-6, M-17)
- Add regression tests (33 passing) and REVIEW.md

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/<id>/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/<id>/state.json` отдаёт api_hash (plaintext), `/media/accounts/<id>/<ch>/*.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/`.

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:

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")
+    BASE_DIR = Path(__file__).resolve().parent
-    session_dir = Path("session")
+    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.")

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

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.DATA_DIR = base_dir / "data" / "accounts" / account_id
-            self.state_store = get_account_store(Path("data"), 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:
+            except Exception as e:
-                pass
+                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)
         )

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)
+    def test_clean_continuous_channels_drops_invalid_and_normalizes(self):
-        assert full["api_hash"] == "secret"
+        """Continuous config ingest must reject path-traversal entries."""
-        assert "api_hash_present" not in full
+        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 ──────────────────────────────────────────────────
 
 

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);

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,6 +645,7 @@ class TelegramAuthManager:
         return future.result()
 
     def _get_auth_data(self, account_id: str) -> Dict[str, Any]:
+        with self.lock:
             if account_id not in self.auth_data:
                 self.auth_data[account_id] = {
                     "phase": "idle",
@@ -492,6 +663,7 @@ class TelegramAuthManager:
             return self.auth_data[account_id]
 
     def _set_state(self, account_id: str, **updates: Any) -> None:
+        with self.lock:
             data = self._get_auth_data(account_id)
             data.update(updates)
             data["updated_at"] = utc_now_iso()
@@ -512,6 +684,7 @@ 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.")
+        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(
@@ -522,22 +695,28 @@ class TelegramAuthManager:
             client = self.clients[account_id]
         if not client.is_connected():
             await client.connect()
-        data = self._get_auth_data(account_id)
+        with self.lock:
-        if data.get("user_id") is None and await client.is_user_authorized():
+            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]:
+        with self.lock:
             data = self._get_auth_data(account_id)
-        acc_state = load_account(DATA_DIR, 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
+        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,12 +769,16 @@ class TelegramAuthManager:
             qr_url=qr_url,
             qr_image=self._make_qr_image(qr_url),
         )
+        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:
+        with self.lock:
             data = self._get_auth_data(account_id)
             qr_login = data.get("qr_login")
         if not qr_login:
@@ -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)
+        with self.lock:
             data = self._get_auth_data(account_id)
-        if not data.get("phone") or not data.get("phone_code_hash"):
+            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:
+        with self.lock:
-            client = self.clients[account_id]
+            client = self.clients.get(account_id)
+        if client is not None:
             try:
                 future = asyncio.run_coroutine_threadsafe(
                     self._disconnect_client(client), self.loop
@@ -712,7 +899,8 @@ class TelegramAuthManager:
                 future.result(timeout=5)
             except Exception:
                 pass
-            del self.clients[account_id]
+        with self.lock:
+            self.clients.pop(account_id, None)
             self.auth_data.pop(account_id, None)
 
     async def _disconnect_client(self, client: TelegramClient) -> None:
@@ -720,15 +908,18 @@ class TelegramAuthManager:
             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,6 +1041,9 @@ class PerAccountContinuousScrapeManager:
                 "run_all_tracked": bool(run_all_tracked),
             }
             self._save_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()
@@ -869,13 +1065,32 @@ class PerAccountContinuousScrapeManager:
         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()
+        if not already_set:
             with self.lock:
                 was_running = self.status["running"]
-            self.status["running"] = False
             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)
         with self.lock:
@@ -887,9 +1102,14 @@ class PerAccountContinuousScrapeManager:
         return [channel for channel in configured if channel in tracked]
 
     def _run_loop(self) -> None:
+        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)
@@ -943,7 +1163,9 @@ class PerAccountContinuousScrapeManager:
                 interrupted = self.stop_event.wait(timeout=sleep_seconds)
                 if interrupted:
                     break
-
+        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")
@@ -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. <img> 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"}
+            return self.send_json({
-            payload = {
                 "version": 1,
                 "account_id": account_id,
-                "state": export_account_state(load_account(DATA_DIR, account_id), include_secrets=include_secrets),
+                "state": export_account_state(load_account(DATA_DIR, account_id)),
-            }
+            })
-            return self.send_json(payload)
         if sub == ["channels"]:
             return self._handle_get_account_channels(account_id)
         if len(sub) >= 3 and sub[0] == "channels" and sub[-1] == "messages":
@@ -1978,6 +2220,9 @@ 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")
+        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")
@@ -1987,7 +2232,10 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
             last_payload = None
             deadline = time.time() + 60 * 30
             while time.time() < deadline:
-            job = self.app.job_runner.get_job(job_id)
+                if runner.is_stream_revoked(stream_id):
+                    # Capped: this stream is the oldest and must close.
+                    break
+                job = runner.get_job(job_id)
                 if not job:
                     break
                 payload = json.dumps(job, ensure_ascii=False)
@@ -1998,9 +2246,11 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
                     except (BrokenPipeError, ConnectionResetError, OSError):
                         break
                     last_payload = payload
-            if job.get("status") in {"completed", "failed"}:
+                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 = [
+                channels, dropped = clean_continuous_channels(body.get("channels", []))
-                    str(item).strip()
-                    for item in body.get("channels", [])
-                    if str(item).strip()
-                ]
                 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 = [
+            channels, dropped = clean_continuous_channels(body.get("channels", []))
-                str(item).strip()
-                for item in body.get("channels", [])
-                if str(item).strip()
-            ]
             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]]:
+    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: