"""Token storage: where a saved login lives.
A *session* is the (access token, refresh token, user, country,
issuing client) tuple Tidal's device flow yields. The default
:class:`FileTokenStore` is a single JSON file with mode ``0600``;
hosts (Django) may supply any object implementing :class:`TokenStore`
(e.g. DB-backed or encrypted-at-rest) — the rest of the library only
talks to the protocol.
Compatibility: files written by Tidal-Media-Downloader
(``~/.tidal-dl.token.json``) are base64-wrapped JSON with the same
field names; :class:`FileTokenStore` reads them transparently and
rewrites them in the plain-JSON format on first save.
"""
from __future__ import annotations
import base64
import json
import logging
import os
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Protocol, runtime_checkable
logger = logging.getLogger("tidalpy.session")
[docs]
@dataclass
class TokenRecord:
"""A saved login, bound to the OAuth client that issued it."""
user_id: int | str | None = None
country_code: str | None = None
access_token: str | None = None
refresh_token: str | None = None
expires_after: float = 0.0 # epoch seconds; 0 = unknown
client_id: str | None = None
updated_at: float = field(default_factory=time.time)
@property
def present(self) -> bool:
return bool(self.access_token)
@property
def expired(self) -> bool:
if not self.expires_after:
return False
return self.expires_after <= time.time()
[docs]
def matches_client(self, client_id: str | None) -> bool:
"""A Bearer token is only valid for its issuing client."""
if not self.client_id or not client_id:
return True # unbound (legacy) records are not rejected here
return self.client_id == client_id
[docs]
def to_dict(self) -> dict[str, Any]:
return {
"userid": str(self.user_id) if self.user_id is not None else None,
"countryCode": self.country_code,
"accessToken": self.access_token,
"refreshToken": self.refresh_token,
"expiresAfter": self.expires_after,
"clientId": self.client_id,
"updatedAt": self.updated_at,
}
[docs]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "TokenRecord":
user_id = data.get("userid", data.get("userId"))
if user_id in ("", None):
user_id = None
expires_after = data.get("expiresAfter") or 0
try:
expires_after = float(expires_after)
except (TypeError, ValueError):
expires_after = 0.0
return cls(
user_id=user_id,
country_code=data.get("countryCode"),
access_token=data.get("accessToken"),
refresh_token=data.get("refreshToken"),
expires_after=expires_after,
client_id=data.get("clientId"),
)
[docs]
@runtime_checkable
class TokenStore(Protocol):
"""Storage a :class:`~tidalpy.client.TidalClient` reads and writes."""
[docs]
def load(self) -> TokenRecord | None:
"""Current saved session, or ``None`` when absent/unreadable."""
[docs]
def save(self, record: TokenRecord) -> None:
"""Persist ``record`` (must be durable and atomic enough)."""
[docs]
def clear(self) -> None:
"""Forget the saved session."""
[docs]
class FileTokenStore:
"""JSON-file token store with atomic writes and legacy format support."""
[docs]
def __init__(self, path: str | Path | None = None) -> None:
self.path = Path(path) if path else Path.home() / ".tidalpy" / "token.json"
[docs]
def load(self) -> TokenRecord | None:
try:
raw = self.path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
if not isinstance(exc, FileNotFoundError):
logger.warning("token file %s unreadable: %s", self.path, exc)
return None
raw = raw.strip()
if not raw:
return None
data = self._parse(raw)
if data is None:
logger.warning("token file %s is not valid JSON; ignoring", self.path)
return None
record = TokenRecord.from_dict(data)
if not record.present:
return None
return record
@staticmethod
def _parse(raw: str) -> dict[str, Any] | None:
"""Plain JSON first; then the legacy base64-wrapped JSON."""
try:
data = json.loads(raw)
return data if isinstance(data, dict) else None
except ValueError:
pass
try:
decoded = base64.b64decode(raw, validate=False).decode("utf-8")
data = json.loads(decoded)
return data if isinstance(data, dict) else None
except (ValueError, UnicodeDecodeError):
return None
[docs]
def save(self, record: TokenRecord) -> None:
parent = self.path.parent
parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(
dir=str(parent), prefix=".token-", suffix=".tmp"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(record.to_dict(), handle, indent=2, sort_keys=True)
handle.write("\n")
try:
os.chmod(tmp_name, 0o600)
except OSError:
pass
os.replace(tmp_name, self.path)
except BaseException:
try:
os.unlink(tmp_name)
except OSError:
pass
raise
[docs]
def clear(self) -> None:
try:
self.path.unlink()
except FileNotFoundError:
pass
except OSError as exc:
logger.warning("could not clear token file %s: %s", self.path, exc)