From 11bfd2b3ee795c85fdde959e55111c5b0cc68ab7 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Fri, 10 Jul 2026 03:40:51 +0300 Subject: [PATCH 1/2] =?UTF-8?q?fix(cache):=20=D0=BA=D0=B0=D0=BD=D0=BE?= =?UTF-8?q?=D0=BD=D0=B8=D0=B7=D0=B0=D1=86=D0=B8=D1=8F=20=D0=BA=D0=BB=D1=8E?= =?UTF-8?q?=D1=87=D0=B0=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=D0=B0=20(Durov/du?= =?UTF-8?q?rov/@name=20->=20=D0=BE=D0=B4=D0=B8=D0=BD=20=D0=BA=D0=BB=D1=8E?= =?UTF-8?q?=D1=87)=20+=20one-shot=20=D0=BC=D0=B8=D0=B3=D1=80=D0=B0=D1=86?= =?UTF-8?q?=D0=B8=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Один канал жил под ключами Durov/durov/@name в трёх слоях (tgcache-файлы, data/cache/ каталоги, SQLite-строки) -> дубли кеша, двойные походы в Telegram, осиротевшие деревья. ID<->username НЕ унифицируем (сознательно). - channel_key.py: canonical_channel_key (strip @, -100... числовые verbatim, иначе lowercase; безопасно и для API-вызовов). - tg_cache._cache_file_path прогоняет ключ через canonical. - post_parser.get_channel_username -> lowercase (обе ветки; канонизирует SQLite/media-URL/t.me). Только эти 2 строки. - api_server.get_media: fs_channel = canonical_channel_key(channel) ПОСЛЕ проверки дайджеста (дайджест — по ИСХОДНОМУ url, иначе ломаются выданные ссылки); fs_channel во всех путях/ключах/download ниже. - migrate_channel_keys_sync в lifespan (после init_db_sync, до client.start, через asyncio.to_thread): канал целиком FS-потом-SQL. Обязательный os.path.samefile guard ДО merge/rmtree — на case-insensitive FS Durov/durov один inode, иначе снесли бы кеш. FS-сбой -> SQL пропуск (строки старорегистровые, дерево видно свиперу, нет вечных orphan). SQL-merge PK-safe (twin -> merge added=max/mime non-NULL + DELETE, иначе plain UPDATE). Идемпотентно. closes #24 Co-Authored-By: Claude Opus 4.8 (1M context) --- api_server.py | 36 ++++-- channel_key.py | 29 +++++ migrate_channel_keys.py | 176 +++++++++++++++++++++++++ post_parser.py | 4 +- tests/test_channel_key.py | 261 ++++++++++++++++++++++++++++++++++++++ tg_cache.py | 9 +- 6 files changed, 498 insertions(+), 17 deletions(-) create mode 100644 channel_key.py create mode 100644 migrate_channel_keys.py create mode 100644 tests/test_channel_key.py diff --git a/api_server.py b/api_server.py index 16d176e..bdf8cad 100644 --- a/api_server.py +++ b/api_server.py @@ -44,6 +44,8 @@ from file_io import (DB_PATH, init_db_sync, get_all_media_file_ids_sync, remove_media_file_ids_sync, get_mime_type_sync, set_mime_type_sync) from tg_cache import cleanup_legacy_cache_files, sweep_tgcache +from channel_key import canonical_channel_key +from migrate_channel_keys import migrate_channel_keys_sync # Global python-magic instance for MIME type detection magic_mime = magic.Magic(mime=True) @@ -270,6 +272,12 @@ async def lifespan(_: FastAPI): # Initialize SQLite database (creates table if not present) await asyncio.to_thread(init_db_sync, DB_PATH) + # One-shot migration of existing cache dirs + DB rows to the canonical channel key + # (case-insensitive username collapse). Runs AFTER init_db_sync and BEFORE client.start() + # and the background tasks. Wrapped in to_thread because it is a blocking FS-rename + + # SQLite routine that would otherwise stall the event loop. Idempotent: a re-run is a no-op. + await asyncio.to_thread(migrate_channel_keys_sync, DB_PATH, MEDIA_CACHE_DIR) + # One-shot startup maintenance of the history/chatinfo cache: drop legacy pickle files # (which are now always a miss), age-sweep stale tgcache entries, and remove the # pre-SQLite data/media_file_ids.json legacy dump (no longer read by any code). @@ -1311,30 +1319,32 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re #else: # logger.info(f"Valid digest for media {url}: {digest}") - # Convert numeric channel ID to int if needed - channel_id: Union[str, int] = channel - if isinstance(channel, str) and channel.startswith('-100'): - channel_id = int(channel) - + # Canonical filesystem/DB/API key. Computed AFTER the digest check (which must run + # against the ORIGINAL url string) so 'Durov', 'durov' and '@durov' collapse to one + # on-disk tree, one _access_updates/media_key identity and one download identity. The + # canonical form is API-safe: usernames are case-insensitive on Telegram's side and + # numeric '-100...' ids are preserved verbatim (download_media_file re-ints them). + fs_channel = canonical_channel_key(channel) + try: # Wrap the download and prepare call import time as _time # Pre-semaphore cache check: serve already-cached files without acquiring the semaphore - cache_path = media_cache_path(str(channel), post_id, file_unique_id) + cache_path = media_cache_path(fs_channel, post_id, file_unique_id) if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0: # File is already in cache — skip semaphore and serve directly - logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}") + logger.info(f"pre_semaphore_cache_hit: {fs_channel}/{post_id}/{file_unique_id}") # Record the access time into the accumulator instead of firing a per-hit - # SQLite write. Key channel as str(channel) — see _access_updates. - _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp() + # SQLite write. Key channel as the canonical fs_channel — see _access_updates. + _access_updates[(fs_channel, post_id, file_unique_id)] = datetime.now().timestamp() return await prepare_file_response(cache_path, request=request, - media_key=(str(channel), post_id, file_unique_id)) + media_key=(fs_channel, post_id, file_unique_id)) # A file that recently kept failing is in backoff: fast-reject instead of # occupying a scarce download slot (and a Pyrogram transmission permit) on a # request that will very likely hang again. Cached files already returned above, # so this only guards the live-download path. - backoff_remaining = _download_backoff_remaining((str(channel), post_id, file_unique_id)) + backoff_remaining = _download_backoff_remaining((fs_channel, post_id, file_unique_id)) if backoff_remaining > 0: logger.info(f"media_backoff_skip: {channel}/{post_id}/{file_unique_id} in backoff {backoff_remaining:.0f}s") return Response(status_code=503, content="Media temporarily unavailable, retry later", @@ -1369,7 +1379,7 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re if _sem_wait > 0.5: logger.warning(f"diag_semaphore_wait: {channel}/{post_id}/{file_unique_id} waited {_sem_wait:.3f}s for HTTP_DOWNLOAD_SEMAPHORE") _dl_start = _time.monotonic() - file_path, delete_after = await _download_deduped(channel_id, post_id, file_unique_id) + file_path, delete_after = await _download_deduped(fs_channel, post_id, file_unique_id) _dl_elapsed = _time.monotonic() - _dl_start logger.info(f"diag_download_timing: {channel}/{post_id}/{file_unique_id} download_media_file took {_dl_elapsed:.3f}s (semaphore_wait={_sem_wait:.3f}s)") finally: @@ -1378,7 +1388,7 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re raise HTTPException(status_code=404, detail="File not found") if file_path: return await prepare_file_response(file_path, request=request, delete_after=delete_after, - media_key=(str(channel), post_id, file_unique_id)) + media_key=(fs_channel, post_id, file_unique_id)) except ZeroSizeFileError as e: # Catch zero-size file errors logger.warning(f"zero_size_file_encountered: {str(e)}. Instructing client to retry.") return Response( diff --git a/channel_key.py b/channel_key.py new file mode 100644 index 0000000..976b56d --- /dev/null +++ b/channel_key.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +# flake8: noqa +# pylint: disable=missing-function-docstring + +"""Canonical channel key. + +Dependency-free helper shared by tg_cache, post_parser and api_server. Kept free of +any project imports so it can be imported by all three layers without introducing a +circular import. +""" + +from typing import Union + + +def canonical_channel_key(channel: Union[str, int]) -> str: + """Canonical cache/DB key for a channel. + + Telegram usernames are case-insensitive -> lowercase them. + Numeric '-100...' ids keep their exact string form. The '@' prefix is stripped. + The canonical form is also SAFE for Telegram API calls (usernames case-insensitive + on the API side; numeric unchanged) -- callers may thread one value through both + filesystem paths and API calls. + """ + s = str(channel).strip().lstrip('@') + if s.startswith('-100') and s[4:].isdigit(): + return s + return s.lower() diff --git a/migrate_channel_keys.py b/migrate_channel_keys.py new file mode 100644 index 0000000..def4aab --- /dev/null +++ b/migrate_channel_keys.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +# flake8: noqa +# pylint: disable=missing-function-docstring, broad-exception-caught, logging-fstring-interpolation + +"""One-shot migration of existing cache data to the canonical channel key. + +Collapses non-canonical (mixed-case) channel names in both the on-disk media cache +(data/cache//...) and the SQLite media_file_ids table onto their canonical, +lowercase form. Numeric '-100...' ids are already canonical and are never touched. + +Unit of work = a WHOLE channel, processed FS FIRST then SQL. The order is essential: +if the FS rename/merge fails, the channel's SQL rows are left old-cased so the old tree +is still reachable by the media sweeper via its DB rows (no eternal orphan directory). + +The migration is idempotent: a second run finds nothing to do and is a no-op. +""" + +import os +import shutil +import sqlite3 +import logging + +from channel_key import canonical_channel_key + +logger = logging.getLogger(__name__) + + +def _merge_dir_tree(src: str, dst: str) -> None: + """Merge every file under ``src`` into ``dst`` where an existing ``dst`` file WINS. + + Both trees share the same filesystem (siblings under the media cache root), so files + are moved with os.rename. Any file that already exists in ``dst`` is kept and the + ``src`` copy is discarded. Emptied ``src`` remnants are removed at the end. + """ + for root, _dirs, files in os.walk(src): + rel = os.path.relpath(root, src) + target_root = dst if rel == '.' else os.path.join(dst, rel) + os.makedirs(target_root, exist_ok=True) + for name in files: + s = os.path.join(root, name) + t = os.path.join(target_root, name) + if os.path.exists(t): + os.remove(s) # existing file in dst wins + else: + os.rename(s, t) + shutil.rmtree(src, ignore_errors=True) + + +def migrate_channel_keys_sync(db_path: str, cache_dir: str) -> None: + """Migrate mixed-case channel cache dirs and DB rows onto the canonical key. + + Blocking (FS renames + SQLite). Call via asyncio.to_thread from an async context. + """ + rows_merged = 0 + dirs_renamed = 0 + samefile_noops = 0 + failures = 0 + + conn = sqlite3.connect(db_path) + conn.execute("PRAGMA busy_timeout = 5000") + try: + # ------------------------------------------------------------------ # + # Candidate channels: mixed-case first-level cache dirs UNION mixed-case DB rows. + # ------------------------------------------------------------------ # + candidates: set[str] = set() + try: + for entry in os.listdir(cache_dir): + full = os.path.join(cache_dir, entry) + if not os.path.isdir(full): + continue + if entry.startswith('-100'): + continue + if entry != entry.lower(): + candidates.add(entry) + except FileNotFoundError: + pass # No cache dir yet — nothing to migrate on the FS side. + + try: + cur = conn.execute( + "SELECT DISTINCT channel FROM media_file_ids " + "WHERE channel != lower(channel) AND channel NOT LIKE '-100%'" + ) + for (ch,) in cur.fetchall(): + candidates.add(ch) + except sqlite3.Error as e: + logger.error(f"migrate_channel_keys: DB candidate query failed: {e}") + + for name in sorted(candidates): + canonical = canonical_channel_key(name) + if canonical == name: + # Nothing to change (already canonical) — defensive, candidates shouldn't hit this. + continue + try: + # ---------------------------- FS step ---------------------------- # + src = os.path.join(cache_dir, name) + dst = os.path.join(cache_dir, canonical) + if os.path.isdir(src): + try: + if os.path.exists(dst) and os.path.samefile(src, dst): + # Case-insensitive FS (macOS/APFS, Docker Desktop for Mac): + # src and dst are ONE directory. Renaming/merging would destroy + # the channel's entire cache — treat as a pure no-op. + samefile_noops += 1 + elif not os.path.exists(dst): + os.rename(src, dst) + dirs_renamed += 1 + else: + # Genuinely different dirs (case-sensitive FS, both forms present): + # per-file merge with the existing dst file winning. + _merge_dir_tree(src, dst) + dirs_renamed += 1 + except OSError as e: + # FS step failed (EACCES etc.): mark orphan candidate and SKIP the SQL + # step so the old-cased DB rows keep the old tree visible to the sweeper. + logger.error( + f"migrate_channel_keys: FS step failed for {name!r} -> {canonical!r} " + f"(orphan candidate, SQL skipped): {e}" + ) + failures += 1 + continue + + # ---------------------------- SQL step --------------------------- # + old_rows = conn.execute( + "SELECT post_id, file_unique_id, added, mime_type " + "FROM media_file_ids WHERE channel = ?", + (name,), + ).fetchall() + for (post_id, file_unique_id, added, mime_type) in old_rows: + twin = conn.execute( + "SELECT added, mime_type FROM media_file_ids " + "WHERE channel = ? AND post_id = ? AND file_unique_id = ?", + (canonical, post_id, file_unique_id), + ).fetchone() + if twin is None: + # No lowercase twin: a plain re-key is safe (no PK conflict). + conn.execute( + "UPDATE media_file_ids SET channel = ? " + "WHERE channel = ? AND post_id = ? AND file_unique_id = ?", + (canonical, name, post_id, file_unique_id), + ) + else: + # Twin exists: merge (added = max, prefer non-NULL mime_type) into the + # canonical row, then delete the old-cased row. A bare + # UPDATE ... SET channel=lower(channel) is FORBIDDEN (PK conflict). + t_added, t_mime = twin + new_added = max(added, t_added) + new_mime = t_mime if t_mime is not None else mime_type + conn.execute( + "UPDATE media_file_ids SET added = ?, mime_type = ? " + "WHERE channel = ? AND post_id = ? AND file_unique_id = ?", + (new_added, new_mime, canonical, post_id, file_unique_id), + ) + conn.execute( + "DELETE FROM media_file_ids " + "WHERE channel = ? AND post_id = ? AND file_unique_id = ?", + (name, post_id, file_unique_id), + ) + rows_merged += 1 + conn.commit() + except Exception as e: + # Never crash startup: log and move on. A re-run is an idempotent no-op. + try: + conn.rollback() + except Exception: + pass + logger.error(f"migrate_channel_keys: channel {name!r} migration failed: {e}") + failures += 1 + finally: + conn.close() + + logger.info( + f"migration_summary: rows merged {rows_merged}, dirs renamed {dirs_renamed}, " + f"samefile no-ops {samefile_noops}, failures {failures}" + ) diff --git a/post_parser.py b/post_parser.py index ad1e61e..91f92ce 100644 --- a/post_parser.py +++ b/post_parser.py @@ -1416,8 +1416,8 @@ class PostParser: if hasattr(chat, 'usernames') and chat.usernames: # Check many usernames active_usernames = [u.username for u in chat.usernames if u.active] if active_usernames: - return active_usernames[0] - if hasattr(chat, 'username') and chat.username: return chat.username # Check single username + return active_usernames[0].lower() # Canonicalize: usernames are case-insensitive + if hasattr(chat, 'username') and chat.username: return chat.username.lower() # Check single username (canonicalized) # Return numeric ID only if no username found if isinstance(chat.id, int) and str(chat.id).startswith('-100'): return str(chat.id) diff --git a/tests/test_channel_key.py b/tests/test_channel_key.py new file mode 100644 index 0000000..7ed272f --- /dev/null +++ b/tests/test_channel_key.py @@ -0,0 +1,261 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +"""Tests for the canonical channel key: helper, the three wiring layers, and the +one-shot migration (issue #24, tasks 8-11).""" + +import os +import sqlite3 +from types import SimpleNamespace + +import pytest + +import tg_cache +from channel_key import canonical_channel_key +from migrate_channel_keys import migrate_channel_keys_sync +from post_parser import PostParser + + +# --------------------------------------------------------------------------- # +# Task 11 — unit: canonical_channel_key. +# --------------------------------------------------------------------------- # +def test_canonical_lowercases_username(): + assert canonical_channel_key('Durov') == 'durov' + + +def test_canonical_strips_at_prefix(): + assert canonical_channel_key('@durov') == 'durov' + assert canonical_channel_key('@Durov') == 'durov' + + +def test_canonical_numeric_id_str_preserved(): + assert canonical_channel_key('-1001234567890') == '-1001234567890' + + +def test_canonical_numeric_id_int_preserved(): + assert canonical_channel_key(-1001234567890) == '-1001234567890' + + +def test_canonical_already_lowercase_idempotent(): + assert canonical_channel_key('durov') == 'durov' + assert canonical_channel_key(canonical_channel_key('Durov')) == 'durov' + + +# --------------------------------------------------------------------------- # +# Task 11 — tg_cache path is identical for the two casings. +# --------------------------------------------------------------------------- # +def test_tgcache_path_identical_for_casings(tmp_path, monkeypatch): + monkeypatch.setattr(tg_cache, "CACHE_DIR", str(tmp_path)) + p_upper = tg_cache._cache_file_path('Durov', 'history.json') + p_lower = tg_cache._cache_file_path('durov', 'history.json') + p_at = tg_cache._cache_file_path('@DUROV', 'history.json') + assert p_upper == p_lower == p_at + assert os.path.basename(p_upper) == 'durov.history.json' + + +def test_tgcache_path_numeric_preserved(tmp_path, monkeypatch): + monkeypatch.setattr(tg_cache, "CACHE_DIR", str(tmp_path)) + p = tg_cache._cache_file_path('-1001234567890', 'chatinfo.json') + assert os.path.basename(p) == '-1001234567890.chatinfo.json' + + +# --------------------------------------------------------------------------- # +# Task 11 — get_channel_username lowercases both branches. +# --------------------------------------------------------------------------- # +def _parser(): + return PostParser(SimpleNamespace()) + + +def test_get_channel_username_single_lowercased(): + parser = _parser() + msg = SimpleNamespace(chat=SimpleNamespace(username='MixedCase', id=1)) + assert parser.get_channel_username(msg) == 'mixedcase' + + +def test_get_channel_username_usernames_list_lowercased(): + parser = _parser() + chat = SimpleNamespace( + usernames=[SimpleNamespace(username='MixedCase', active=True)], + username=None, + id=1, + ) + assert parser.get_channel_username(SimpleNamespace(chat=chat)) == 'mixedcase' + + +def test_get_channel_username_numeric_id_unchanged(): + parser = _parser() + chat = SimpleNamespace(usernames=None, username=None, id=-1001234567890) + assert parser.get_channel_username(SimpleNamespace(chat=chat)) == '-1001234567890' + + +# --------------------------------------------------------------------------- # +# Migration helpers. +# --------------------------------------------------------------------------- # +def _make_db(path): + conn = sqlite3.connect(path) + conn.execute( + """CREATE TABLE media_file_ids ( + channel TEXT NOT NULL, + post_id INTEGER NOT NULL, + file_unique_id TEXT NOT NULL, + added REAL NOT NULL, + mime_type TEXT, + PRIMARY KEY (channel, post_id, file_unique_id) + )""" + ) + conn.commit() + conn.close() + + +def _rows(path): + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + rows = [dict(r) for r in conn.execute( + "SELECT channel, post_id, file_unique_id, added, mime_type FROM media_file_ids " + "ORDER BY channel, post_id, file_unique_id")] + conn.close() + return rows + + +# --------------------------------------------------------------------------- # +# Task 11 (a) — SQL merge of both forms → one row, max(added), non-NULL mime_type. +# --------------------------------------------------------------------------- # +def test_migration_sql_merge(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + # Old-cased row (higher added, NULL mime) and lowercase twin (lower added, has mime). + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 200.0, None)) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 100.0, 'image/jpeg')) + conn.commit() + conn.close() + + migrate_channel_keys_sync(db, str(cache)) + + rows = _rows(db) + assert len(rows) == 1 + r = rows[0] + assert r['channel'] == 'durov' + assert r['added'] == 200.0 # max(added) + assert r['mime_type'] == 'image/jpeg' # non-NULL preferred + + +def test_migration_sql_rekey_no_twin(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 7, 'fid', 50.0, 'video/mp4')) + conn.commit() + conn.close() + + migrate_channel_keys_sync(db, str(cache)) + + rows = _rows(db) + assert len(rows) == 1 + assert rows[0]['channel'] == 'durov' + assert rows[0]['mime_type'] == 'video/mp4' + + +# --------------------------------------------------------------------------- # +# Task 11 (b) — samefile guard → no-op, data intact (case-insensitive FS simulated). +# --------------------------------------------------------------------------- # +def test_migration_samefile_guard_no_data_loss(tmp_path, monkeypatch): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + # Single dir 'Durov' with a file; on a case-insensitive FS 'durov' would be the same dir. + src = cache / "Durov" / "5" + src.mkdir(parents=True) + (src / "fid").write_bytes(b"payload") + + # Simulate a case-insensitive FS: any existence check on the lowercase twin resolves, + # and samefile reports src == dst so the FS step must be a pure no-op. + real_exists = os.path.exists + + def fake_exists(p): + if os.path.basename(p) == 'durov': + return True + return real_exists(p) + + monkeypatch.setattr(os.path, "exists", fake_exists) + monkeypatch.setattr(os.path, "samefile", lambda a, b: True) + + migrate_channel_keys_sync(db, str(cache)) + + # Data intact: the original file must still be present and unchanged. + assert (cache / "Durov" / "5" / "fid").read_bytes() == b"payload" + + +# --------------------------------------------------------------------------- # +# Task 11 (c) — merge of two genuinely different dirs → combined, target wins. +# --------------------------------------------------------------------------- # +def test_migration_fs_merge_different_dirs(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + + # Old-cased tree. + (cache / "Durov" / "5").mkdir(parents=True) + (cache / "Durov" / "5" / "shared").write_bytes(b"OLD") # collides -> dst wins + (cache / "Durov" / "5" / "only_old").write_bytes(b"OLD_ONLY") + # Canonical tree. + (cache / "durov" / "5").mkdir(parents=True) + (cache / "durov" / "5" / "shared").write_bytes(b"NEW") # existing dst wins + (cache / "durov" / "6").mkdir(parents=True) + (cache / "durov" / "6" / "only_new").write_bytes(b"NEW_ONLY") + + migrate_channel_keys_sync(db, str(cache)) + + # Old dir gone, everything merged under 'durov'. + assert not (cache / "Durov").exists() + assert (cache / "durov" / "5" / "shared").read_bytes() == b"NEW" # target wins + assert (cache / "durov" / "5" / "only_old").read_bytes() == b"OLD_ONLY" # brought over + assert (cache / "durov" / "6" / "only_new").read_bytes() == b"NEW_ONLY" # untouched + + +def test_migration_fs_rename_when_dst_missing(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + (cache / "Durov" / "5").mkdir(parents=True) + (cache / "Durov" / "5" / "fid").write_bytes(b"data") + + migrate_channel_keys_sync(db, str(cache)) + + assert not (cache / "Durov").exists() + assert (cache / "durov" / "5" / "fid").read_bytes() == b"data" + + +# --------------------------------------------------------------------------- # +# Task 11 (d) — re-run is a no-op. +# --------------------------------------------------------------------------- # +def test_migration_rerun_noop(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 200.0, None)) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 100.0, 'image/jpeg')) + conn.commit() + conn.close() + (cache / "Durov" / "5").mkdir(parents=True) + (cache / "Durov" / "5" / "fid").write_bytes(b"data") + + migrate_channel_keys_sync(db, str(cache)) + first = _rows(db) + # Second run: nothing left to migrate. + migrate_channel_keys_sync(db, str(cache)) + second = _rows(db) + + assert first == second + assert all(r['channel'] == r['channel'].lower() for r in second) + assert (cache / "durov" / "5" / "fid").read_bytes() == b"data" diff --git a/tg_cache.py b/tg_cache.py index c61c535..26304bd 100644 --- a/tg_cache.py +++ b/tg_cache.py @@ -27,6 +27,7 @@ from message_snapshot import ( snapshot_messages, restore_messages, ) +from channel_key import canonical_channel_key logger = logging.getLogger(__name__) @@ -51,11 +52,15 @@ def _safe_key(key: Union[str, int]) -> str: def _cache_file_path(key: Union[str, int], suffix: str) -> str: - """Return the cache file path . (e.g. 'history.json' / 'chatinfo.json').""" + """Return the cache file path . (e.g. 'history.json' / 'chatinfo.json'). + + The key is first canonicalized so that 'Durov', 'durov' and '@durov' collapse to a + single cache file (Telegram usernames are case-insensitive). + """ if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR, exist_ok=True) logger.info(f"cache_dir_created: path {CACHE_DIR}") - return os.path.join(CACHE_DIR, f"{_safe_key(key)}.{suffix}") + return os.path.join(CACHE_DIR, f"{_safe_key(canonical_channel_key(key))}.{suffix}") # --------------------------------------------------------------------------- # From 76f79a1c5f1a268a2ffa6ecb58899458dc44c7fe Mon Sep 17 00:00:00 2001 From: agent_coder Date: Fri, 10 Jul 2026 03:56:54 +0300 Subject: [PATCH 2/2] =?UTF-8?q?fix(cache):=20traversal-=D0=B3=D0=B0=D1=80?= =?UTF-8?q?=D0=B4=20=D0=BD=D0=B0=20=D0=B4=D0=B5=D1=81=D1=82=D1=80=D1=83?= =?UTF-8?q?=D0=BA=D1=82=D0=B8=D0=B2=D0=BD=D1=8B=D0=B5=20FS-=D0=BE=D0=BF?= =?UTF-8?q?=D0=B5=D1=80=D0=B0=D1=86=D0=B8=D0=B8=20=D0=BC=D0=B8=D0=B3=D1=80?= =?UTF-8?q?=D0=B0=D1=86=D0=B8=D0=B8=20+=20=D1=82=D0=B5=D1=81=D1=82=D1=8B?= =?UTF-8?q?=20data-loss-guard/reverse-max=20(=D1=80=D0=B5=D0=B2=D1=8C?= =?UTF-8?q?=D1=8E=20#47)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _is_safe_channel_segment: отвергает имя канала с '/'/'\\'/'..' или не-basename; вызывается в начале цикла миграции ДО любого os.rename/_merge_dir_tree/rmtree, покрывает оба источника кандидатов (SELECT DISTINCT channel + листинг каталогов). Грязный канал (напр. из pre-existing route-traversal) -> skip+warning+failures++, старт не падает, деструктивная rename/rmtree не уходит за cache_dir. - Тесты (мутационно проверены): FS-fail -> SQL пропущен, строки old-cased, failures++ (снять continue -> тест краснеет); reverse max(added) (Durov=100/ durov=300 -> 300); traversal-гард скипает '../Evil'/'a/B' без FS-операций. Co-Authored-By: Claude Opus 4.8 (1M context) --- migrate_channel_keys.py | 31 ++++++++++++++ tests/test_channel_key.py | 89 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+) diff --git a/migrate_channel_keys.py b/migrate_channel_keys.py index def4aab..023e670 100644 --- a/migrate_channel_keys.py +++ b/migrate_channel_keys.py @@ -27,6 +27,26 @@ from channel_key import canonical_channel_key logger = logging.getLogger(__name__) +def _is_safe_channel_segment(name: str) -> bool: + """True iff ``name`` is a plain single path segment safe to join under cache_dir. + + Defends the destructive FS ops below (os.rename / _merge_dir_tree / shutil.rmtree) + against a dirty channel string that reached the DB or a crafted dir on disk, + regardless of any upstream route-traversal. A dirty name (containing '/', '\\' or a + '..' component, or that is not a bare basename, or empty/'.') would let a rename or + rmtree escape cache_dir, so it is rejected and the channel is left un-migrated. + """ + if not name or name == '.': + return False + if '/' in name or '\\' in name: + return False + if '..' in name.replace('\\', '/').split('/'): + return False + if os.path.basename(name) != name: + return False + return True + + def _merge_dir_tree(src: str, dst: str) -> None: """Merge every file under ``src`` into ``dst`` where an existing ``dst`` file WINS. @@ -88,6 +108,17 @@ def migrate_channel_keys_sync(db_path: str, cache_dir: str) -> None: logger.error(f"migrate_channel_keys: DB candidate query failed: {e}") for name in sorted(candidates): + # Traversal guard: reject any dirty channel name BEFORE it reaches the + # destructive FS ops (rename/merge/rmtree). Applies to both candidate sources + # (DB rows and the dir listing). A rejected channel is left as-is (its old-cased + # rows/dirs survive until the 20-day sweep, same as any un-migrated channel). + if not _is_safe_channel_segment(name): + logger.warning( + f"migrate_channel_keys: unsafe channel name {name!r} rejected " + f"(would escape cache_dir), skipping" + ) + failures += 1 + continue canonical = canonical_channel_key(name) if canonical == name: # Nothing to change (already canonical) — defensive, candidates shouldn't hit this. diff --git a/tests/test_channel_key.py b/tests/test_channel_key.py index 7ed272f..0cb48b1 100644 --- a/tests/test_channel_key.py +++ b/tests/test_channel_key.py @@ -5,7 +5,9 @@ """Tests for the canonical channel key: helper, the three wiring layers, and the one-shot migration (issue #24, tasks 8-11).""" +import logging import os +import shutil import sqlite3 from types import SimpleNamespace @@ -259,3 +261,90 @@ def test_migration_rerun_noop(tmp_path): assert first == second assert all(r['channel'] == r['channel'].lower() for r in second) assert (cache / "durov" / "5" / "fid").read_bytes() == b"data" + + +# --------------------------------------------------------------------------- # +# Data-loss guard — an FS-step failure must SKIP the SQL step (rows stay old-cased). +# Has teeth: fails if the `continue` after the FS OSError is removed. +# --------------------------------------------------------------------------- # +def test_migration_fs_failure_skips_sql(tmp_path, monkeypatch, caplog): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 100.0, 'image/jpeg')) + conn.commit() + conn.close() + # Old-cased dir with no lowercase twin → the migration takes the os.rename path. + (cache / "Durov" / "5").mkdir(parents=True) + (cache / "Durov" / "5" / "fid").write_bytes(b"data") + + def boom(*_a, **_k): + raise OSError("disk on fire") + + monkeypatch.setattr(os, "rename", boom) + + with caplog.at_level(logging.INFO): + migrate_channel_keys_sync(db, str(cache)) # (a) must NOT crash startup + + # (b) SQL step skipped: the row is still OLD-cased. + rows = _rows(db) + assert len(rows) == 1 + assert rows[0]['channel'] == 'Durov' + # (c) failures counter increased (surfaced in the summary log). + assert "failures 1" in caplog.text + + +# --------------------------------------------------------------------------- # +# Reverse-direction max(added): lowercase twin has the LARGER `added` → it wins. +# Proves max() is not one-directional ("old always wins"). +# --------------------------------------------------------------------------- # +def test_migration_sql_merge_reverse_max(tmp_path): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + # OLD-cased added=100 (SMALLER), lowercase twin added=300 (LARGER). + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 100.0, 'image/jpeg')) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 300.0, None)) + conn.commit() + conn.close() + + migrate_channel_keys_sync(db, str(cache)) + + rows = _rows(db) + assert len(rows) == 1 + assert rows[0]['channel'] == 'durov' + assert rows[0]['added'] == 300.0 # max(added): the twin's LARGER value survives + assert rows[0]['mime_type'] == 'image/jpeg' # non-NULL preferred (from the old row) + + +# --------------------------------------------------------------------------- # +# Traversal guard — a DB channel name with '/' or '..' is SKIPPED before any FS op. +# Has teeth: fails if the _is_safe_channel_segment guard is removed. +# --------------------------------------------------------------------------- # +def test_migration_traversal_guard_skips_dirty_channel(tmp_path, monkeypatch): + db = str(tmp_path / "m.db") + cache = tmp_path / "cache" + cache.mkdir() + _make_db(db) + conn = sqlite3.connect(db) + # Dirty mixed-case channel names → become migration candidates via the DB query. + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('../Evil', 5, 'fid', 100.0, None)) + conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('a/B', 6, 'fid', 100.0, None)) + conn.commit() + conn.close() + + # Spy: no destructive FS op may run (all dirty candidates rejected before the FS step). + called = [] + monkeypatch.setattr(os, "rename", lambda *a, **k: called.append(('rename', a))) + monkeypatch.setattr(shutil, "rmtree", lambda *a, **k: called.append(('rmtree', a))) + + migrate_channel_keys_sync(db, str(cache)) # must NOT crash + + assert called == [] # nothing renamed/removed → no op escaped cache_dir + # Rejected channels are left un-migrated (rows keep their original dirty names). + channels = {r['channel'] for r in _rows(db)} + assert channels == {'../Evil', 'a/B'}