brokapi_alor_protocol/http/client.rs
1//! The authorized HTTP client: token handling, the 401 repeat of reads and the mapping of error replies.
2//!
3//! The generic methods are here ([`AlorHttpClient::read`], [`AlorHttpClient::command`]). The endpoint methods
4//! (`get_order_book`, `create_limit_order`, ...) are in [`super::endpoints`].
5
6use std::sync::Arc;
7use std::time::Duration;
8
9use base64::Engine as _;
10pub use reqwest::Method;
11use reqwest::{Client, StatusCode};
12use serde_json::Value;
13use tokio::sync::{Mutex, RwLock};
14use tracing::{debug, warn};
15
16use crate::config::AlorConfig;
17use crate::error::{AlorApiError, ApiErrorBody};
18
19/// How long an access token is used before a new one is asked for. The broker recommends a new token every
20/// 20-25 minutes of the 30 that a token lives.
21pub const ACCESS_TOKEN_REUSE: Duration = Duration::from_secs(20 * 60);
22
23/// The access token and the moment after which the client does not trust it any more.
24#[derive(Clone)]
25struct CachedToken {
26 token: String,
27 valid_until: tokio::time::Instant,
28}
29
30struct Shared {
31 access: RwLock<Option<CachedToken>>,
32 refresh_token: String,
33 refresh_mutex: Mutex<()>,
34}
35
36/// The authorized HTTP client of the ALOR OpenAPI.
37///
38/// It holds the refresh token and the current access token and adds `Authorization: Bearer ...` to every call:
39///
40/// * The first call trades the refresh token for an access token (`POST /refresh` of the authorization
41/// server). The access token is cached for 20 minutes (or less, if its own expiry is nearer).
42/// * When the broker answers 401 to a **read**, the client gets a new access token and repeats the **same
43/// request once**.
44/// * A **command** (placing, changing or cancelling an order) is never repeated, not even after a 401: the
45/// client renews the token for the next call and returns the 401 as an error.
46/// * A transport error or a timeout is never repeated.
47///
48/// The client is cheap to clone. Clones share the tokens and the connection pool. The endpoint methods (one per
49/// REST call) are documented on this type too, they are in [`super::endpoints`]. The broker limits HTTP calls to
50/// 100 a second per user; the client does not throttle, the connector does it through the shared limiter.
51///
52/// # Examples
53///
54/// ```
55/// # #[tokio::main(flavor = "current_thread")]
56/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
57/// # let server = brokapi_mocks::alor::MockServer::start(Default::default()).await;
58/// # let cfg = brokapi_alor_protocol::config::AlorConfig::for_mock("mock-refresh-token", &server.http_base);
59/// use brokapi_alor_protocol::http::client::AlorHttpClient;
60///
61/// let client = AlorHttpClient::new(cfg)?;
62/// let book = client.get_order_book("MOEX", "SBER", Some("TQBR"), Some(5)).await?;
63/// println!("{} levels of bids", book.bids.len());
64/// # Ok(())
65/// # }
66/// ```
67#[derive(Clone)]
68pub struct AlorHttpClient {
69 client: Client,
70 shared: Arc<Shared>,
71 http_base_url: Arc<str>,
72 auth_base_url: Arc<str>,
73 ws_base_url: Arc<str>,
74}
75
76impl AlorHttpClient {
77 /// Creates a client from a configuration. No network call is made.
78 ///
79 /// # Errors
80 ///
81 /// [`AlorApiError::Http`] (no status) if the HTTP stack cannot be built.
82 ///
83 /// # Examples
84 ///
85 /// ```
86 /// use brokapi_alor_protocol::config::AlorConfig;
87 /// use brokapi_alor_protocol::http::client::AlorHttpClient;
88 ///
89 /// let client = AlorHttpClient::new(AlorConfig::new("refresh-token".to_string()))?;
90 /// assert_eq!(client.base_url(), "https://api.alor.ru");
91 /// # Ok::<(), brokapi_alor_protocol::error::AlorApiError>(())
92 /// ```
93 pub fn new(config: AlorConfig) -> Result<Self, AlorApiError> {
94 let client = Client::builder()
95 .timeout(config.request_timeout)
96 .gzip(true)
97 .build()?;
98 Ok(Self {
99 client,
100 shared: Arc::new(Shared {
101 access: RwLock::new(None),
102 refresh_token: config.refresh_token,
103 refresh_mutex: Mutex::new(()),
104 }),
105 http_base_url: Arc::from(config.http_base_url.as_str()),
106 auth_base_url: Arc::from(config.auth_base_url.as_str()),
107 ws_base_url: Arc::from(config.ws_base_url.as_str()),
108 })
109 }
110
111 /// The access token now: the cached one, or a new one if there is none or it is past its time.
112 ///
113 /// The WebSocket session calls this before every message that needs a token.
114 ///
115 /// # Errors
116 ///
117 /// As [`AlorHttpClient::refresh_access_token`], and only when a new token is needed.
118 ///
119 /// # Examples
120 ///
121 /// ```
122 /// # #[tokio::main(flavor = "current_thread")]
123 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
124 /// # let server = brokapi_mocks::alor::MockServer::start(Default::default()).await;
125 /// # let cfg = brokapi_alor_protocol::config::AlorConfig::for_mock("mock-refresh-token", &server.http_base);
126 /// use brokapi_alor_protocol::http::client::AlorHttpClient;
127 ///
128 /// let client = AlorHttpClient::new(cfg)?;
129 /// let first = client.get_access_token().await?;
130 /// let second = client.get_access_token().await?; // from the cache, no second token request
131 /// assert_eq!(first, second);
132 /// # Ok(())
133 /// # }
134 /// ```
135 pub async fn get_access_token(&self) -> Result<String, AlorApiError> {
136 {
137 let cached = self.shared.access.read().await;
138 if let Some(c) = cached.as_ref()
139 && tokio::time::Instant::now() < c.valid_until
140 {
141 return Ok(c.token.clone());
142 }
143 }
144 self.refresh_access_token().await
145 }
146
147 /// Trades the refresh token for a new access token and stores it. Always calls the authorization server.
148 ///
149 /// Sends `POST /refresh` with the refresh token in a JSON body (not in the URL, so it stays out of logs).
150 /// The new access token is cached. Refreshes are serialised by a lock.
151 ///
152 /// # Errors
153 ///
154 /// * [`AlorApiError::Auth`]: the server refused the refresh token (HTTP 400, 401 or 403).
155 /// * [`AlorApiError::Http`]: a transport failure, or another status.
156 /// * [`AlorApiError::Parse`]: the answer has no `AccessToken`.
157 pub async fn refresh_access_token(&self) -> Result<String, AlorApiError> {
158 let _guard = self.shared.refresh_mutex.lock().await;
159 let url = format!("{}/refresh", self.auth_base_url);
160 debug!("Refreshing the access token...");
161 let resp = self
162 .client
163 .post(&url)
164 .json(&serde_json::json!({ "token": self.shared.refresh_token }))
165 .send()
166 .await?;
167 let status = resp.status();
168 let body = resp.text().await?;
169 if !status.is_success() {
170 let message = error_text(&body);
171 return Err(if matches!(status.as_u16(), 400 | 401 | 403) {
172 AlorApiError::Auth { message }
173 } else {
174 AlorApiError::Http {
175 status: Some(status.as_u16()),
176 message,
177 }
178 });
179 }
180 let token = serde_json::from_str::<Value>(&body)?
181 .get("AccessToken")
182 .and_then(Value::as_str)
183 .map(str::to_string)
184 .ok_or_else(|| {
185 AlorApiError::Parse("the answer of /refresh has no AccessToken".into())
186 })?;
187 let lifetime = token_lifetime(&token, unix_now());
188 *self.shared.access.write().await = Some(CachedToken {
189 token: token.clone(),
190 valid_until: tokio::time::Instant::now() + lifetime,
191 });
192 Ok(token)
193 }
194
195 /// The portfolios named in the claims of the current access token (the `portfolios` claim of the JWT).
196 ///
197 /// The connector uses it to find the portfolio when the config names none. Empty when the token is not a
198 /// JWT or carries no portfolios.
199 ///
200 /// # Errors
201 ///
202 /// As [`AlorHttpClient::get_access_token`].
203 pub async fn token_portfolios(&self) -> Result<Vec<String>, AlorApiError> {
204 let token = self.get_access_token().await?;
205 Ok(jwt_claims(&token)
206 .and_then(|c| {
207 c.get("portfolios")
208 .and_then(Value::as_str)
209 .map(str::to_string)
210 })
211 .map(|s| s.split_whitespace().map(str::to_string).collect())
212 .unwrap_or_default())
213 }
214
215 /// `GET`s a path with query parameters and parses the JSON reply into `T`.
216 ///
217 /// `path` starts with `/` and is added to the base URL. A 401 is answered with one token refresh and one
218 /// repeat. Other failures are not retried. The typed methods of [`super::endpoints`] call it.
219 ///
220 /// # Errors
221 ///
222 /// * [`AlorApiError::NotFound`]: HTTP 404. [`AlorApiError::RateLimitExceeded`]: HTTP 429.
223 /// * [`AlorApiError::Http`]: any other non-success status, or a transport failure.
224 /// * [`AlorApiError::Parse`]: the reply is a success but is not the JSON that `T` expects.
225 /// * [`AlorApiError::Auth`]: the token refresh after a 401 failed.
226 ///
227 /// # Examples
228 ///
229 /// ```
230 /// # #[tokio::main(flavor = "current_thread")]
231 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
232 /// # let server = brokapi_mocks::alor::MockServer::start(Default::default()).await;
233 /// # let cfg = brokapi_alor_protocol::config::AlorConfig::for_mock("mock-refresh-token", &server.http_base);
234 /// use brokapi_alor_protocol::error::AlorApiError;
235 /// use brokapi_alor_protocol::http::client::AlorHttpClient;
236 ///
237 /// let client = AlorHttpClient::new(cfg)?;
238 /// let time: serde_json::Value = client.read("/md/v2/time", &[]).await?;
239 /// assert!(time.is_number());
240 ///
241 /// // A path the broker does not know ends in an error that carries the status.
242 /// let err = client.read::<serde_json::Value>("/no/such/path", &[]).await.unwrap_err();
243 /// assert!(matches!(err, AlorApiError::NotFound { .. }));
244 /// # Ok(())
245 /// # }
246 /// ```
247 pub async fn read<T: serde::de::DeserializeOwned>(
248 &self,
249 path: &str,
250 query: &[(&str, String)],
251 ) -> Result<T, AlorApiError> {
252 let url = format!("{}{}", self.http_base_url, path);
253 let token = self.get_access_token().await?;
254 let resp = self
255 .send(Method::GET, &url, query, None::<&()>, &[], &token)
256 .await?;
257 if resp.status() == StatusCode::UNAUTHORIZED {
258 warn!("Got 401, refreshing the token and repeating the read once...");
259 let new_token = self.refresh_access_token().await?;
260 let resp = self
261 .send(Method::GET, &url, query, None::<&()>, &[], &new_token)
262 .await?;
263 return self.handle(resp).await;
264 }
265 self.handle(resp).await
266 }
267
268 /// Sends a command (place, change or cancel an order) and parses the JSON reply into `T`. **Never repeats
269 /// the request.**
270 ///
271 /// A repeat of a command is the one way to place an order twice, so the client sends it once. If the
272 /// broker answers 401, the client renews the access token (so the next call works) and returns the 401 as
273 /// [`AlorApiError::Http`] with `status: Some(401)`; a 401 means the broker did not act. A transport error
274 /// or a timeout is an unknown outcome, as for every command. `req_id` becomes the header `X-REQID`: the
275 /// broker answers a repeated id with a copy of its first answer instead of doing the command again.
276 ///
277 /// # Errors
278 ///
279 /// The same as [`AlorHttpClient::read`].
280 pub async fn command<T, B>(
281 &self,
282 method: Method,
283 path: &str,
284 query: &[(&str, String)],
285 body: Option<&B>,
286 req_id: Option<&str>,
287 ) -> Result<T, AlorApiError>
288 where
289 T: serde::de::DeserializeOwned,
290 B: serde::Serialize,
291 {
292 let url = format!("{}{}", self.http_base_url, path);
293 let token = self.get_access_token().await?;
294 let headers: Vec<(&str, &str)> = req_id.map(|id| vec![("X-REQID", id)]).unwrap_or_default();
295 let resp = self
296 .send(method, &url, query, body, &headers, &token)
297 .await?;
298 if resp.status() == StatusCode::UNAUTHORIZED {
299 warn!("Got 401 on a command: renewing the token, the command is not repeated");
300 // The next call needs a fresh token. A failure of the renewal does not hide the 401.
301 let _ = self.refresh_access_token().await;
302 }
303 self.handle(resp).await
304 }
305
306 /// `GET`s a path and returns the status and the body text, without parsing and without mapping errors.
307 ///
308 /// An error status is **not** an `Err` here: you get it as the first element of the pair. Only a 401 is
309 /// handled (one refresh, one repeat).
310 ///
311 /// # Errors
312 ///
313 /// [`AlorApiError::Http`] (no status) on a transport failure, and the errors of
314 /// [`AlorHttpClient::refresh_access_token`] when a 401 needs a new token.
315 pub async fn read_raw(
316 &self,
317 path: &str,
318 query: &[(&str, String)],
319 ) -> Result<(StatusCode, String), AlorApiError> {
320 let url = format!("{}{}", self.http_base_url, path);
321 let token = self.get_access_token().await?;
322 let mut resp = self
323 .send(Method::GET, &url, query, None::<&()>, &[], &token)
324 .await?;
325 if resp.status() == StatusCode::UNAUTHORIZED {
326 let new_token = self.refresh_access_token().await?;
327 resp = self
328 .send(Method::GET, &url, query, None::<&()>, &[], &new_token)
329 .await?;
330 }
331 let status = resp.status();
332 Ok((status, resp.text().await?))
333 }
334
335 async fn send<B: serde::Serialize>(
336 &self,
337 method: Method,
338 url: &str,
339 query: &[(&str, String)],
340 body: Option<&B>,
341 headers: &[(&str, &str)],
342 token: &str,
343 ) -> Result<reqwest::Response, AlorApiError> {
344 let mut req = self.client.request(method, url).bearer_auth(token);
345 if !query.is_empty() {
346 req = req.query(query);
347 }
348 for (k, v) in headers {
349 req = req.header(*k, *v);
350 }
351 if let Some(b) = body {
352 req = req.json(b);
353 }
354 Ok(req.send().await?)
355 }
356
357 /// Turns a reply into `T` or into the matching [`AlorApiError`].
358 async fn handle<T: serde::de::DeserializeOwned>(
359 &self,
360 resp: reqwest::Response,
361 ) -> Result<T, AlorApiError> {
362 let status = resp.status();
363 let body = resp.text().await?;
364 if status.is_success() {
365 // An empty success body (some deletes) reads as JSON null.
366 let text = if body.trim().is_empty() {
367 "null"
368 } else {
369 body.as_str()
370 };
371 return serde_json::from_str(text).map_err(|e| {
372 AlorApiError::Parse(format!(
373 "Failed to parse response: {e} | body: {}",
374 truncate_utf8(&body, 200)
375 ))
376 });
377 }
378 let code = status.as_u16();
379 Err(match code {
380 404 => AlorApiError::NotFound {
381 message: error_text(&body),
382 },
383 429 => AlorApiError::RateLimitExceeded,
384 _ => AlorApiError::Http {
385 status: Some(code),
386 message: error_text(&body),
387 },
388 })
389 }
390
391 /// The underlying `reqwest` client, for calls this crate does not wrap. It carries **no** authorization:
392 /// add the token yourself with [`AlorHttpClient::get_access_token`].
393 pub fn inner_client(&self) -> &Client {
394 &self.client
395 }
396
397 /// The REST base URL this client calls, without a trailing slash.
398 pub fn base_url(&self) -> &str {
399 &self.http_base_url
400 }
401
402 /// The WebSocket base URL that sessions started from this client connect to.
403 pub fn ws_base_url(&self) -> &str {
404 &self.ws_base_url
405 }
406}
407
408/// The text of an error body: its `message`, else the whole body.
409fn error_text(body: &str) -> String {
410 match serde_json::from_str::<ApiErrorBody>(body) {
411 Ok(ApiErrorBody {
412 message: Some(m),
413 code,
414 }) => match code {
415 Some(c) if !c.is_empty() => format!("{c}: {m}"),
416 _ => m,
417 },
418 _ => body.to_string(),
419 }
420}
421
422fn unix_now() -> u64 {
423 std::time::SystemTime::now()
424 .duration_since(std::time::UNIX_EPOCH)
425 .map(|d| d.as_secs())
426 .unwrap_or(0)
427}
428
429/// The claims of a JWT (the middle part), if `token` is one.
430pub fn jwt_claims(token: &str) -> Option<Value> {
431 let payload = token.split('.').nth(1)?;
432 let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
433 .decode(payload.trim_end_matches('='))
434 .ok()?;
435 serde_json::from_slice(&bytes).ok()
436}
437
438/// How long to use an access token: [`ACCESS_TOKEN_REUSE`], or 80% of what is left of its own life if that
439/// is shorter. A token that is not a JWT gets the 20 minutes.
440pub fn token_lifetime(token: &str, now_unix: u64) -> Duration {
441 let exp = jwt_claims(token).and_then(|c| c.get("exp").and_then(Value::as_u64));
442 match exp {
443 Some(exp) => {
444 let left = Duration::from_secs(exp.saturating_sub(now_unix));
445 left.mul_f64(0.8).min(ACCESS_TOKEN_REUSE)
446 }
447 None => ACCESS_TOKEN_REUSE,
448 }
449}
450
451/// The first `max` bytes of `s`, cut back to a character boundary.
452fn truncate_utf8(s: &str, max: usize) -> &str {
453 if s.len() <= max {
454 return s;
455 }
456 let mut end = max;
457 while !s.is_char_boundary(end) {
458 end -= 1;
459 }
460 &s[..end]
461}
462
463#[cfg(test)]
464mod tests {
465 use super::*;
466
467 fn jwt(claims: &str) -> String {
468 let enc = |s: &str| base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(s);
469 format!("{}.{}.sig", enc("{}"), enc(claims))
470 }
471
472 #[test]
473 fn a_token_is_used_for_twenty_minutes_or_less() {
474 assert_eq!(token_lifetime("not-a-jwt", 0), ACCESS_TOKEN_REUSE);
475 let long = jwt(r#"{"exp": 100000}"#);
476 assert_eq!(token_lifetime(&long, 0), ACCESS_TOKEN_REUSE);
477 // 10 minutes left: 80% of it.
478 let short = jwt(r#"{"exp": 600}"#);
479 assert_eq!(token_lifetime(&short, 0), Duration::from_secs(480));
480 // Already past its time.
481 assert_eq!(token_lifetime(&short, 1000), Duration::ZERO);
482 }
483
484 #[test]
485 fn claims_are_read_from_a_jwt() {
486 let t = jwt(r#"{"portfolios":"D1 7500A","exp":5}"#);
487 let c = jwt_claims(&t).unwrap();
488 assert_eq!(c["portfolios"], "D1 7500A");
489 assert!(jwt_claims("nope").is_none());
490 }
491
492 #[test]
493 fn the_error_text_prefers_the_message_of_the_body() {
494 assert_eq!(error_text(r#"{"message":"bad","code":"X"}"#), "X: bad");
495 assert_eq!(error_text(r#"{"message":"bad"}"#), "bad");
496 assert_eq!(error_text("plain"), "plain");
497 }
498
499 #[test]
500 fn truncation_never_splits_a_character() {
501 let s = "ж".repeat(150);
502 let cut = truncate_utf8(&s, 199);
503 assert_eq!(cut.len(), 198);
504 }
505}