brokapi_finam_protocol/
auth.rs

1//! JWT lifecycle: secret -> `AuthService.Auth` -> JWT, refreshed ahead of expiry.
2//!
3//! One [`TokenManager`] per secret. It is shared (`Arc`) by every connector that uses the secret,
4//! so several accounts or roles do not each run their own `Auth`. The refresher task holds only a
5//! `Weak` reference and ends when the last `Arc` is dropped.
6//!
7//! Lifetime is the shorter of `TokenDetails` (`expires_at - created_at`, both server clock, so no local
8//! clock skew) and the JWT claims `exp - iat`. If only one of them is known, that one. If neither, a
9//! conservative assumption. `Auth` and `TokenDetails` each have a deadline of [`AUTH_TIMEOUT`].
10//!
11//! # Examples
12//!
13//! Start a manager against the in-process imitation of the broker, read the token and the accounts
14//! that the secret can reach:
15//!
16//! ```
17//! use std::sync::Arc;
18//! use brokapi_finam_protocol::client::connect_lazy;
19//! use brokapi_finam_protocol::{ApiEndpoint, GrpcTokenApi, TokenManager};
20//! # use brokapi_mocks::finam::{MockServer, SECRET};
21//!
22//! # #[tokio::main(flavor = "current_thread")]
23//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
24//! # let server = MockServer::start().await;
25//! # let endpoint = ApiEndpoint::new(server.url.clone());
26//! # let secret = SECRET;
27//! let channel = connect_lazy(&endpoint)?;
28//! let api = Arc::new(GrpcTokenApi::new(channel));
29//! let tokens = TokenManager::start(api, secret, "my-app").await?;
30//!
31//! assert_eq!(tokens.generation(), 1);
32//! assert!(!tokens.current().is_empty());
33//! assert_eq!(tokens.account_ids(), vec!["ACC1", "ACC2"]);
34//!
35//! // Renew by hand, for example after you saw UNAUTHENTICATED.
36//! tokens.refresh().await?;
37//! assert_eq!(tokens.generation(), 2);
38//! # Ok(())
39//! # }
40//! ```
41
42use std::fmt;
43use std::sync::{Arc, RwLock, Weak};
44use std::time::Duration;
45
46use async_trait::async_trait;
47use base64::Engine;
48use base64::engine::general_purpose::URL_SAFE_NO_PAD;
49use tokio::sync::{Mutex, Notify};
50use tokio::time::Instant;
51use tonic::transport::Channel;
52use tracing::{debug, warn};
53
54use crate::error::FinamError;
55use crate::pb::grpc::tradeapi::v1::auth::auth_service_client::AuthServiceClient;
56use crate::pb::grpc::tradeapi::v1::auth::{AuthRequest, TokenDetailsRequest};
57
58/// Used only when neither `TokenDetails` nor the JWT claims tell the lifetime.
59pub const ASSUMED_TTL: Duration = Duration::from_secs(5 * 60);
60
61/// Deadline of the `Auth` call and of the `TokenDetails` call. Without it a connection that
62/// accepted the request and never answers would hold the refresh lock for ever.
63pub const AUTH_TIMEOUT: Duration = Duration::from_secs(10);
64
65/// Refresh retry backoff bounds after a failed refresh.
66const RETRY_MIN: Duration = Duration::from_secs(1);
67const RETRY_MAX: Duration = Duration::from_secs(30);
68
69/// What `TokenDetails` says about a token.
70#[derive(Debug, Clone, Default, PartialEq, Eq)]
71pub struct TokenInfo {
72    /// Token lifetime (`expires_at - created_at`), if known.
73    pub ttl: Option<Duration>,
74    /// Accounts reachable with this token.
75    pub account_ids: Vec<String>,
76    /// Trading is disabled for this session.
77    pub readonly: bool,
78}
79
80/// Auth transport, abstracted so refresh logic is testable without a network.
81///
82/// [`GrpcTokenApi`] is the real implementation. Implement the trait yourself to test code that depends
83/// on a [`TokenManager`] without a server.
84///
85/// # Examples
86///
87/// ```
88/// use std::sync::Arc;
89/// use async_trait::async_trait;
90/// use brokapi_finam_protocol::{FinamError, TokenApi, TokenInfo, TokenManager};
91///
92/// /// Gives numbered tokens and no details.
93/// struct Fake(std::sync::atomic::AtomicU32);
94///
95/// #[async_trait]
96/// impl TokenApi for Fake {
97///     async fn auth(&self, _secret: &str, _app: &str) -> Result<String, FinamError> {
98///         let n = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
99///         Ok(format!("token-{n}"))
100///     }
101///     async fn token_details(&self, _token: &str) -> Result<TokenInfo, FinamError> {
102///         Ok(TokenInfo { account_ids: vec!["A1".into()], ..TokenInfo::default() })
103///     }
104/// }
105///
106/// # #[tokio::main(flavor = "current_thread")]
107/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
108/// let tokens = TokenManager::start(Arc::new(Fake(Default::default())), "secret", "").await?;
109/// assert_eq!(&*tokens.current(), "token-1");
110/// assert_eq!(tokens.account_ids(), vec!["A1"]);
111/// # Ok(())
112/// # }
113/// ```
114#[async_trait]
115pub trait TokenApi: Send + Sync {
116    /// Exchanges the API secret for a JWT (`AuthService.Auth`).
117    ///
118    /// `source_app_id` is the application id the broker shows in its records. It may be empty.
119    ///
120    /// # Errors
121    ///
122    /// [`FinamError::Status`] when the server refuses the secret or cannot be reached.
123    async fn auth(&self, secret: &str, source_app_id: &str) -> Result<String, FinamError>;
124    /// Reads the lifetime and the account list of `token` (`AuthService.TokenDetails`).
125    ///
126    /// The manager does not fail when this call fails: it falls back to the JWT claims for the
127    /// lifetime and to an empty account list.
128    ///
129    /// # Errors
130    ///
131    /// [`FinamError::Status`] when the server refuses the token or cannot be reached.
132    async fn token_details(&self, token: &str) -> Result<TokenInfo, FinamError>;
133}
134
135/// [`TokenApi`] over the gRPC `AuthService` of the broker.
136///
137/// See the [module docs](self) for how it is used with a [`TokenManager`].
138pub struct GrpcTokenApi {
139    client: AuthServiceClient<Channel>,
140}
141
142impl GrpcTokenApi {
143    /// Auth transport over `channel`. The channel needs no authorization: `Auth` is the call that
144    /// creates the token.
145    pub fn new(channel: Channel) -> Self {
146        Self {
147            client: AuthServiceClient::new(channel),
148        }
149    }
150}
151
152#[async_trait]
153impl TokenApi for GrpcTokenApi {
154    async fn auth(&self, secret: &str, source_app_id: &str) -> Result<String, FinamError> {
155        let resp = self
156            .client
157            .clone()
158            .auth(AuthRequest {
159                secret: secret.to_string(),
160                source_app_id: source_app_id.to_string(),
161            })
162            .await?
163            .into_inner();
164        Ok(resp.token)
165    }
166
167    async fn token_details(&self, token: &str) -> Result<TokenInfo, FinamError> {
168        let d = self
169            .client
170            .clone()
171            .token_details(TokenDetailsRequest {
172                token: token.to_string(),
173            })
174            .await?
175            .into_inner();
176        let ttl = match (d.created_at, d.expires_at) {
177            (Some(c), Some(e)) => {
178                let secs = e.seconds - c.seconds;
179                (secs > 0).then(|| Duration::from_secs(secs as u64))
180            }
181            _ => None,
182        };
183        Ok(TokenInfo {
184            ttl,
185            account_ids: d.account_ids,
186            readonly: d.readonly,
187        })
188    }
189}
190
191/// Lifetime from the JWT payload (`exp` - `iat`), if the token is a readable JWT.
192///
193/// The signature is not checked and the token is not validated. `None` when the token is not a JWT,
194/// the claims are missing, or `exp` is not after `iat`.
195///
196/// # Examples
197///
198/// ```
199/// use std::time::Duration;
200/// use brokapi_finam_protocol::auth::jwt_ttl;
201///
202/// // Header {"alg":"none"}, payload {"iat":1000,"exp":1900}.
203/// let token = "eyJhbGciOiJub25lIn0.eyJpYXQiOjEwMDAsImV4cCI6MTkwMH0.sig";
204/// assert_eq!(jwt_ttl(token), Some(Duration::from_secs(900)));
205/// assert_eq!(jwt_ttl("opaque-token"), None);
206/// ```
207pub fn jwt_ttl(token: &str) -> Option<Duration> {
208    let payload = token.split('.').nth(1)?;
209    let bytes = URL_SAFE_NO_PAD.decode(payload.trim_end_matches('=')).ok()?;
210    let v: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
211    let exp = v.get("exp")?.as_i64()?;
212    let iat = v.get("iat")?.as_i64()?;
213    (exp > iat).then(|| Duration::from_secs((exp - iat) as u64))
214}
215
216/// How long after issuing to renew a token that lives `ttl`.
217///
218/// Renews `ttl/5` (clamped to 5s..120s) before expiry, but never later than half of the lifetime.
219///
220/// # Examples
221///
222/// ```
223/// use std::time::Duration;
224/// use brokapi_finam_protocol::auth::refresh_after;
225///
226/// // A 15 minute token is renewed 2 minutes before it ends.
227/// assert_eq!(refresh_after(Duration::from_secs(900)), Duration::from_secs(780));
228/// // A very short token is renewed at half of its life.
229/// assert_eq!(refresh_after(Duration::from_secs(8)), Duration::from_secs(4));
230/// ```
231pub fn refresh_after(ttl: Duration) -> Duration {
232    let lead = (ttl / 5).clamp(Duration::from_secs(5), Duration::from_secs(120));
233    if ttl <= lead * 2 { ttl / 2 } else { ttl - lead }
234}
235
236struct State {
237    token: Arc<str>,
238    info: TokenInfo,
239    refresh_at: Instant,
240    /// Bumped on every successful refresh.
241    generation: u64,
242}
243
244/// Holds the current JWT and keeps it fresh. Share via `Arc` between connectors.
245///
246/// A background task renews the token ahead of expiry (see [`refresh_after`] for the rule). If a renewal
247/// fails, the old token stays in use and the task retries with a growing pause (1 s up to 30 s).
248/// The task ends when the last `Arc` is dropped.
249///
250/// The `Debug` output hides the secret.
251///
252/// See the [module docs](self) for an example.
253pub struct TokenManager {
254    api: Arc<dyn TokenApi>,
255    secret: String,
256    source_app_id: String,
257    state: RwLock<State>,
258    wake: Arc<Notify>,
259    refresh_lock: Mutex<()>,
260}
261
262impl fmt::Debug for TokenManager {
263    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
264        f.debug_struct("TokenManager")
265            .field("secret", &"<redacted>")
266            .field("generation", &self.generation())
267            .finish()
268    }
269}
270
271impl TokenManager {
272    /// Authenticates once (failure here means a bad secret or no connectivity) and starts the
273    /// background refresher.
274    ///
275    /// Must be called inside a Tokio runtime, because it spawns the refresher task.
276    ///
277    /// # Errors
278    ///
279    /// - [`FinamError::Config`]: the secret is empty or only spaces. Nothing is sent.
280    /// - [`FinamError::Auth`]: the server answered with an empty token.
281    /// - [`FinamError::Status`]: the `Auth` call failed (a wrong secret is `UNAUTHENTICATED`).
282    ///
283    /// A failure of `TokenDetails` is not an error: the lifetime then comes from the JWT claims, or
284    /// from [`ASSUMED_TTL`].
285    pub async fn start(
286        api: Arc<dyn TokenApi>,
287        secret: impl Into<String>,
288        source_app_id: impl Into<String>,
289    ) -> Result<Arc<Self>, FinamError> {
290        let secret = secret.into();
291        if secret.trim().is_empty() {
292            return Err(FinamError::Config("empty secret".into()));
293        }
294        let source_app_id = source_app_id.into();
295        let fresh = fetch(api.as_ref(), &secret, &source_app_id).await?;
296        let manager = Arc::new(Self {
297            api,
298            secret,
299            source_app_id,
300            state: RwLock::new(fresh),
301            wake: Arc::new(Notify::new()),
302            refresh_lock: Mutex::new(()),
303        });
304        tokio::spawn(refresh_loop(Arc::downgrade(&manager)));
305        Ok(manager)
306    }
307
308    /// Current JWT (what goes into the `authorization` metadata).
309    ///
310    /// Cheap: no network. Call it for every request instead of keeping the value.
311    pub fn current(&self) -> Arc<str> {
312        Arc::clone(&self.state.read().expect("token state poisoned").token)
313    }
314
315    /// Details of the current token: lifetime, accounts, read-only flag. No network.
316    pub fn info(&self) -> TokenInfo {
317        self.state
318            .read()
319            .expect("token state poisoned")
320            .info
321            .clone()
322    }
323
324    /// Accounts available to this secret (`TokenDetails.account_ids`).
325    ///
326    /// Empty when `TokenDetails` failed or listed none.
327    pub fn account_ids(&self) -> Vec<String> {
328        self.info().account_ids
329    }
330
331    /// Number of successful issues so far (1 after [`TokenManager::start`]).
332    ///
333    /// Changes every time the token is replaced, so it tells a reader whether its copy is stale.
334    pub fn generation(&self) -> u64 {
335        self.state.read().expect("token state poisoned").generation
336    }
337
338    /// Asks the refresher to renew now, e.g. after `UNAUTHENTICATED`. Non-blocking.
339    ///
340    /// The renewal runs in the background. Use [`TokenManager::refresh`] to wait for it.
341    pub fn request_refresh(&self) {
342        self.wake.notify_one();
343    }
344
345    /// Renews the token immediately.
346    ///
347    /// If another task renewed the token while this one waited for the lock, nothing more is sent
348    /// and the call returns `Ok`. So parallel callers cause one `Auth`, not many.
349    ///
350    /// # Errors
351    ///
352    /// The same as [`TokenManager::start`] without the empty-secret check. The old token stays in
353    /// use when the renewal fails.
354    pub async fn refresh(&self) -> Result<(), FinamError> {
355        let seen = self.generation();
356        let _guard = self.refresh_lock.lock().await;
357        if self.generation() != seen {
358            // Someone else refreshed while we waited for the lock.
359            return Ok(());
360        }
361        let mut fresh = fetch(self.api.as_ref(), &self.secret, &self.source_app_id).await?;
362        let mut st = self.state.write().expect("token state poisoned");
363        fresh.generation = st.generation + 1;
364        *st = fresh;
365        Ok(())
366    }
367
368    fn until_refresh(&self) -> Duration {
369        let at = self.state.read().expect("token state poisoned").refresh_at;
370        at.saturating_duration_since(Instant::now())
371    }
372}
373
374async fn fetch(api: &dyn TokenApi, secret: &str, source_app_id: &str) -> Result<State, FinamError> {
375    let started = Instant::now();
376    let token = with_deadline("Auth", api.auth(secret, source_app_id)).await?;
377    if token.is_empty() {
378        return Err(FinamError::Auth("server returned an empty token".into()));
379    }
380    let mut info = match with_deadline("TokenDetails", api.token_details(&token)).await {
381        Ok(info) => info,
382        Err(e) => {
383            // The token itself is good; only the lifetime/account list is missing.
384            warn!(error = %e, "TokenDetails failed, falling back to JWT claims");
385            TokenInfo::default()
386        }
387    };
388    // The shorter of the two: a token that the JWT says ends sooner than TokenDetails claims must not be
389    // trusted past its `exp`.
390    info.ttl = match (info.ttl, jwt_ttl(&token)) {
391        (Some(a), Some(b)) => Some(a.min(b)),
392        (a, b) => a.or(b),
393    };
394    let ttl = info.ttl.unwrap_or(ASSUMED_TTL);
395    debug!(?ttl, "finam token issued");
396    Ok(State {
397        token: Arc::from(token),
398        info,
399        // Counted from the start of the Auth call: the server issued it somewhere inside it.
400        refresh_at: started + refresh_after(ttl),
401        generation: 1,
402    })
403}
404
405async fn with_deadline<T>(
406    what: &str,
407    call: impl std::future::Future<Output = Result<T, FinamError>>,
408) -> Result<T, FinamError> {
409    match tokio::time::timeout(AUTH_TIMEOUT, call).await {
410        Ok(result) => result,
411        Err(_) => Err(FinamError::Status {
412            code: tonic::Code::DeadlineExceeded,
413            message: format!("{what}: no response within {AUTH_TIMEOUT:?}"),
414        }),
415    }
416}
417
418async fn refresh_loop(weak: Weak<TokenManager>) {
419    let mut backoff = RETRY_MIN;
420    loop {
421        // Wait without holding a strong reference, so dropping the last Arc ends this task.
422        let (wait, wake) = {
423            let Some(m) = weak.upgrade() else { return };
424            (m.until_refresh(), Arc::clone(&m.wake))
425        };
426        if !wait.is_zero() {
427            tokio::select! {
428                _ = tokio::time::sleep(wait) => {}
429                _ = wake.notified() => {}
430            }
431        }
432        let Some(m) = weak.upgrade() else { return };
433        match m.refresh().await {
434            Ok(()) => backoff = RETRY_MIN,
435            Err(e) => {
436                warn!(error = %e, retry_in = ?backoff, "finam token refresh failed");
437                drop(m);
438                tokio::time::sleep(backoff).await;
439                backoff = (backoff * 2).min(RETRY_MAX);
440            }
441        }
442    }
443}
444
445#[cfg(test)]
446mod tests {
447    use std::sync::Mutex as StdMutex;
448    use std::sync::atomic::{AtomicU64, Ordering};
449
450    use super::*;
451
452    #[test]
453    fn refresh_policy() {
454        // 15 min token: renew 2 min before expiry.
455        assert_eq!(
456            refresh_after(Duration::from_secs(900)),
457            Duration::from_secs(780)
458        );
459        // 60 s token: lead 12 s.
460        assert_eq!(
461            refresh_after(Duration::from_secs(60)),
462            Duration::from_secs(48)
463        );
464        // Very short token: half of the lifetime.
465        assert_eq!(
466            refresh_after(Duration::from_secs(8)),
467            Duration::from_secs(4)
468        );
469        // Long token: lead is capped at 120 s.
470        assert_eq!(
471            refresh_after(Duration::from_secs(86_400)),
472            Duration::from_secs(86_280)
473        );
474    }
475
476    fn jwt(iat: i64, exp: i64) -> String {
477        let h = URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256"}"#);
478        let p = URL_SAFE_NO_PAD.encode(format!(r#"{{"iat":{iat},"exp":{exp}}}"#));
479        format!("{h}.{p}.sig")
480    }
481
482    #[test]
483    fn jwt_claims_give_ttl() {
484        assert_eq!(jwt_ttl(&jwt(1000, 1900)), Some(Duration::from_secs(900)));
485        assert_eq!(jwt_ttl("not-a-jwt"), None);
486        assert_eq!(jwt_ttl(&jwt(2000, 1000)), None);
487    }
488
489    /// Fake auth server: numbered tokens, scripted lifetime and failures.
490    struct FakeApi {
491        issued: AtomicU64,
492        ttl: Option<Duration>,
493        details_fail: bool,
494        fail_next_auths: StdMutex<u32>,
495    }
496
497    impl FakeApi {
498        fn new(ttl: Option<Duration>) -> Arc<Self> {
499            Arc::new(Self {
500                issued: AtomicU64::new(0),
501                ttl,
502                details_fail: false,
503                fail_next_auths: StdMutex::new(0),
504            })
505        }
506    }
507
508    #[async_trait]
509    impl TokenApi for FakeApi {
510        async fn auth(&self, _s: &str, _a: &str) -> Result<String, FinamError> {
511            tokio::task::yield_now().await;
512            {
513                let mut f = self.fail_next_auths.lock().unwrap();
514                if *f > 0 {
515                    *f -= 1;
516                    return Err(FinamError::Status {
517                        code: tonic::Code::Unavailable,
518                        message: "down".into(),
519                    });
520                }
521            }
522            let n = self.issued.fetch_add(1, Ordering::SeqCst) + 1;
523            Ok(format!("jwt-{n}"))
524        }
525
526        async fn token_details(&self, _t: &str) -> Result<TokenInfo, FinamError> {
527            if self.details_fail {
528                return Err(FinamError::Status {
529                    code: tonic::Code::Internal,
530                    message: "x".into(),
531                });
532            }
533            Ok(TokenInfo {
534                ttl: self.ttl,
535                account_ids: vec!["A1".into(), "A2".into()],
536                readonly: false,
537            })
538        }
539    }
540
541    #[tokio::test(start_paused = true)]
542    async fn token_is_renewed_before_expiry_not_after() {
543        let api = FakeApi::new(Some(Duration::from_secs(900)));
544        let m = TokenManager::start(api.clone(), "secret", "")
545            .await
546            .unwrap();
547        assert_eq!(&*m.current(), "jwt-1");
548        assert_eq!(m.account_ids(), vec!["A1", "A2"]);
549
550        // Just before the renewal point the old token is still served.
551        tokio::time::sleep(Duration::from_secs(770)).await;
552        assert_eq!(&*m.current(), "jwt-1");
553
554        // Past it (780 s) and still before expiry (900 s) the token has been replaced.
555        tokio::time::sleep(Duration::from_secs(20)).await;
556        assert_eq!(&*m.current(), "jwt-2");
557        assert_eq!(m.generation(), 2);
558
559        // And keeps being renewed.
560        tokio::time::sleep(Duration::from_secs(780)).await;
561        assert_eq!(&*m.current(), "jwt-3");
562    }
563
564    #[tokio::test(start_paused = true)]
565    async fn failed_refresh_is_retried_and_old_token_kept() {
566        let api = FakeApi::new(Some(Duration::from_secs(600)));
567        let m = TokenManager::start(api.clone(), "secret", "")
568            .await
569            .unwrap();
570        *api.fail_next_auths.lock().unwrap() = 2;
571        // Renewal point at 480 s; first two attempts fail (1 s, 2 s backoff), third succeeds.
572        tokio::time::sleep(Duration::from_secs(482)).await;
573        assert_eq!(&*m.current(), "jwt-1", "old token kept while refresh fails");
574        tokio::time::sleep(Duration::from_secs(5)).await;
575        assert_eq!(&*m.current(), "jwt-2");
576    }
577
578    #[tokio::test(start_paused = true)]
579    async fn lifetime_falls_back_to_jwt_claims_then_assumption() {
580        // TokenDetails fails, token is not a JWT: assumed 5 min -> renewal at 240 s.
581        let api = Arc::new(FakeApi {
582            issued: AtomicU64::new(0),
583            ttl: None,
584            details_fail: true,
585            fail_next_auths: StdMutex::new(0),
586        });
587        let m = TokenManager::start(api, "secret", "").await.unwrap();
588        assert!(m.account_ids().is_empty());
589        tokio::time::sleep(Duration::from_secs(245)).await;
590        assert_eq!(m.generation(), 2);
591    }
592
593    #[tokio::test(start_paused = true)]
594    async fn request_refresh_renews_immediately() {
595        let api = FakeApi::new(Some(Duration::from_secs(900)));
596        let m = TokenManager::start(api, "secret", "").await.unwrap();
597        m.request_refresh();
598        tokio::time::sleep(Duration::from_millis(10)).await;
599        assert_eq!(&*m.current(), "jwt-2");
600    }
601
602    #[tokio::test(start_paused = true)]
603    async fn concurrent_refresh_calls_issue_one_token() {
604        let api = FakeApi::new(Some(Duration::from_secs(900)));
605        let m = TokenManager::start(api.clone(), "secret", "")
606            .await
607            .unwrap();
608        let (a, b) = tokio::join!(m.refresh(), m.refresh());
609        a.unwrap();
610        b.unwrap();
611        assert_eq!(api.issued.load(Ordering::SeqCst), 2);
612    }
613
614    #[tokio::test(start_paused = true)]
615    async fn refresher_stops_when_manager_is_dropped() {
616        let api = FakeApi::new(Some(Duration::from_secs(60)));
617        let m = TokenManager::start(api.clone(), "secret", "")
618            .await
619            .unwrap();
620        drop(m);
621        tokio::time::sleep(Duration::from_secs(600)).await;
622        assert_eq!(
623            api.issued.load(Ordering::SeqCst),
624            1,
625            "no refresh after drop"
626        );
627    }
628
629    #[tokio::test]
630    async fn empty_secret_is_a_config_error() {
631        let api = FakeApi::new(None);
632        let err = TokenManager::start(api, "  ", "").await.unwrap_err();
633        assert!(matches!(err, FinamError::Config(_)));
634    }
635
636    #[tokio::test]
637    async fn debug_hides_the_secret() {
638        let m = TokenManager::start(FakeApi::new(None), "hunter2-secret", "")
639            .await
640            .unwrap();
641        let shown = format!("{m:?}");
642        assert!(!shown.contains("hunter2"), "{shown}");
643    }
644
645    /// An auth server whose token is a JWT of a given lifetime, and whose `TokenDetails` says another one.
646    struct JwtApi {
647        jwt_ttl: i64,
648        details_ttl: Option<Duration>,
649    }
650
651    #[async_trait]
652    impl TokenApi for JwtApi {
653        async fn auth(&self, _s: &str, _a: &str) -> Result<String, FinamError> {
654            Ok(jwt(1000, 1000 + self.jwt_ttl))
655        }
656
657        async fn token_details(&self, _t: &str) -> Result<TokenInfo, FinamError> {
658            Ok(TokenInfo {
659                ttl: self.details_ttl,
660                account_ids: vec![],
661                readonly: false,
662            })
663        }
664    }
665
666    async fn ttl_of(jwt_ttl: i64, details_ttl: Option<u64>) -> Option<Duration> {
667        let api = Arc::new(JwtApi {
668            jwt_ttl,
669            details_ttl: details_ttl.map(Duration::from_secs),
670        });
671        TokenManager::start(api, "secret", "")
672            .await
673            .unwrap()
674            .info()
675            .ttl
676    }
677
678    #[tokio::test]
679    async fn the_lifetime_is_the_shorter_of_the_details_and_the_claims() {
680        // The claims say 600 s, the details say 900 s: 600 s.
681        assert_eq!(ttl_of(600, Some(900)).await, Some(Duration::from_secs(600)));
682        // The other way round: 300 s.
683        assert_eq!(ttl_of(900, Some(300)).await, Some(Duration::from_secs(300)));
684        // One of the two is enough.
685        assert_eq!(ttl_of(600, None).await, Some(Duration::from_secs(600)));
686        // Claims that are not a lifetime (exp before iat) leave the details alone.
687        assert_eq!(ttl_of(-5, Some(120)).await, Some(Duration::from_secs(120)));
688    }
689
690    /// An auth server that takes the request and never answers.
691    struct HangApi {
692        auth_hangs: bool,
693    }
694
695    #[async_trait]
696    impl TokenApi for HangApi {
697        async fn auth(&self, _s: &str, _a: &str) -> Result<String, FinamError> {
698            if self.auth_hangs {
699                std::future::pending::<()>().await;
700            }
701            Ok("jwt-1".into())
702        }
703
704        async fn token_details(&self, _t: &str) -> Result<TokenInfo, FinamError> {
705            std::future::pending::<()>().await;
706            unreachable!()
707        }
708    }
709
710    #[tokio::test(start_paused = true)]
711    async fn an_auth_call_that_never_answers_ends_with_a_deadline() {
712        let started = tokio::time::Instant::now();
713        let err = TokenManager::start(Arc::new(HangApi { auth_hangs: true }), "secret", "")
714            .await
715            .unwrap_err();
716        assert!(
717            matches!(
718                err,
719                FinamError::Status { code: tonic::Code::DeadlineExceeded, ref message } if message.contains("Auth")
720            ),
721            "{err:?}"
722        );
723        assert_eq!(started.elapsed(), AUTH_TIMEOUT);
724    }
725
726    #[tokio::test(start_paused = true)]
727    async fn token_details_that_never_answer_do_not_stop_the_start() {
728        // The token itself is good: the lifetime falls back to the assumption.
729        let m = TokenManager::start(Arc::new(HangApi { auth_hangs: false }), "secret", "")
730            .await
731            .unwrap();
732        assert_eq!(&*m.current(), "jwt-1");
733        assert!(m.account_ids().is_empty());
734        // No lifetime is known; the renewal is planned with the assumption.
735        assert_eq!(m.info().ttl, None);
736    }
737
738    #[tokio::test(start_paused = true)]
739    async fn a_hanging_renewal_does_not_hold_the_lock_for_ever() {
740        // Start with a good server, then renew against a hanging one: the renewal ends with a deadline
741        // and the old token stays.
742        struct Flip(std::sync::atomic::AtomicBool);
743        #[async_trait]
744        impl TokenApi for Flip {
745            async fn auth(&self, _s: &str, _a: &str) -> Result<String, FinamError> {
746                if self.0.load(Ordering::SeqCst) {
747                    std::future::pending::<()>().await;
748                }
749                Ok("jwt-1".into())
750            }
751            async fn token_details(&self, _t: &str) -> Result<TokenInfo, FinamError> {
752                Ok(TokenInfo::default())
753            }
754        }
755        let api = Arc::new(Flip(std::sync::atomic::AtomicBool::new(false)));
756        let m = TokenManager::start(api.clone(), "secret", "")
757            .await
758            .unwrap();
759        api.0.store(true, Ordering::SeqCst);
760        let err = m.refresh().await.unwrap_err();
761        assert!(matches!(
762            err,
763            FinamError::Status {
764                code: tonic::Code::DeadlineExceeded,
765                ..
766            }
767        ));
768        assert_eq!(&*m.current(), "jwt-1");
769        // And a second renewal can take the lock again.
770        assert!(m.refresh().await.is_err());
771    }
772}
brokapi_finam_protocol/auth.rs

brokapi_finam_protocol/
auth.rs
