| 1 | import json |
| 2 | import os |
| 3 | import sys |
| 4 | from pathlib import Path |
| 5 | from typing import Dict |
| 6 | |
| 7 | from utils.cookie_utils import sanitize_cookies |
| 8 | from utils.logger import setup_logger |
| 9 | |
| 10 | logger = setup_logger("CookieManager") |
| 11 | |
| 12 | |
| 13 | class CookieManager: |
| 14 | def __init__(self, cookie_file: str = ".cookies.json"): |
| 15 | self.cookie_file = Path(cookie_file) |
| 16 | self.cookies: Dict[str, str] = {} |
| 17 | |
| 18 | def set_cookies(self, cookies: Dict[str, str]): |
| 19 | self.cookies = sanitize_cookies(cookies) |
| 20 | self._save_cookies() |
| 21 | |
| 22 | def get_cookies(self) -> Dict[str, str]: |
| 23 | if not self.cookies: |
| 24 | self._load_cookies() |
| 25 | return self.cookies |
| 26 | |
| 27 | def get_cookie_string(self) -> str: |
| 28 | cookies = self.get_cookies() |
| 29 | return "; ".join([f"{k}={v}" for k, v in cookies.items()]) |
| 30 | |
| 31 | def _save_cookies(self): |
| 32 | try: |
| 33 | # The cookie file lives alongside the config.yml in the |
| 34 | # per-user app-data dir. The directory is normally created by |
| 35 | # Electron (for config.yml) well before login, but create it |
| 36 | # defensively so a first-run login can't lose cookies to a |
| 37 | # missing parent dir. |
| 38 | self.cookie_file.parent.mkdir(parents=True, exist_ok=True) |
| 39 | with open(self.cookie_file, "w", encoding="utf-8") as f: |
| 40 | json.dump(self.cookies, f, ensure_ascii=False, indent=2) |
| 41 | # Restrict perms to owner-only on POSIX. Windows uses ACL-based |
| 42 | # isolation so chmod is a no-op there. |
| 43 | if sys.platform != "win32": |
| 44 | try: |
| 45 | os.chmod(self.cookie_file, 0o600) |
| 46 | except OSError as exc: |
| 47 | logger.warning("Could not chmod cookie file: %s", exc) |
| 48 | except Exception as e: |
| 49 | logger.error("Failed to save cookies to %s: %s", self.cookie_file, e) |
| 50 | |
| 51 | def _load_cookies(self): |
| 52 | if not self.cookie_file.exists(): |
| 53 | return |
| 54 | |
| 55 | try: |
| 56 | with open(self.cookie_file, "r", encoding="utf-8") as f: |
| 57 | self.cookies = sanitize_cookies(json.load(f)) |
| 58 | except Exception as e: |
| 59 | logger.error("Failed to load cookies: %s", e) |
| 60 | |
| 61 | def validate_cookies(self) -> bool: |
| 62 | required_keys = {"ttwid", "odin_tt", "passport_csrf_token"} |
| 63 | cookies = self.get_cookies() |
| 64 | missing = [key for key in required_keys if key not in cookies or not cookies.get(key)] |
| 65 | if missing: |
| 66 | logger.warning("Cookie validation failed, missing: %s", ", ".join(missing)) |
| 67 | return False |
| 68 | if not cookies.get("msToken"): |
| 69 | logger.info("msToken not found, it will be generated automatically if needed") |
| 70 | return True |
| 71 | |
| 72 | def clear_cookies(self): |
| 73 | self.cookies = {} |
| 74 | if self.cookie_file.exists(): |
| 75 | self.cookie_file.unlink() |
| 76 |