brokapi_finam_protocol/
client.rs

1//! Channel setup and authorized service clients.
2//!
3//! [`FinamApi`] is the entry point: one lazily connecting channel plus one shared
4//! [`TokenManager`]. Every service client it hands out puts the current JWT
5//! into the `authorization` header of each call, so a client never holds an expired token.
6//!
7//! # Examples
8//!
9//! ```
10//! use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata::QuoteRequest;
11//! use brokapi_finam_protocol::{ApiEndpoint, FinamApi, start_token_manager};
12//! # use brokapi_mocks::finam::{MockServer, SECRET};
13//!
14//! # #[tokio::main(flavor = "current_thread")]
15//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
16//! # let server = MockServer::start().await;
17//! # let endpoint = ApiEndpoint::new(server.url.clone());
18//! # let secret = SECRET;
19//! // `endpoint` is `ApiEndpoint::default()` for the real broker, `secret` is your API secret.
20//! let tokens = start_token_manager(&endpoint, secret, "my-app").await?;
21//! let api = FinamApi::connect(&endpoint, tokens, false)?;
22//!
23//! let quote = api
24//!     .market_data()
25//!     .last_quote(QuoteRequest { symbol: "SBER@MISX".into() })
26//!     .await?
27//!     .into_inner();
28//! assert_eq!(quote.symbol, "SBER@MISX");
29//! # Ok(())
30//! # }
31//! ```
32
33use std::sync::Arc;
34use std::time::Duration;
35
36use tonic::metadata::MetadataValue;
37use tonic::service::Interceptor;
38use tonic::service::interceptor::InterceptedService;
39use tonic::transport::{Channel, ClientTlsConfig, Endpoint};
40use tonic::{Request, Status};
41
42use crate::auth::{GrpcTokenApi, TokenManager};
43use crate::error::FinamError;
44use crate::pb::grpc::tradeapi::v1::accounts::accounts_service_client::AccountsServiceClient;
45use crate::pb::grpc::tradeapi::v1::assets::assets_service_client::AssetsServiceClient;
46use crate::pb::grpc::tradeapi::v1::auth::auth_service_client::AuthServiceClient;
47use crate::pb::grpc::tradeapi::v1::corporateactions::corporate_actions_service_client::CorporateActionsServiceClient;
48use crate::pb::grpc::tradeapi::v1::marketdata::market_data_service_client::MarketDataServiceClient;
49use crate::pb::grpc::tradeapi::v1::metrics::usage_metrics_service_client::UsageMetricsServiceClient;
50use crate::pb::grpc::tradeapi::v1::orders::orders_service_client::OrdersServiceClient;
51use crate::pb::grpc::tradeapi::v1::reports::reports_service_client::ReportsServiceClient;
52
53/// Address of the production Trade API (`https://api.finam.ru:443`, TLS).
54pub const DEFAULT_ENDPOINT: &str = "https://api.finam.ru:443";
55
56/// Where and how to connect: the address and the timeouts of the channel.
57///
58/// # Examples
59///
60/// ```
61/// use std::time::Duration;
62/// use brokapi_finam_protocol::{ApiEndpoint, DEFAULT_ENDPOINT};
63///
64/// // Production: the default.
65/// assert_eq!(ApiEndpoint::default().url, DEFAULT_ENDPOINT);
66///
67/// // A proxy or a local imitation: plain `http://`, a shorter connect timeout.
68/// let mut ep = ApiEndpoint::new("http://127.0.0.1:50051");
69/// ep.connect_timeout = Duration::from_secs(2);
70/// assert_eq!(ep.keepalive_interval, Duration::from_secs(30));
71/// ```
72#[derive(Debug, Clone)]
73pub struct ApiEndpoint {
74    /// `https://host:port` (TLS, webpki roots) or `http://host:port` (plaintext, tests/proxies).
75    pub url: String,
76    /// How long to wait for the TCP and TLS handshake. Default 10 seconds.
77    pub connect_timeout: Duration,
78    /// HTTP/2 PING interval, detects dead connections under long-lived streams. Default 30 seconds.
79    ///
80    /// Pings are also sent while the channel is idle.
81    pub keepalive_interval: Duration,
82    /// How long to wait for the answer to a PING before the connection is declared dead.
83    /// Default 10 seconds.
84    pub keepalive_timeout: Duration,
85}
86
87impl ApiEndpoint {
88    /// Endpoint for `url` with the default timeouts (connect 10 s, keepalive 30 s and 10 s).
89    ///
90    /// The URL is not checked here. A bad URL is reported by [`connect_lazy`] as
91    /// [`FinamError::Config`].
92    pub fn new(url: impl Into<String>) -> Self {
93        Self {
94            url: url.into(),
95            connect_timeout: Duration::from_secs(10),
96            keepalive_interval: Duration::from_secs(30),
97            keepalive_timeout: Duration::from_secs(10),
98        }
99    }
100}
101
102impl Default for ApiEndpoint {
103    fn default() -> Self {
104        Self::new(DEFAULT_ENDPOINT)
105    }
106}
107
108/// Builds a lazily connecting channel to `ep`.
109///
110/// Nothing is sent here. Connection errors surface on the first call, and tonic reconnects on its
111/// own after a drop. TLS (webpki roots) is switched on when the URL starts with `https://`.
112///
113/// # Errors
114///
115/// [`FinamError::Config`] when the URL is not a valid endpoint or the TLS settings are refused.
116///
117/// # Examples
118///
119/// ```
120/// use brokapi_finam_protocol::client::connect_lazy;
121/// use brokapi_finam_protocol::{ApiEndpoint, FinamError};
122///
123/// # #[tokio::main(flavor = "current_thread")]
124/// # async fn main() {
125/// // A valid address gives a channel without touching the network.
126/// assert!(connect_lazy(&ApiEndpoint::new("http://127.0.0.1:1")).is_ok());
127/// // Garbage is a configuration error.
128/// let err = connect_lazy(&ApiEndpoint::new("not a url")).unwrap_err();
129/// assert!(matches!(err, FinamError::Config(_)));
130/// # }
131/// ```
132pub fn connect_lazy(ep: &ApiEndpoint) -> Result<Channel, FinamError> {
133    let mut endpoint = Endpoint::from_shared(ep.url.clone())
134        .map_err(|e| FinamError::Config(format!("bad endpoint {:?}: {e}", ep.url)))?
135        .connect_timeout(ep.connect_timeout)
136        .tcp_nodelay(true)
137        .http2_keep_alive_interval(ep.keepalive_interval)
138        .keep_alive_timeout(ep.keepalive_timeout)
139        .keep_alive_while_idle(true);
140    if ep.url.starts_with("https://") {
141        endpoint = endpoint
142            .tls_config(ClientTlsConfig::new().with_webpki_roots())
143            .map_err(|e| FinamError::Config(format!("tls: {e}")))?;
144    }
145    Ok(endpoint.connect_lazy())
146}
147
148/// Starts a [`TokenManager`] for `secret` on a fresh channel to `ep`.
149///
150/// Logs in once (`AuthService.Auth`) and starts the background refresher. Put the returned `Arc` into
151/// every [`FinamApi`] that uses this secret: one secret needs one manager.
152///
153/// `source_app_id` is sent with `Auth` as the application id. An empty string is allowed.
154///
155/// # Errors
156///
157/// - [`FinamError::Config`]: the secret is empty or the endpoint URL is invalid.
158/// - [`FinamError::Auth`]: the server returned an empty token.
159/// - [`FinamError::Status`]: the server refused the secret (code `UNAUTHENTICATED`) or could not
160///   be reached.
161///
162/// # Examples
163///
164/// ```
165/// use brokapi_finam_protocol::{ApiEndpoint, FinamError, start_token_manager};
166/// # use brokapi_mocks::finam::{MockServer, SECRET};
167///
168/// # #[tokio::main(flavor = "current_thread")]
169/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
170/// # let server = MockServer::start().await;
171/// # let endpoint = ApiEndpoint::new(server.url.clone());
172/// # let secret = SECRET;
173/// let tokens = start_token_manager(&endpoint, secret, "my-app").await?;
174/// assert!(!tokens.current().is_empty());
175///
176/// // A wrong secret is refused by the server.
177/// let err = start_token_manager(&endpoint, "wrong-secret", "my-app").await.unwrap_err();
178/// assert!(err.is_unauthenticated());
179/// // An empty secret never leaves the process.
180/// let err = start_token_manager(&endpoint, " ", "my-app").await.unwrap_err();
181/// assert!(matches!(err, FinamError::Config(_)));
182/// # Ok(())
183/// # }
184/// ```
185pub async fn start_token_manager(
186    ep: &ApiEndpoint,
187    secret: &str,
188    source_app_id: &str,
189) -> Result<Arc<TokenManager>, FinamError> {
190    let channel = connect_lazy(ep)?;
191    TokenManager::start(Arc::new(GrpcTokenApi::new(channel)), secret, source_app_id).await
192}
193
194/// Tonic interceptor that puts the current JWT into the `authorization` metadata of every call.
195///
196/// Built inside [`FinamApi`], not by hand. It asks the [`TokenManager`] for the token on each call,
197/// so a renewed token is used at once. A token that cannot be written as header text fails the
198/// call with status `UNAUTHENTICATED` before anything is sent.
199#[derive(Clone)]
200pub struct AuthInterceptor {
201    tokens: Arc<TokenManager>,
202    bearer: bool,
203}
204
205impl Interceptor for AuthInterceptor {
206    fn call(&mut self, mut req: Request<()>) -> Result<Request<()>, Status> {
207        let token = self.tokens.current();
208        // Official SDKs send the bare JWT; some doc examples show `Bearer <jwt>`.
209        let value = if self.bearer {
210            format!("Bearer {token}")
211        } else {
212            token.to_string()
213        };
214        let mut value = MetadataValue::try_from(value.as_str())
215            .map_err(|_| Status::unauthenticated("token is not valid metadata"))?;
216        // Sensitive: the JWT does not show in the Debug output of the request or in HTTP/2 traces.
217        value.set_sensitive(true);
218        req.metadata_mut().insert("authorization", value);
219        Ok(req)
220    }
221}
222
223/// Channel with the [`AuthInterceptor`] on it: the transport type of every client of [`FinamApi`].
224pub type Authorized = InterceptedService<Channel, AuthInterceptor>;
225
226/// One channel plus a shared token manager. Hands out service clients that attach the JWT to each call.
227///
228/// Cheap to clone (the channel is reference counted). Each accessor builds a new client over the same
229/// channel, so there is no need to keep clients around.
230///
231/// A streaming call keeps the JWT it was opened with. Open it again after the token is renewed
232/// (the connectors do this for their own streams).
233///
234/// See the [module docs](self) for a complete example.
235#[derive(Clone)]
236pub struct FinamApi {
237    channel: Channel,
238    tokens: Arc<TokenManager>,
239    bearer: bool,
240}
241
242impl FinamApi {
243    /// Opens a lazy channel to `ep` and uses `tokens` for authorization.
244    ///
245    /// `bearer` selects the header format. `false` sends the bare JWT, which is what the official
246    /// SDKs do. `true` sends `Bearer <jwt>`, which some examples in the broker documentation show.
247    /// Which of the two the live broker accepts is not verified against the live API here.
248    ///
249    /// # Errors
250    ///
251    /// [`FinamError::Config`] when the endpoint URL is invalid (see [`connect_lazy`]).
252    ///
253    /// # Examples
254    ///
255    /// ```
256    /// use brokapi_finam_protocol::{ApiEndpoint, FinamApi, FinamError, start_token_manager};
257    /// # use brokapi_mocks::finam::{MockServer, SECRET};
258    ///
259    /// # #[tokio::main(flavor = "current_thread")]
260    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
261    /// # let server = MockServer::start().await;
262    /// # let endpoint = ApiEndpoint::new(server.url.clone());
263    /// # let tokens = start_token_manager(&endpoint, SECRET, "").await?;
264    /// let api = FinamApi::connect(&endpoint, tokens.clone(), false)?;
265    /// assert!(std::sync::Arc::ptr_eq(api.tokens(), &tokens));
266    ///
267    /// let bad = FinamApi::connect(&ApiEndpoint::new("not a url"), tokens, false);
268    /// assert!(matches!(bad, Err(FinamError::Config(_))));
269    /// # Ok(())
270    /// # }
271    /// ```
272    pub fn connect(
273        ep: &ApiEndpoint,
274        tokens: Arc<TokenManager>,
275        bearer: bool,
276    ) -> Result<Self, FinamError> {
277        Ok(Self {
278            channel: connect_lazy(ep)?,
279            tokens,
280            bearer,
281        })
282    }
283
284    /// Wraps a channel you built yourself (for example with custom TLS or a proxy).
285    ///
286    /// Nothing is checked. `bearer` has the same meaning as in [`FinamApi::connect`].
287    pub fn from_channel(channel: Channel, tokens: Arc<TokenManager>, bearer: bool) -> Self {
288        Self {
289            channel,
290            tokens,
291            bearer,
292        }
293    }
294
295    /// The token manager used by this client.
296    ///
297    /// Use it to read the account list ([`TokenManager::account_ids`]) or to ask for a renewal.
298    pub fn tokens(&self) -> &Arc<TokenManager> {
299        &self.tokens
300    }
301
302    fn svc(&self) -> Authorized {
303        InterceptedService::new(
304            self.channel.clone(),
305            AuthInterceptor {
306                tokens: Arc::clone(&self.tokens),
307                bearer: self.bearer,
308            },
309        )
310    }
311
312    /// A generic gRPC client over the same channel and token, for a call that is not made through a
313    /// generated client: the method is named by a path (`/package.Service/Method`) and the messages are
314    /// whatever codec you give it. With [`FILE_DESCRIPTOR_SET`](crate::FILE_DESCRIPTOR_SET) it calls any
315    /// method of the API by name.
316    pub fn grpc(&self) -> tonic::client::Grpc<Authorized> {
317        tonic::client::Grpc::new(self.svc())
318    }
319
320    /// `OrdersService`: place, cancel and read orders, order and trade streams.
321    pub fn orders(&self) -> OrdersServiceClient<Authorized> {
322        OrdersServiceClient::new(self.svc())
323    }
324
325    /// `AccountsService`: account state, own trades, account stream.
326    pub fn accounts(&self) -> AccountsServiceClient<Authorized> {
327        AccountsServiceClient::new(self.svc())
328    }
329
330    /// `MarketDataService`: quotes, order books, trades, bars and their streams.
331    pub fn market_data(&self) -> MarketDataServiceClient<Authorized> {
332        MarketDataServiceClient::new(self.svc())
333    }
334
335    /// `AssetsService`: instrument directory, schedules, exchange clock.
336    pub fn assets(&self) -> AssetsServiceClient<Authorized> {
337        AssetsServiceClient::new(self.svc())
338    }
339
340    /// `UsageMetricsService`: how much of the request quota is left.
341    pub fn usage_metrics(&self) -> UsageMetricsServiceClient<Authorized> {
342        UsageMetricsServiceClient::new(self.svc())
343    }
344
345    /// `AuthService` with the current JWT attached. `Auth` and `SubscribeJwtRenewal` are driven by the
346    /// [`TokenManager`]; `TokenDetails` is reachable here.
347    pub fn auth(&self) -> AuthServiceClient<Authorized> {
348        AuthServiceClient::new(self.svc())
349    }
350
351    /// `ReportsService`: account reports.
352    pub fn reports(&self) -> ReportsServiceClient<Authorized> {
353        ReportsServiceClient::new(self.svc())
354    }
355
356    /// `CorporateActionsService`: splits, dividends and bond events.
357    pub fn corporate_actions(&self) -> CorporateActionsServiceClient<Authorized> {
358        CorporateActionsServiceClient::new(self.svc())
359    }
360}
361
362#[cfg(test)]
363mod tests {
364    use std::sync::Arc;
365
366    use async_trait::async_trait;
367
368    use super::*;
369    use crate::auth::{TokenApi, TokenInfo};
370    use crate::error::FinamError;
371
372    struct Fixed;
373
374    #[async_trait]
375    impl TokenApi for Fixed {
376        async fn auth(&self, _s: &str, _a: &str) -> Result<String, FinamError> {
377            Ok("the-jwt".into())
378        }
379        async fn token_details(&self, _t: &str) -> Result<TokenInfo, FinamError> {
380            Ok(TokenInfo::default())
381        }
382    }
383
384    #[tokio::test]
385    async fn the_jwt_header_is_sensitive_and_is_sent_bare_or_with_bearer() {
386        let tokens = crate::auth::TokenManager::start(Arc::new(Fixed), "secret", "")
387            .await
388            .unwrap();
389        for (bearer, want) in [(false, "the-jwt"), (true, "Bearer the-jwt")] {
390            let mut interceptor = AuthInterceptor {
391                tokens: Arc::clone(&tokens),
392                bearer,
393            };
394            let req = interceptor.call(Request::new(())).unwrap();
395            let value = req.metadata().get("authorization").unwrap();
396            assert_eq!(value, want);
397            // The value does not show in `Debug` output.
398            assert!(value.is_sensitive());
399            assert!(!format!("{req:?}").contains("the-jwt"), "{req:?}");
400        }
401    }
402}
brokapi_finam_protocol/client.rs

brokapi_finam_protocol/
client.rs
