fix: protect Telegram session files from crash-loop overwrites (v7.1.1)

Back up session files before TelegramClient touches them and restore
if auth fails with a smaller file. Add non-interactive auth script for
headless environments.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: GeiserX

docs/CHANGELOG.md

@@ -6,6 +6,17 @@ For upgrade instructions, see [Upgrading](#upgrading) at the bottom.
 
 ## [Unreleased]
 
+## [7.1.1] - 2026-03-05
+
+### Added
+
+- **Non-interactive auth script** — `scripts/auth_noninteractive.py` for authenticating Telegram sessions without a TTY (useful for SSH automation, CI pipelines)
+
+### Fixed
+
+- **Session file protection** — Telethon session files are now backed up before each connect attempt. If the container crash-loops (e.g. due to database permission errors), the authenticated session is preserved and restored instead of being overwritten with an empty one
+- **Duplicate session_path assignment** in config.py removed
+
 ## [7.1.0] - 2026-03-05
 
 ### Added

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "telegram-archive"
-version = "7.1.0"
+version = "7.1.1"
 description = "Automated Telegram backup with Docker. Performs incremental backups of messages and media on a configurable schedule."
 readme = "README.md"
 requires-python = ">=3.14"

scripts/auth_noninteractive.py

@@ -0,0 +1,99 @@
+"""
+Non-interactive Telegram authentication helper.
+
+Use this script when you cannot run the interactive setup_auth
+(e.g., from SSH without TTY, CI pipelines, or automation scripts).
+
+Usage:
+    # Step 1: Send verification code to Telegram app
+    docker compose run --rm backup python scripts/auth_noninteractive.py send
+
+    # Step 2: Enter the code (and 2FA password if enabled)
+    docker compose run --rm backup python scripts/auth_noninteractive.py verify CODE
+    docker compose run --rm backup python scripts/auth_noninteractive.py verify CODE 2FA_PASSWORD
+
+Environment variables required (same as normal operation):
+    TELEGRAM_API_ID, TELEGRAM_API_HASH, TELEGRAM_PHONE, SESSION_NAME, BACKUP_PATH
+"""
+
+import asyncio
+import os
+import sys
+
+from telethon import TelegramClient
+
+
+def _get_session_path() -> str:
+    """Derive session path from environment (same logic as Config)."""
+    backup_path = os.getenv("BACKUP_PATH", "/data/backups")
+    session_dir = os.getenv("SESSION_DIR", os.path.join(os.path.dirname(backup_path.rstrip("/\\")), "session"))
+    session_name = os.getenv("SESSION_NAME", "telegram_backup")
+    os.makedirs(session_dir, exist_ok=True)
+    return os.path.join(session_dir, session_name)
+
+
+async def main():
+    if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help", "help"):
+        print(__doc__)
+        sys.exit(0)
+
+    action = sys.argv[1]
+
+    api_id = int(os.environ["TELEGRAM_API_ID"])
+    api_hash = os.environ["TELEGRAM_API_HASH"]
+    phone = os.environ["TELEGRAM_PHONE"]
+    session_path = _get_session_path()
+
+    client = TelegramClient(session_path, api_id, api_hash)
+    await client.connect()
+
+    if await client.is_user_authorized():
+        me = await client.get_me()
+        print(f"Already authorized as {me.first_name} (@{me.username})")
+        await client.disconnect()
+        return
+
+    if action == "send":
+        result = await client.send_code_request(phone)
+        print(f"Verification code sent to Telegram app ({result.type.__class__.__name__})")
+        print(f"Phone code hash: {result.phone_code_hash}")
+        await client.disconnect()
+
+    elif action == "verify":
+        if len(sys.argv) < 3:
+            print("Usage: python scripts/auth_noninteractive.py verify CODE [2FA_PASSWORD]")
+            sys.exit(1)
+
+        code = sys.argv[2]
+        password = sys.argv[3] if len(sys.argv) > 3 else None
+
+        # Must send_code_request first to establish the flow
+        await client.send_code_request(phone)
+
+        try:
+            await client.sign_in(phone, code)
+        except Exception as e:
+            error_str = str(e)
+            if "Two-steps verification" in error_str or "password" in error_str.lower() or "SessionPasswordNeeded" in error_str:
+                if not password:
+                    print("2FA is enabled. Re-run with: verify CODE 2FA_PASSWORD")
+                    await client.disconnect()
+                    sys.exit(1)
+                await client.sign_in(password=password)
+            else:
+                print(f"Authentication failed: {e}")
+                await client.disconnect()
+                sys.exit(1)
+
+        me = await client.get_me()
+        print(f"Authenticated as {me.first_name} (@{me.username})")
+        await client.disconnect()
+
+    else:
+        print(f"Unknown action: {action}")
+        print("Usage: send | verify CODE [2FA_PASSWORD]")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/config.py

@@ -118,9 +118,7 @@ def __init__(self):
         # If BACKUP_PATH is /data/backups, session goes to /data/session
         backup_parent = os.path.dirname(self.backup_path.rstrip("/\\"))
         self.session_dir = os.getenv("SESSION_DIR", os.path.join(backup_parent, "session"))
-        self.session_path = os.path.join(self.session_dir, f"{self.session_name}.session")
-
-        self.session_path = os.path.join(self.session_dir, f"{self.session_name}.session")
+        self.session_path = os.path.join(self.session_dir, self.session_name)
 
         # Database path configuration
         # Default: inside backup_path

src/connection.py

@@ -13,6 +13,8 @@
 
 import asyncio
 import logging
+import os
+import shutil
 
 from telethon import TelegramClient
 
@@ -86,6 +88,16 @@ async def connect(self) -> TelegramClient:
 
         logger.info("Connecting to Telegram...")
 
+        session_file = self.config.session_path + ".session"
+        backup_file = self.config.session_path + ".session.bak"
+
+        # Protect existing authenticated session: back it up before TelegramClient
+        # touches it. Telethon overwrites sessions on connect, so if the container
+        # crash-loops (e.g. DB permission errors) the authenticated session would be
+        # replaced with an empty one. The backup lets us recover.
+        if os.path.isfile(session_file) and os.path.getsize(session_file) > 0:
+            shutil.copy2(session_file, backup_file)
+
         self._client = TelegramClient(self.config.session_path, self.config.api_id, self.config.api_hash)
 
         # Enable WAL mode for session DB to handle concurrent access
@@ -96,10 +108,16 @@ async def connect(self) -> TelegramClient:
 
         # Check authorization
         if not await self._client.is_user_authorized():
+            await self._client.disconnect()
+            # Restore the backup if the live file was overwritten with an empty session
+            if os.path.isfile(backup_file) and os.path.getsize(backup_file) > os.path.getsize(session_file):
+                logger.warning("Session lost during connect — restoring from backup")
+                shutil.copy2(backup_file, session_file)
             logger.error("❌ Session not authorized!")
             logger.error("Please run the authentication setup first:")
             logger.error("  Docker: ./init_auth.bat (Windows) or ./init_auth.sh (Linux/Mac)")
             logger.error("  Local:  python -m src.setup_auth")
+            logger.error("  Non-interactive: python scripts/auth_noninteractive.py send")
             raise RuntimeError("Session not authorized. Please run authentication setup.")
 
         self._me = await self._client.get_me()