tidal_player_api/auth/
authenticator.rs

1//! The per-process [`Authenticator`] (spec 0002, AC4–AC9).
2
3use std::fmt;
4use std::sync::Arc;
5use std::time::{Duration, SystemTime};
6
7use reqwest::{Method, StatusCode, Url};
8use serde::de::DeserializeOwned;
9use tokio::sync::{Mutex, watch};
10
11use crate::{ApiResponse, SUB_STATUS_NOT_AVAILABLE};
12
13use super::{
14    AuthConfig, AuthError, Clock, LOSSY_WARNING, LOST_RECHECK, RefreshClient, RefreshFailure,
15    Session, SessionStore, TokenGrant, classify_refresh_failure, error_code, form_body,
16    needs_refresh, post_refresh, rejects_client, session_from_refresh,
17};
18
19/// Whether the session is usable, for the player to broadcast
20/// `Event::LoginRequired` / `Event::LoginRestored` on changes.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum AuthStatus {
23    /// Requests are sent (refreshing as needed).
24    Active,
25    /// The refresh token was rejected: requests fail with
26    /// [`AuthError::LoginRequired`] until a new login is found in storage.
27    LoginRequired,
28}
29
30/// In-memory state, behind one async mutex so refreshes are single flight.
31struct State {
32    session: Session,
33    /// `Some(last storage re-read)` while the session is lost.
34    lost_since_check: Option<SystemTime>,
35    /// The client of the last successful refresh (AC19).
36    refresh_client: Option<RefreshClient>,
37}
38
39/// Owns the in-memory session and sends authenticated requests: bearer
40/// header, proactive and reactive refresh (single flight, persisted to the
41/// store), cross-process pickup, and the "session lost" state.
42pub struct Authenticator {
43    http: reqwest::Client,
44    config: AuthConfig,
45    store: Arc<dyn SessionStore>,
46    clock: Arc<dyn Clock>,
47    state: Mutex<State>,
48    status: watch::Sender<AuthStatus>,
49}
50
51impl fmt::Debug for Authenticator {
52    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
53        f.debug_struct("Authenticator")
54            .field("config", &self.config)
55            .field("status", &*self.status.borrow())
56            .finish_non_exhaustive()
57    }
58}
59
60impl Authenticator {
61    /// An authenticator for `session`, which `store` holds.
62    pub fn new(
63        config: AuthConfig,
64        store: Arc<dyn SessionStore>,
65        session: Session,
66        clock: Arc<dyn Clock>,
67    ) -> Result<Self, AuthError> {
68        let http = reqwest::Client::builder()
69            .build()
70            .map_err(|e| AuthError::transport(&e))?;
71        Ok(Self {
72            http,
73            config,
74            store,
75            clock,
76            state: Mutex::new(State {
77                session,
78                lost_since_check: None,
79                refresh_client: None,
80            }),
81            status: watch::Sender::new(AuthStatus::Active),
82        })
83    }
84
85    /// An authenticator for the session in `store`, or `None` when not
86    /// logged in.
87    pub fn from_store(
88        config: AuthConfig,
89        store: Arc<dyn SessionStore>,
90        clock: Arc<dyn Clock>,
91    ) -> Result<Option<Self>, AuthError> {
92        match store.load()? {
93            Some(session) => Self::new(config, store, session, clock).map(Some),
94            None => Ok(None),
95        }
96    }
97
98    /// Follows [`AuthStatus`] changes.
99    pub fn status(&self) -> watch::Receiver<AuthStatus> {
100        self.status.subscribe()
101    }
102
103    /// The account's user ID and country code.
104    pub async fn account(&self) -> (u64, String) {
105        let state = self.state.lock().await;
106        (state.session.user_id, state.session.country_code.clone())
107    }
108
109    /// The client of this process's last successful refresh, `None` before
110    /// one (AC19).
111    pub async fn refresh_client(&self) -> Option<RefreshClient> {
112        self.state.lock().await.refresh_client
113    }
114
115    /// `GET {api_base}{path}` with the bearer token, decoded as JSON.
116    ///
117    /// As [`Self::get`], then any non-2xx status is an
118    /// [`AuthError::Http`].
119    pub async fn get_json<T: DeserializeOwned>(&self, path: &str) -> Result<T, AuthError> {
120        let response = self.get(path, &[], None).await?;
121        if !response.status().is_success() {
122            return Err(AuthError::Http {
123                status: response.status().as_u16(),
124                code: None,
125            });
126        }
127        response.json().map_err(|_| AuthError::Decode("API"))
128    }
129
130    /// `GET {api_base}{path}?{query}` with the bearer token; the response
131    /// is returned whatever its status, with its body read.
132    ///
133    /// Refreshes first when the token expires within 60 s, and once more on
134    /// a `401` before retrying once (AC5); a second `401` is
135    /// [`AuthError::Unauthorized`]. A `401` with `subStatus`
136    /// [`SUB_STATUS_NOT_AVAILABLE`] is not about the token: it is returned
137    /// as is, with no refresh and no retry (AC18). Fails fast with
138    /// [`AuthError::LoginRequired`] while the session is lost (AC7, AC9).
139    /// `timeout` bounds each HTTP request (none by default).
140    pub async fn get(
141        &self,
142        path: &str,
143        query: &[(&str, &str)],
144        timeout: Option<Duration>,
145    ) -> Result<ApiResponse, AuthError> {
146        self.send(Method::GET, path, query, None, &[], timeout)
147            .await
148    }
149
150    /// `{method} {api_base}{path}?{query}` with the bearer token, an optional
151    /// form body (`application/x-www-form-urlencoded`) and extra request
152    /// headers; the refresh and `401` rules of [`Self::get`] apply, and the
153    /// response is returned whatever its status, headers included.
154    pub async fn send(
155        &self,
156        method: Method,
157        path: &str,
158        query: &[(&str, &str)],
159        form: Option<&[(&str, &str)]>,
160        headers: &[(&str, &str)],
161        timeout: Option<Duration>,
162    ) -> Result<ApiResponse, AuthError> {
163        let url = self.api_url_with_query(path, query)?;
164        let request = Outgoing {
165            method,
166            url,
167            body: form.map(form_body),
168            headers,
169            timeout,
170        };
171        let token = self.valid_token().await?;
172        let mut response = self.send_once(&request, &token).await?;
173        if is_token_rejection(&response) {
174            let token = self.token_after_401(&token).await?;
175            response = self.send_once(&request, &token).await?;
176            if is_token_rejection(&response) {
177                return Err(AuthError::Unauthorized);
178            }
179        }
180        Ok(response)
181    }
182
183    fn api_url_with_query(&self, path: &str, query: &[(&str, &str)]) -> Result<Url, AuthError> {
184        let mut url = Url::parse(&self.config.api_url(path))
185            .map_err(|e| AuthError::Transport(format!("invalid API URL: {e}")))?;
186        if !query.is_empty() {
187            url.query_pairs_mut().extend_pairs(query);
188        }
189        Ok(url)
190    }
191
192    async fn send_once(
193        &self,
194        outgoing: &Outgoing<'_>,
195        token: &str,
196    ) -> Result<ApiResponse, AuthError> {
197        let mut request = self
198            .http
199            .request(outgoing.method.clone(), outgoing.url.clone())
200            .bearer_auth(token);
201        for (name, value) in outgoing.headers {
202            request = request.header(*name, *value);
203        }
204        if let Some(body) = &outgoing.body {
205            request = request
206                .header("content-type", "application/x-www-form-urlencoded")
207                .body(body.clone());
208        }
209        if let Some(timeout) = outgoing.timeout {
210            request = request.timeout(timeout);
211        }
212        let response = request.send().await.map_err(|e| AuthError::transport(&e))?;
213        let status = response.status();
214        let headers = response.headers().clone();
215        let body = response
216            .bytes()
217            .await
218            .map_err(|e| AuthError::transport(&e))?
219            .to_vec();
220        Ok(ApiResponse::new(status, headers, body))
221    }
222
223    /// The access token to send, refreshed first if it is about to expire.
224    async fn valid_token(&self) -> Result<String, AuthError> {
225        let mut state = self.state.lock().await;
226        self.leave_lost_state(&mut state)?;
227        if needs_refresh(&state.session, self.clock.now()) {
228            self.refresh(&mut state, Reason::Expiring).await?;
229        }
230        Ok(state.session.access_token.clone())
231    }
232
233    /// The access token to retry with after `rejected` got a `401`.
234    async fn token_after_401(&self, rejected: &str) -> Result<String, AuthError> {
235        let mut state = self.state.lock().await;
236        self.leave_lost_state(&mut state)?;
237        // Another request refreshed while this one was in flight.
238        if state.session.access_token == rejected {
239            self.refresh(&mut state, Reason::Rejected).await?;
240        }
241        Ok(state.session.access_token.clone())
242    }
243
244    /// In the lost state, re-reads storage at most every [`LOST_RECHECK`]
245    /// and leaves the state when it holds a different refresh token (AC9).
246    fn leave_lost_state(&self, state: &mut State) -> Result<(), AuthError> {
247        let Some(last_check) = state.lost_since_check else {
248            return Ok(());
249        };
250        let now = self.clock.now();
251        let due = now
252            .duration_since(last_check)
253            .is_ok_and(|elapsed| elapsed >= LOST_RECHECK);
254        if !due {
255            return Err(AuthError::LoginRequired);
256        }
257        state.lost_since_check = Some(now);
258        match self.load_stored() {
259            Some(stored) if stored.refresh_token != state.session.refresh_token => {
260                tracing::info!("found a new login in the session store");
261                state.session = stored;
262                state.lost_since_check = None;
263                self.set_status(AuthStatus::Active);
264                Ok(())
265            }
266            _ => Err(AuthError::LoginRequired),
267        }
268    }
269
270    /// Refreshes the session in place (holding the state lock, so concurrent
271    /// requests share it): adopts a newer stored session instead when there
272    /// is one (AC8a), persists the result (AC6), and on "session lost" tries
273    /// a different stored refresh token once (AC8b) before giving up (AC7).
274    async fn refresh(&self, state: &mut State, reason: Reason) -> Result<(), AuthError> {
275        if let Some(stored) = self.load_stored()
276            && stored.expires_at > state.session.expires_at
277        {
278            tracing::info!("adopting a newer session from the session store");
279            state.session = stored;
280            let usable = match reason {
281                Reason::Expiring => !needs_refresh(&state.session, self.clock.now()),
282                Reason::Rejected => true,
283            };
284            if usable {
285                return Ok(());
286            }
287        }
288        match self.refresh_once(state).await {
289            Err(Failure::Lost) => {}
290            other => return other.map_err(Failure::into_error),
291        }
292        if let Some(stored) = self.load_stored()
293            && stored.refresh_token != state.session.refresh_token
294        {
295            tracing::info!("refresh token rejected; retrying with the stored one");
296            state.session = stored;
297            match self.refresh_once(state).await {
298                Err(Failure::Lost) => {}
299                other => return other.map_err(Failure::into_error),
300            }
301        }
302        tracing::warn!("refresh token rejected: run \"tidal-player login\"");
303        state.lost_since_check = Some(self.clock.now());
304        self.set_status(AuthStatus::LoginRequired);
305        Err(AuthError::LoginRequired)
306    }
307
308    /// One `grant_type=refresh_token` call under the PKCE client, retried
309    /// once under the device-flow client if Tidal rejects the PKCE client
310    /// (AC19); on success the new session is saved, then installed.
311    async fn refresh_once(&self, state: &mut State) -> Result<(), Failure> {
312        let token = &state.session.refresh_token;
313        let mut client = RefreshClient::Pkce;
314        let mut reply = post_refresh(&self.http, &self.config, client, token)
315            .await
316            .map_err(Failure::Transient)?;
317        if rejects_client(reply.0, &reply.1) {
318            tracing::warn!("{LOSSY_WARNING}");
319            client = RefreshClient::DeviceFlow;
320            reply = post_refresh(&self.http, &self.config, client, token)
321                .await
322                .map_err(Failure::Transient)?;
323        }
324        let (status, body) = reply;
325        if !(200..300).contains(&status) {
326            return match classify_refresh_failure(status, &body) {
327                RefreshFailure::SessionLost => Err(Failure::Lost),
328                RefreshFailure::Transient => Err(Failure::Transient(AuthError::Http {
329                    status,
330                    code: error_code(&body).map(Into::into),
331                })),
332            };
333        }
334        let grant = TokenGrant::from_json(&body).map_err(Failure::Transient)?;
335        let session = session_from_refresh(&state.session, &grant, self.clock.now());
336        if let Err(error) = self.store.save(&session) {
337            tracing::warn!(%error, "could not save the refreshed session");
338        }
339        state.session = session;
340        state.refresh_client = Some(client);
341        Ok(())
342    }
343
344    fn load_stored(&self) -> Option<Session> {
345        self.store.load().unwrap_or_else(|error| {
346            tracing::warn!(%error, "could not read the session store");
347            None
348        })
349    }
350
351    fn set_status(&self, status: AuthStatus) {
352        self.status.send_if_modified(|current| {
353            let changed = *current != status;
354            *current = status;
355            changed
356        });
357    }
358}
359
360/// One request, kept whole so a `401` can resend it.
361struct Outgoing<'a> {
362    method: Method,
363    url: Url,
364    body: Option<String>,
365    headers: &'a [(&'a str, &'a str)],
366    timeout: Option<Duration>,
367}
368
369/// Whether `response` rejects the access token: a `401`, unless its
370/// `subStatus` is [`SUB_STATUS_NOT_AVAILABLE`].
371fn is_token_rejection(response: &ApiResponse) -> bool {
372    response.status() == StatusCode::UNAUTHORIZED
373        && response.sub_status() != Some(SUB_STATUS_NOT_AVAILABLE)
374}
375
376/// Why a refresh is attempted.
377#[derive(Clone, Copy)]
378enum Reason {
379    /// The token expires within the refresh margin.
380    Expiring,
381    /// The token got a `401`.
382    Rejected,
383}
384
385/// A failed refresh call.
386enum Failure {
387    /// The refresh token is rejected.
388    Lost,
389    /// Anything else; the session stays usable.
390    Transient(AuthError),
391}
392
393impl Failure {
394    fn into_error(self) -> AuthError {
395        match self {
396            Self::Lost => AuthError::LoginRequired,
397            Self::Transient(error) => error,
398        }
399    }
400}
tidal_player_api/auth/authenticator.rs

tidal_player_api/auth/
authenticator.rs
