"""OAuth2 device flow + token lifecycle.
Flow for a headless server (the Django case)::
auth = client.start_device_login()
# operator opens auth.verification_url, enters auth.user_code
session = client.finish_device_login(auth, timeout=300)
The library never touches a browser: the host displays the user code
(terminal, admin screen, e-mail) and polls. Polling honours the
``slow_down`` instruction (interval grows) and the overall deadline.
Saved sessions stay bound to the OAuth *client* that issued them:
Tidal only honours a Bearer token for its own client, so switching
clients silently invalidates the login. :class:`OAuth` enforces that
binding at load time instead of failing on every request with opaque
401s.
"""
from __future__ import annotations
import base64
import json
import logging
import time
from dataclasses import dataclass
from typing import Callable
from . import transport as transport_mod
from .config import Config
from .errors import AuthError, CancelledError, DownloadError, TidalError
from .session import TokenRecord, TokenStore
logger = logging.getLogger("tidalpy.oauth")
_DEVICE_GRANT = "urn:ietf:params:oauth:grant-type:device_code"
_REFRESH_GRANT = "refresh_token"
_SCOPE = "r_usr w_usr w_sub"
[docs]
@dataclass
class DeviceAuth:
"""A pending device-flow login; show ``verification_url`` to a human."""
user_code: str
verification_uri: str
device_code: str
interval: int
expires_in: int
@property
def verification_url(self) -> str:
uri = self.verification_uri.rstrip("/")
return f"{uri}/{self.user_code}"
[docs]
class OAuth:
"""Token lifecycle: load, verify, refresh, device flow, re-login."""
[docs]
def __init__(
self,
config: Config,
transport: transport_mod.Transport,
store: TokenStore,
clock: Callable[[], float] | None = None,
sleeper: Callable[[float], None] | None = None,
) -> None:
self._config = config
self._transport = transport
self._store = store
self._clock = clock or time.monotonic
self._sleep = sleeper or time.sleep
client = config.oauth_client
self._client = client
self._user_agent = config.user_agent_for(client)
self._record = self._load_bound_record()
transport.bind_auth(self)
# -- AuthProvider (used by Transport) ------------------------------------
[docs]
def access_token(self) -> str | None:
return self._record.access_token if self._record else None
[docs]
def user_agent(self) -> str:
return self._user_agent
[docs]
def country_code(self) -> str | None:
return self._record.country_code if self._record else None
[docs]
def handle_401(self) -> bool:
"""One-shot refresh; True when a new access token is in place."""
if not (self._record and self._record.refresh_token):
return False
return self.refresh()
[docs]
def logout(self) -> bool:
"""End the session: best-effort remote revocation, then forget
the saved session locally.
Returns whether the remote revocation succeeded (``False`` when
there was no token to revoke or the request failed — the local
session is cleared either way; a token Tidal no longer accepts
is already as good as revoked).
"""
record = self._record
revoked = (
self._transport.revoke_session(record.access_token)
if record is not None and record.access_token
else False
)
# A concurrent login() may have replaced the session while the
# (remote) revocation was in flight: only forget the record we
# captured, never a newer one.
if self._record is record:
self.clear_saved_session()
return revoked
[docs]
def clear_saved_session(self) -> None:
self._record = None
self._store.clear()
# -- record management ----------------------------------------------------
def _load_bound_record(self) -> TokenRecord | None:
record = self._store.load()
if record is None:
return None
if not record.matches_client(self._client.client_id):
logger.info("saved login belongs to a different OAuth client; clearing")
self._store.clear()
return None
return record
def _commit(self, record: TokenRecord) -> None:
record.updated_at = time.time()
self._store.save(record)
self._record = record
@property
def record(self) -> TokenRecord | None:
return self._record
[docs]
def status(self) -> dict:
"""``auth_status()`` payload for hosts."""
record = self._record
if record is None or not record.present:
return {"logged_in": False, "client": self._client.platform}
seconds_left = None
if record.expires_after:
seconds_left = max(0, int(record.expires_after - time.time()))
return {
"logged_in": True,
"userId": record.user_id,
"countryCode": record.country_code,
"clientId": record.client_id,
# The platform persona this client is presenting as (the
# registry entry behind Config.client_id): a UI can show
# "logging in as Tidal HiRes" and flag a login that was
# minted for another client (see OAuthClient.notes).
"client": self._client.platform,
"tokenType": "web" if _is_web_token(record.access_token) else "device",
"expired": record.expired,
"secondsLeft": seconds_left,
}
# -- verification / refresh ------------------------------------------------
[docs]
def verify(self) -> bool:
"""Network check that the current access token works."""
if not self._record or not self._record.access_token:
return False
try:
body = self._transport.get_json("sessions")
except Exception:
return False
if not isinstance(body, dict):
return False
if body.get("status") not in (None, 200):
return False
user_id = body.get("userId")
if user_id is not None and self._record.user_id is not None:
if str(user_id) != str(self._record.user_id):
return False
return True
[docs]
def refresh(self) -> bool:
"""Refresh the access token using the stored refresh token."""
if not (self._record and self._record.refresh_token):
return False
data = {
"client_id": self._client.client_id,
"client_secret": self._client.client_secret,
"scope": _SCOPE,
"refresh_token": self._record.refresh_token,
"grant_type": _REFRESH_GRANT,
}
try:
result = self._transport.post_json("/token", data)
except (DownloadError, CancelledError):
return False
if not isinstance(result, dict) or result.get("status") not in (None, 200):
return False
user = result.get("user") or {}
new_record = TokenRecord(
user_id=user.get("userId") or result.get("userId") or self._record.user_id,
country_code=(
user.get("countryCode")
or result.get("countryCode")
or self._record.country_code
),
access_token=result.get("access_token"),
refresh_token=result.get("refresh_token") or self._record.refresh_token,
expires_after=time.time() + int(result.get("expires_in") or 0),
client_id=self._client.client_id,
)
if not new_record.access_token:
return False
self._commit(new_record)
return True
[docs]
def ensure_logged_in(self) -> TokenRecord:
"""A usable session, or :class:`AuthError`.
Order: fresh token -> expired token + refresh -> no refresh
token -> :class:`AuthError`.
"""
if self._record and self._record.present and not self._record.expired:
return self._record
if self._record and self._record.refresh_token and self.refresh():
return self._record # type: ignore[return-value]
if self._record is not None:
self.clear_saved_session()
raise AuthError(
"no valid Tidal login available; run a device login "
"or supply an access token first"
)
# -- re-login paths -----------------------------------------------------------
[docs]
def login_with_token(
self, access_token: str, user_id: int | str | None = None
) -> TokenRecord:
"""Adopt a token the operator pasted in (browser session / other app)."""
if not access_token or not str(access_token).strip():
raise AuthError("access token is empty")
access_token = str(access_token).strip()
if access_token.isdigit():
# A purely numeric "token" is a pasted user id (or a
# partial paste), never a Tidal OAuth token — fail fast
# with an actionable message instead of burning a probe
# that would reject it opaquely.
raise AuthError(
"the access token is numeric — Tidal tokens are not "
"numbers; did you paste a user id instead?"
)
# Probe with the candidate token, then restore the previous record
# whether or not the probe succeeded.
previous = self._record
probe = TokenRecord(access_token=access_token, client_id=self._client.client_id)
self._record = probe
try:
try:
body = self._transport.get_json("sessions")
except TidalError as exc:
# Any API rejection (401, stale client, ...) means the
# candidate token is unusable.
raise AuthError(f"token rejected by Tidal: {exc}") from exc
finally:
self._record = previous
if not isinstance(body, dict) or body.get("status") not in (None, 200):
raise AuthError("token rejected by Tidal (sessions check failed)")
if user_id is not None and str(body.get("userId")) != str(user_id):
raise AuthError("token belongs to a different user than expected")
record = TokenRecord(
user_id=body.get("userId"),
country_code=body.get("countryCode"),
access_token=access_token,
client_id=self._client.client_id,
)
self._commit(record)
return record
# -- device flow ------------------------------------------------------------
[docs]
def start_device_login(self) -> DeviceAuth:
"""Begin the device flow; returns the code to show to a human."""
data = {"client_id": self._client.client_id, "scope": _SCOPE}
result = self._transport.post_json("/device_authorization", data)
if not isinstance(result, dict):
raise DownloadError("device authorization returned no JSON")
if result.get("status") not in (None, 200):
raise AuthError(
"device authorization failed; try another OAuth client: "
f"{result.get('userMessage') or result.get('error') or 'unknown'}"
)
try:
return DeviceAuth(
user_code=result["userCode"],
verification_uri=result["verificationUri"],
device_code=result["deviceCode"],
interval=int(result.get("interval") or 5),
expires_in=int(result.get("expiresIn") or 300),
)
except KeyError as exc:
raise DownloadError(f"device authorization response missing {exc}") from exc
[docs]
def finish_device_login(
self,
auth: DeviceAuth,
timeout: float | None = None,
user_id: int | str | None = None,
) -> TokenRecord:
"""Poll the token endpoint until the operator approves (or not)."""
deadline = self._clock() + (timeout if timeout is not None else auth.expires_in)
interval = max(1, auth.interval)
data = {
"client_id": self._client.client_id,
"client_secret": self._client.client_secret,
"scope": _SCOPE,
"device_code": auth.device_code,
"grant_type": _DEVICE_GRANT,
}
while True:
if self._clock() >= deadline:
raise AuthError("device login timed out before approval")
# The injected sleeper is expected to be interruptible; a plain
# time.sleep here would make cancellation wait a full interval.
self._sleep(interval)
try:
result = self._transport.post_json("/token", data)
except DownloadError:
continue # transient auth-endpoint failure; keep polling
if not isinstance(result, dict):
continue
status = result.get("status")
if status is not None and status != 200:
sub_status = result.get("sub_status") or result.get("subStatus")
error = result.get("error")
if status == 400 and (
sub_status == 1002
or error in ("authorization_pending", "slow_down")
):
if error == "slow_down":
interval += 5
continue
raise AuthError(
result.get("userMessage")
or result.get("error_description")
or error
or "device login was denied"
)
return self._apply_token_result(result, user_id)
def _apply_token_result(
self, result: dict, user_id: int | str | None
) -> TokenRecord:
user = result.get("user") or {}
access_token = result.get("access_token")
if not access_token:
raise AuthError("device login returned no access token")
record = TokenRecord(
user_id=user.get("userId") or result.get("userId"),
country_code=user.get("countryCode") or result.get("countryCode"),
access_token=access_token,
refresh_token=result.get("refresh_token"),
expires_after=time.time() + int(result.get("expires_in") or 0),
client_id=self._client.client_id,
)
if user_id is not None and str(record.user_id) != str(user_id):
raise AuthError(
"device login was approved by a different user than expected"
)
self._commit(record)
return record
# ---------------------------------------------------------------------------
# Token introspection
# ---------------------------------------------------------------------------
[docs]
def decode_jwt_payload(access_token: str | None) -> dict:
"""Best-effort JWT payload decode (no signature check)."""
if not access_token or "." not in access_token:
return {}
try:
payload = access_token.split(".", 2)[1]
padding = "=" * (-len(payload) % 4)
raw = base64.urlsafe_b64decode(payload + padding)
data = json.loads(raw)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _is_web_token(access_token: str | None) -> bool:
"""True for tidal.com browser session tokens (INTERNAL OAuth)."""
claims = decode_jwt_payload(access_token)
if not claims:
return False
return claims.get("at") == "INTERNAT" or claims.get("type") == "o2_access"