connected_accounts/service.py
Rama main · réplica de lectura del repositorio original.
"""Trusted desktop account lifecycle, shared by UI, watches and integrations.
No public method returns tokens. Provider operations are the only path from
an account to its credential, and the host decides who can invoke them.
Locking: ``_lock`` guards the roster, flows and providers and is never held
across a network call, so ``accounts()`` (which the EventSub receive thread
and the Settings page read) cannot stall behind a slow Twitch request. A
per-account token lock serialises refresh and validation instead, because
Twitch public-client refresh tokens are single use. Order: token lock first,
then ``_lock``.
"""
from __future__ import annotations
import copy
import json
import secrets
import threading
import time
from pathlib import Path
from connected_accounts.models import AccountError, Provider, TokenStore
from connected_accounts.store import SecureTokenStore, atomic_write, set_aside
_ROSTER_FIELDS = {"id", "provider", "user_id", "client_id", "login", "display_name", "scopes", "status"}
_INVALID_GRANT = {"invalid_grant", "invalid_token", "revoked", "access_denied"}
_PENDING = {"authorization_pending", "slow_down"}
def _scopes(value) -> list[str]:
if isinstance(value, str):
value = value.split()
return sorted({s for s in value or [] if isinstance(s, str) and len(s) <= 128})
class ConnectedAccounts:
def __init__(self, data_dir: Path, providers: dict[str, Provider] | None = None,
*, store: TokenStore | None = None, clock=time.time):
self._path = Path(data_dir) / "connected-accounts.json"
self._store = store if store is not None else SecureTokenStore(data_dir)
self._providers = dict(providers or {})
self._clock = clock
self._lock = threading.RLock()
self._token_locks: dict[str, threading.Lock] = {}
self._flows: dict[str, dict] = {}
self._validated: dict[str, float] = {}
self._stop = threading.Event()
self._worker = None
self._accounts: dict[str, dict] = {}
# Set when an unreadable roster could not be moved aside: saving would
# overwrite it, so the list stays read-only until the next start.
self._roster_locked = False
for row in self._read_roster():
if isinstance(row, dict) and isinstance(row.get("id"), str):
clean = {k: v for k, v in row.items() if k in _ROSTER_FIELDS}
clean["scopes"] = _scopes(clean.get("scopes"))
self._accounts[clean["id"]] = clean
def _read_roster(self) -> list:
try:
saved = json.loads(self._path.read_text(encoding="utf-8"))
if isinstance(saved, list):
return saved
except FileNotFoundError:
return []
except (OSError, ValueError):
# Unreadable or damaged: handled below, like a non-list.
pass
# Connected Accounts is optional. A damaged list must not stop the
# engine, and every unrelated feature with it, from starting.
if set_aside(self._path):
print("The saved Twitch connection list could not be read and was set aside "
"(connected-accounts.json.corrupt). Connect Twitch again in Settings.")
else:
self._roster_locked = True
print("The saved Twitch connection list could not be read. "
"Twitch stays disconnected until it can be read again.")
return []
@property
def providers(self) -> dict[str, Provider]:
with self._lock:
return dict(self._providers)
def register_provider(self, provider: Provider) -> None:
with self._lock:
previous = self._providers.get(provider.id)
if previous and previous.client_id != provider.client_id:
previous_accounts = self._accounts
self._accounts = copy.deepcopy(previous_accounts)
for account in self._accounts.values():
if account["provider"] == provider.id:
account["status"] = "reconnect_required"
try:
self._persist()
except AccountError:
self._accounts = previous_accounts
raise
# Only commit registry/session changes after the account list
# is durable. An unsuccessful configuration change must leave
# the previous provider, pending flows and accounts usable.
self._flows = {k: f for k, f in self._flows.items() if f["provider"] != provider.id}
for account in self._accounts.values():
if account["provider"] == provider.id:
self._validated.pop(account["id"], None)
self._providers[provider.id] = provider
if previous is not None and previous is not provider:
close = getattr(previous, "close", None)
if close:
try:
self._call(close)
except AccountError:
# Best effort: the replacement is already registered.
pass
def credential_store_status(self) -> dict:
return {"available": bool(self._store.available), "backend": self._store.backend}
def status(self) -> dict:
with self._lock:
return {"credential_store": self.credential_store_status(), "accounts": self.accounts(),
"providers": [{"id": p.id, "configured": bool(p.client_id),
"capabilities": copy.deepcopy(getattr(p, "capabilities", {}))}
for p in self._providers.values()]}
def accounts(self, provider_id: str | None = None) -> list[dict]:
with self._lock:
return [copy.deepcopy(a) for a in self._accounts.values()
if provider_id is None or a.get("provider") == provider_id]
def _persist(self):
with self._lock:
if self._roster_locked:
raise AccountError("account_store_unavailable", "The account list could not be saved.", status=503)
try:
atomic_write(self._path, json.dumps(list(self._accounts.values())).encode("utf-8"))
except OSError:
raise AccountError("account_store_unavailable", "The account list could not be saved.",
status=503) from None
def _provider(self, provider_id: str) -> Provider:
provider = self._providers.get(provider_id)
if provider is None or not provider.client_id:
raise AccountError("provider_not_configured", "Configure this account provider before connecting.", status=409)
return provider
def _token_lock(self, account_id: str) -> threading.Lock:
with self._lock:
return self._token_locks.setdefault(account_id, threading.Lock())
def _call(self, function, *args, **kwargs):
try:
return function(*args, **kwargs)
except AccountError:
raise
except Exception:
# Network libraries commonly put URLs, query tokens and response
# bodies in their exceptions. Those must not reach the renderer.
raise AccountError("provider_unavailable", "The account provider could not complete this request.",
status=502) from None
def _revoke(self, provider: Provider, access_token: str, client_id: str | None) -> None:
"""Revoke with the client ID that issued the token, not just ours."""
if client_id and client_id != provider.client_id:
self._call(provider.revoke, access_token, client_id=client_id)
else:
self._call(provider.revoke, access_token)
def _revocable(self, provider: Provider, account: dict, token: dict) -> str:
"""The access token to revoke on disconnect.
Twitch refuses to revoke an expired access token, while its refresh
token keeps the grant alive (an app closed for a few hours is enough).
Refresh once with the issuing client, then revoke the fresh token.
"""
if (token.get("expires_at", 0) > self._clock() + 60 or not token.get("refresh_token")
or account.get("client_id") != provider.client_id):
return token["access_token"]
try:
return self._token(self._call(provider.refresh, token["refresh_token"]), token)["access_token"]
except AccountError:
# Revoke what there is; a refusal is reported as revoked: false.
return token["access_token"]
def begin(self, provider_id: str, scopes: list[str] | None = None) -> dict:
with self._lock:
if not self._store.available:
raise AccountError("credential_store_unavailable",
"An unlocked OS credential store is required to connect accounts.", status=503)
provider = self._provider(provider_id)
requested = _scopes(provider.default_scopes if scopes is None else scopes)
# Drop expired flows and bound retained device credentials.
now = self._clock()
self._flows = {k: f for k, f in self._flows.items() if f["expires_at"] > now}
if len(self._flows) >= 8:
raise AccountError("too_many_requests", "Finish or cancel an existing sign-in first.", status=429)
grant = self._call(provider.begin, requested)
if not grant.get("device_code") or not grant.get("verification_uri") or not grant.get("user_code"):
raise AccountError("invalid_provider_response", "The provider did not return a usable sign-in request.", status=502)
interval = max(1, min(float(grant.get("interval", 5)), 60))
expires_in = max(1, min(float(grant.get("expires_in", 600)), 1800))
flow_id = secrets.token_urlsafe(32)
with self._lock:
if self._providers.get(provider_id) is not provider:
raise AccountError("invalid_flow", "The sign-in request no longer exists. Start again.", status=404)
if len(self._flows) >= 8:
raise AccountError("too_many_requests", "Finish or cancel an existing sign-in first.", status=429)
self._flows[flow_id] = {"provider": provider_id, "client_id": provider.client_id,
"device_code": grant["device_code"], "scopes": requested,
"expires_at": now + expires_in, "interval": interval,
"next_poll": now + interval}
result = {"flow_id": flow_id, "status": "pending", "user_code": grant["user_code"],
"verification_uri": grant["verification_uri"], "expires_in": int(expires_in),
"interval": interval}
if grant.get("verification_uri_complete"):
result["verification_uri_complete"] = grant["verification_uri_complete"]
return result
def poll(self, flow_id: str) -> dict:
with self._lock:
flow = self._flows.get(flow_id)
if not flow:
raise AccountError("invalid_flow", "The sign-in request no longer exists. Start again.", status=404)
now = self._clock()
if flow["expires_at"] <= now:
self._flows.pop(flow_id, None)
raise AccountError("expired_token", "The sign-in request expired. Start again.", status=410)
if now < flow["next_poll"]:
return {"status": "pending", "interval": flow["interval"],
"expires_in": max(0, int(flow["expires_at"] - now))}
provider = self._provider(flow["provider"])
# Set before the request: a second poll meanwhile answers pending.
flow["next_poll"] = now + flow["interval"]
device_code = flow["device_code"]
try:
token = self._call(provider.poll, device_code)
except AccountError as error:
with self._lock:
if error.code in _PENDING:
if error.code == "slow_down":
flow["interval"] += 5
flow["next_poll"] = now + flow["interval"]
return {"status": "pending", "interval": flow["interval"],
"expires_in": max(0, int(flow["expires_at"] - now))}
if error.code in {"expired_token", "access_denied", "invalid_grant"}:
self._flows.pop(flow_id, None)
raise
with self._lock:
# A successful exchange is single use even if persistence fails.
current = self._flows.pop(flow_id, None) is flow and self._providers.get(flow["provider"]) is provider
try:
if not current:
# Cancelled, or the Client ID changed, while Twitch answered.
raise AccountError("invalid_flow", "The sign-in request no longer exists. Start again.", status=404)
account = self._finish(provider, token)
except Exception:
if isinstance(token, dict) and token.get("access_token"):
try:
self._call(provider.revoke, token["access_token"])
except AccountError:
# Best effort: the grant is unusable here either way.
pass
raise
return {"status": "connected", "account": account}
def cancel(self, flow_id: str) -> dict:
with self._lock:
self._flows.pop(flow_id, None)
return {"status": "cancelled"}
def _token(self, token: dict, previous: dict | None = None) -> dict:
if not isinstance(token, dict) or not isinstance(token.get("access_token"), str) or not token["access_token"]:
raise AccountError("invalid_provider_response", "The provider did not return a usable authorization.", status=502)
previous = previous or {}
return {"access_token": token["access_token"],
"refresh_token": token.get("refresh_token") or previous.get("refresh_token"),
"expires_at": self._clock() + max(0, float(token.get("expires_in", 0))),
"scope": _scopes(token.get("scope", token.get("scopes", previous.get("scope", []))))}
def _validate(self, provider: Provider, token: dict, *, user_id: str | None = None) -> dict:
verified = self._call(provider.validate, token["access_token"])
if (not isinstance(verified, dict) or verified.get("client_id") != provider.client_id
or not verified.get("user_id") or (user_id is not None and verified.get("user_id") != user_id)):
raise AccountError("invalid_token", "This authorization does not match the connected account.", status=401)
return verified
def _finish(self, provider: Provider, raw: dict) -> dict:
token = self._token(raw)
verified = self._validate(provider, token)
identity = self._call(provider.identity, token["access_token"])
if not isinstance(identity, dict) or identity.get("id") != verified["user_id"]:
raise AccountError("invalid_token", "This authorization does not match the connected account.", status=401)
account_id = provider.id + ":" + verified["user_id"]
account = {"id": account_id, "provider": provider.id, "user_id": verified["user_id"],
"client_id": provider.client_id, "login": str(identity.get("login", ""))[:128],
"display_name": str(identity.get("display_name", ""))[:256],
"scopes": _scopes(verified.get("scopes")), "status": "connected"}
token["scope"] = account["scopes"]
if "expires_in" in verified:
token["expires_at"] = self._clock() + max(0, float(verified["expires_in"]))
with self._token_lock(account_id), self._lock:
previous = copy.deepcopy(self._accounts.get(account_id))
try:
previous_token = self._store.load(account_id) if previous is not None else None
except AccountError:
# An unreadable old credential is no reason to throw away the
# one the person just approved; there is nothing to restore.
previous_token = None
self._store.save(account_id, token)
self._accounts[account_id] = account
try:
self._persist()
except AccountError:
# A newly granted token is revoked by poll after this failure.
# Revalidate restored credentials before using them: provider
# revocation semantics can invalidate related authorizations.
self._validated.pop(account_id, None)
if previous is not None:
self._accounts[account_id] = previous
else:
self._accounts.pop(account_id, None)
try:
if previous_token is not None:
self._store.save(account_id, previous_token)
else:
self._store.delete(account_id)
except AccountError:
# Never leave a roster entry claiming that a credential which
# could not be restored is usable. Best-effort removal also
# prevents keeping the newly granted token that poll revokes.
if previous is not None:
self._accounts[account_id]["status"] = "reconnect_required"
self._validated.pop(account_id, None)
try:
self._store.delete(account_id)
except AccountError:
# Already failing; the status above stays truthful.
pass
raise AccountError("credential_store_unavailable",
"The account could not be saved or restored. Reconnect this account.",
status=503) from None
raise
self._validated[account_id] = self._clock()
replaced = (previous_token or {}).get("access_token")
old_client = (previous or {}).get("client_id")
if replaced and replaced != token["access_token"] and old_client and old_client != provider.client_id:
# The grant under the previous Client ID stays live on Twitch
# unless revoked with that ID. The same app's grant is left alone:
# revoking it could take the new authorization with it.
try:
self._revoke(provider, replaced, old_client)
except AccountError:
# Best effort: the person can still remove it on Twitch.
pass
return copy.deepcopy(account)
def _account(self, account_id: str) -> dict:
account = self._accounts.get(account_id)
if account is None:
raise AccountError("account_not_found", "This account is no longer connected.", status=404)
return account
def _reconnect(self, account_id: str):
with self._lock:
account = self._accounts.get(account_id)
self._validated.pop(account_id, None)
if account is not None and account.get("status") != "reconnect_required":
account["status"] = "reconnect_required"
self._persist()
def _access(self, account: dict, *, force_validate: bool = False) -> tuple[Provider, str]:
"""Return a usable token for this snapshot. Hold its token lock."""
account_id = account["id"]
with self._lock:
provider = self._provider(account["provider"])
if account.get("client_id") != provider.client_id:
self._reconnect(account_id)
raise AccountError("reconnect_required", "Reconnect this account to continue.", status=401)
if account.get("status") == "reconnect_required":
raise AccountError("reconnect_required", "Reconnect this account to continue.", status=401)
token = self._store.load(account_id)
if not token or not token.get("access_token"):
self._reconnect(account_id)
raise AccountError("reconnect_required", "Reconnect this account to continue.", status=401)
refreshed = False
if token.get("expires_at", 0) <= self._clock() + 60:
if not token.get("refresh_token"):
self._reconnect(account_id)
raise AccountError("reconnect_required", "Reconnect this account to continue.", status=401)
try:
token = self._token(self._call(provider.refresh, token["refresh_token"]), token)
# Rotating refresh tokens must be committed before any other
# request; the old token may already have become unusable.
self._store.save(account_id, token)
refreshed = True
except AccountError as error:
if error.code in _INVALID_GRANT:
self._reconnect(account_id)
raise
with self._lock:
last = self._validated.get(account_id)
if force_validate or refreshed or last is None or self._clock() - last >= 3600:
try:
verified = self._validate(provider, token, user_id=account["user_id"])
except AccountError as error:
if error.code in _INVALID_GRANT:
self._reconnect(account_id)
raise
scopes = _scopes(verified.get("scopes"))
with self._lock:
current = self._accounts.get(account_id)
if current is not None and scopes != current.get("scopes"):
current["scopes"] = scopes
self._persist()
self._validated[account_id] = self._clock()
return provider, token["access_token"]
def _authorise(self, account_id: str, *, force_validate: bool = False) -> tuple[Provider, str, dict]:
with self._token_lock(account_id):
with self._lock:
account = copy.deepcopy(self._account(account_id))
provider, access_token = self._access(account, force_validate=force_validate)
with self._lock:
return provider, access_token, copy.deepcopy(self._account(account_id))
def test(self, account_id: str) -> dict:
_, _, account = self._authorise(account_id, force_validate=True)
return {"ok": True, "account": account}
def perform(self, account_id: str, operation: str, **params):
provider, access_token, account = self._authorise(account_id)
scope_map = getattr(provider, "operation_scopes", {})
required = set(scope_map.get(operation, ()))
if not required.issubset(set(account.get("scopes", []))):
raise AccountError("missing_scope", "Reconnect this account with permission for this action.", status=403)
# No lock across the provider's network request.
try:
return self._call(provider.perform, access_token, account, operation, **params)
except AccountError as error:
if error.code in _INVALID_GRANT:
self._reconnect(account_id)
raise
def disconnect(self, account_id: str) -> dict:
with self._token_lock(account_id):
with self._lock:
account = copy.deepcopy(self._account(account_id))
provider = self._providers.get(account["provider"])
revoked = False
try:
if provider:
close_account = getattr(provider, "close_account", None)
if close_account:
self._call(close_account, copy.deepcopy(account))
token = self._store.load(account_id)
if token and token.get("access_token"):
self._revoke(provider, self._revocable(provider, account, token),
account.get("client_id"))
revoked = True
except AccountError:
# An offline provider must not prevent local disconnection.
pass
with self._lock:
try:
self._store.delete(account_id)
except AccountError:
# Revocation may already have succeeded; keep any retry from
# reusing those credentials even if local deletion failed.
current = self._accounts.get(account_id)
if current is not None:
current["status"] = "reconnect_required"
self._validated.pop(account_id, None)
try:
self._persist()
except AccountError:
# The delete failure below is the error that matters.
pass
raise
current = self._accounts.pop(account_id, None) or account
self._validated.pop(account_id, None)
try:
self._persist()
except AccountError:
# The persisted roster may still list this account although
# its credentials are gone. Keep a truthful retryable entry
# in memory until the account list can be saved again.
self._accounts[account_id] = {**current, "status": "reconnect_required"}
raise
return {"disconnected": True, "revoked": revoked}
def start(self) -> None:
"""Validate restored sessions now and at least hourly while running."""
with self._lock:
if self._worker and self._worker.is_alive():
return
self._stop.clear()
self._worker = threading.Thread(target=self._maintain, name="connected-accounts", daemon=True)
self._worker.start()
def _maintain(self):
while not self._stop.is_set():
for account in self.accounts():
if account.get("status") == "connected":
try:
self._authorise(account["id"])
except AccountError:
# _access already marked a lost grant; anything else
# (offline, rate limited) is retried next round.
pass
self._stop.wait(60)
def close(self) -> None:
self._stop.set()
if self._worker:
self._worker.join(timeout=2)
with self._lock:
self._flows.clear()
providers = list(self._providers.values())
for provider in providers:
close = getattr(provider, "close", None)
if close:
self._call(close)