1use crate::billing::app_store_client::AppStoreClient;
2use crate::billing::billing_service::*;
3use crate::billing::google_play_client::GooglePlayClient;
4use crate::billing::stripe_client::StripeClient;
5use crate::config::Config;
6use crate::document_service::DocumentService;
7use crate::utils::get_build_info;
8use crate::{ServerError, ServerState, handle_version_header, router_service, verify_auth};
9use lazy_static::lazy_static;
10use lb_rs::model::api::{ErrorWrapper, Request, RequestWrapper, *};
11use lb_rs::model::errors::{LbErrKind, SignError};
12use lb_rs::model::wire::WireFormat;
13use prometheus::{
14 CounterVec, HistogramVec, TextEncoder, register_counter_vec, register_histogram_vec,
15};
16use serde::Serialize;
17use serde::de::DeserializeOwned;
18use std::collections::HashMap;
19use std::sync::Arc;
20use tracing::*;
21use warp::http::{HeaderValue, Method, StatusCode};
22use warp::hyper::body::Bytes;
23use warp::{Filter, Rejection, reject};
24
25lazy_static! {
26 pub static ref HTTP_REQUEST_DURATION_HISTOGRAM: HistogramVec = register_histogram_vec!(
27 "lockbook_server_request_duration_seconds",
28 "Lockbook server's HTTP request duration in seconds",
29 &["request"]
30 )
31 .unwrap();
32 pub static ref CORE_VERSION_COUNTER: CounterVec = register_counter_vec!(
33 "lockbook_server_core_version",
34 "Core version request attempts",
35 &["request"]
36 )
37 .unwrap();
38 pub static ref CLIENT_OS_COUNTER: CounterVec = register_counter_vec!(
39 "lockbook_server_client_os",
40 "Client operating system request attempts",
41 &["os"]
42 )
43 .unwrap();
44 pub static ref CLIENT_TYPE_COUNTER: CounterVec = register_counter_vec!(
45 "lockbook_server_client_type",
46 "Client type request attempts",
47 &["client_type"]
48 )
49 .unwrap();
50}
51
52pub fn normalize_os(os: Option<&str>) -> &'static str {
53 match os {
54 Some("windows") => "windows",
55 Some("linux") => "linux",
56 Some("macOS") => "macOS",
57 Some("iOS") => "iOS",
58 Some("android") => "android",
59 Some("unknown") => "unknown",
60 None => "not-present",
61 Some(_) => "other",
62 }
63}
64
65pub fn normalize_client_type(client_type: Option<&str>) -> &'static str {
66 match client_type {
67 Some("cli") => "cli",
68 Some("ui") => "ui",
69 Some("unknown") => "unknown",
70 None => "not-present",
71 Some(_) => "other",
72 }
73}
74
75#[macro_export]
76macro_rules! core_req {
77 ($Req: ty, $handler: path, $state: ident) => {{
78 use lb_rs::model::api::{ErrorWrapper, Request};
79 use lb_rs::model::file_metadata::Owner;
80 use lb_rs::model::wire::{CLIENT_HEADER, OS_HEADER, WIRE_FORMAT_HEADER, WireFormat};
81 use std::net::SocketAddr;
82 use tracing::*;
83 use $crate::router_service::{self, build_response, deserialize_and_check, method};
84 use $crate::{RequestContext, ServerError};
85
86 let cloned_state = $state.clone();
87
88 method(<$Req>::METHOD)
89 .and(warp::path(&<$Req>::ROUTE[1..]))
90 .and(warp::any().map(move || cloned_state.clone()))
91 .and(warp::any().map(|| lb_rs::model::clock::get_time())) .and(warp::body::bytes())
93 .and(warp::header::optional::<String>("Accept-Version"))
94 .and(warp::header::optional::<String>(WIRE_FORMAT_HEADER))
95 .and(warp::header::optional::<String>(OS_HEADER))
96 .and(warp::header::optional::<String>(CLIENT_HEADER))
97 .and(warp::filters::addr::remote())
98 .then(
99 |state: Arc<ServerState<S, A, G, D>>,
100 request_time: lb_rs::model::clock::Timestamp,
101 request: Bytes,
102 version: Option<String>,
103 wire_format_header: Option<String>,
104 os: Option<String>,
105 client_type: Option<String>,
106 ip: Option<SocketAddr>| {
107 if ip.is_none() {
108 tracing::error!("ip not present in request");
109 }
110 let wire_format = WireFormat::from_header(wire_format_header.as_deref());
111 router_service::CLIENT_OS_COUNTER
112 .with_label_values(&[router_service::normalize_os(os.as_deref())])
113 .inc();
114 router_service::CLIENT_TYPE_COUNTER
115 .with_label_values(&[router_service::normalize_client_type(
116 client_type.as_deref(),
117 )])
118 .inc();
119 let span1 = span!(
120 Level::INFO,
121 "matched_request",
122 http.method = &<$Req>::METHOD.as_str(),
123 http.url = &<$Req>::ROUTE,
124 http.remote_ip = ip
125 .map(|ip| ip.to_string())
126 .unwrap_or_else(|| String::from("unsupported")),
127 core_version = version
128 .clone()
129 .unwrap_or_else(|| String::from("not-present")),
130 wire_format = wire_format.as_str(),
131 os = os.unwrap_or_else(|| String::from("not-present")),
132 client_type = client_type.unwrap_or_else(|| String::from("not-present")),
133 );
134
135 async move {
136 let state = state.as_ref();
137 let timer = router_service::HTTP_REQUEST_DURATION_HISTOGRAM
138 .with_label_values(&[<$Req>::ROUTE])
139 .start_timer();
140
141 let request: RequestWrapper<$Req> = match deserialize_and_check(
142 &state.config,
143 request,
144 version,
145 wire_format,
146 request_time,
147 ) {
148 Ok(req) => req,
149 Err(err) => {
150 warn!("request failed to parse: {:?}", err);
151 let (body, status) = match wire_format.serialize::<Result<
152 RequestWrapper<$Req>,
153 _,
154 >>(
155 &Err(err)
156 ) {
157 Ok(body) => (body, warp::http::StatusCode::BAD_REQUEST),
158 Err(e) => {
159 error!(
160 "parse-error response serialize failed in {:?}: {e}",
161 wire_format
162 );
163 let fallback = wire_format
164 .serialize::<Result<(), ErrorWrapper<()>>>(&Err(
165 ErrorWrapper::InternalError,
166 ))
167 .unwrap_or_default();
168 (fallback, warp::http::StatusCode::INTERNAL_SERVER_ERROR)
169 }
170 };
171 return build_response(body, status);
172 }
173 };
174
175 let req_pk = request.signed_request.public_key;
176 let username = {
177 let db = state.index_db.lock().await;
178 match db
179 .accounts
180 .get()
181 .get(&Owner(req_pk))
182 .map(|account| account.username.clone())
183 {
184 Some(username) => username,
185 None => "~unknown~".to_string(),
186 }
187 };
188 let req_pk = base64::encode(req_pk.serialize_compressed());
189
190 let span2 = span!(
191 Level::INFO,
192 "verified_request_signature",
193 username = username.as_str(),
194 public_key = req_pk.as_str()
195 );
196 if ip.is_none() {
197 tracing::error!("ip not present in request");
198 }
199 let rc: RequestContext<$Req> = RequestContext {
200 request: request.signed_request.timestamped_value.value,
201 public_key: request.signed_request.public_key,
202 ip,
203 };
204
205 async move {
206 let mut status;
207 let mut level = tracing::Level::INFO;
208 let to_serialize = match $handler(state, rc).await {
209 Ok(response) => {
210 status = warp::http::StatusCode::OK;
211 Ok(response)
212 }
213 Err(ServerError::ClientError(e)) => {
214 status = warp::http::StatusCode::BAD_REQUEST;
215 level = tracing::Level::WARN;
216 Err(ErrorWrapper::Endpoint(e))
217 }
218 Err(ServerError::InternalError(e)) => {
219 status = warp::http::StatusCode::INTERNAL_SERVER_ERROR;
220 level = tracing::Level::ERROR;
221 tracing::error!("internal: {e}");
222 Err(ErrorWrapper::InternalError)
223 }
224 };
225 let body = match wire_format.serialize(&to_serialize) {
226 Ok(body) => body,
227 Err(e) => {
228 error!("response serialize failed in {:?}: {e}", wire_format);
229 status = warp::http::StatusCode::INTERNAL_SERVER_ERROR;
230 level = tracing::Level::ERROR;
231 wire_format
232 .serialize::<Result<(), ErrorWrapper<()>>>(&Err(
233 ErrorWrapper::InternalError,
234 ))
235 .unwrap_or_default()
236 }
237 };
238 let response = build_response(body, status);
239 let log = format!("{status} {} {username}", &<$Req>::ROUTE);
240 let latency = timer.stop_and_record();
241 match level {
242 tracing::Level::INFO => {
243 tracing::info!(
244 http.latency = latency,
245 http.status = status.as_u16(),
246 "{log}"
247 );
248 }
249 tracing::Level::WARN => {
250 tracing::warn!(
251 http.latency = latency,
252 http.status = status.as_u16(),
253 "{log}"
254 );
255 }
256 tracing::Level::ERROR => {
257 tracing::error!(
258 http.latency = latency,
259 http.status = status.as_u16(),
260 "{log}"
261 );
262 }
263 _ => {
264 tracing::debug!(
265 http.latency = latency,
266 http.status = status.as_u16(),
267 "{log}"
268 );
269 }
270 }
271
272 response
273 }
274 .instrument(span2)
275 .await
276 }
277 .instrument(span1)
278 },
279 )
280 }};
281}
282
283const RESPONSE_STREAM_CHUNK_BYTES: usize = 4 * 1024 * 1024;
284
285const RESPONSE_STREAM_THRESHOLD: usize = 1024 * 1024 * 1024;
286
287pub fn build_response(
288 body: Vec<u8>, status: StatusCode,
289) -> warp::http::Response<warp::hyper::Body> {
290 let response_body = if body.len() < RESPONSE_STREAM_THRESHOLD {
291 warp::hyper::Body::from(body)
292 } else {
293 let mut buf = Bytes::from(body);
294 let mut chunks: Vec<Result<Bytes, std::io::Error>> =
295 Vec::with_capacity(buf.len().div_ceil(RESPONSE_STREAM_CHUNK_BYTES));
296 while !buf.is_empty() {
297 let n = buf.len().min(RESPONSE_STREAM_CHUNK_BYTES);
298 chunks.push(Ok(buf.split_to(n)));
299 }
300 warp::hyper::Body::wrap_stream(futures::stream::iter(chunks))
301 };
302 let mut response = warp::http::Response::new(response_body);
303 *response.status_mut() = status;
304 response
305}
306
307pub fn core_routes<S, A, G, D>(
308 server_state: &Arc<ServerState<S, A, G, D>>,
309) -> impl Filter<Extract = (impl warp::Reply + use<S, A, G, D>,), Error = Rejection>
310+ Clone
311+ use<S, A, G, D>
312where
313 S: StripeClient,
314 A: AppStoreClient,
315 G: GooglePlayClient,
316 D: DocumentService,
317{
318 core_req!(NewAccountRequestV2, ServerState::new_account_v2, server_state)
319 .or(core_req!(ChangeDocRequestV2, ServerState::change_doc_v2, server_state))
320 .or(core_req!(UpsertRequestV2, ServerState::upsert_file_metadata_v2, server_state))
321 .or(core_req!(GetDocRequest, ServerState::get_document, server_state))
322 .or(core_req!(GetPublicKeyRequest, ServerState::get_public_key, server_state))
323 .or(core_req!(GetUsernameRequest, ServerState::get_username, server_state))
324 .or(core_req!(GetUsageRequest, ServerState::get_usage, server_state))
325 .or(core_req!(GetFileIdsRequest, ServerState::get_file_ids, server_state))
326 .or(core_req!(GetUpdatesRequestV2, ServerState::get_updates_v2, server_state))
327 .or(core_req!(
328 UpgradeAccountGooglePlayRequest,
329 ServerState::upgrade_account_google_play,
330 server_state
331 ))
332 .or(core_req!(
333 UpgradeAccountStripeRequest,
334 ServerState::upgrade_account_stripe,
335 server_state
336 ))
337 .or(core_req!(
338 UpgradeAccountAppStoreRequest,
339 ServerState::upgrade_account_app_store,
340 server_state
341 ))
342 .or(core_req!(CancelSubscriptionRequest, ServerState::cancel_subscription, server_state))
343 .or(core_req!(GetSubscriptionInfoRequest, ServerState::get_subscription_info, server_state))
344 .or(core_req!(
345 GetSubscriptionInfoRequestV2,
346 ServerState::get_subscription_info_v2,
347 server_state
348 ))
349 .or(core_req!(DeleteAccountRequest, ServerState::delete_account, server_state))
350 .or(core_req!(UpsertDebugInfoRequest, ServerState::upsert_debug_info, server_state))
351 .or(core_req!(
352 AdminDisappearAccountRequest,
353 ServerState::admin_disappear_account,
354 server_state
355 ))
356 .or(core_req!(AdminDisappearFileRequest, ServerState::admin_disappear_file, server_state))
357 .or(core_req!(AdminListUsersRequest, ServerState::admin_list_users, server_state))
358 .or(core_req!(
359 AdminGetAccountInfoRequest,
360 ServerState::admin_get_account_info,
361 server_state
362 ))
363 .or(core_req!(
364 AdminValidateAccountRequest,
365 ServerState::admin_validate_account,
366 server_state
367 ))
368 .or(core_req!(AdminValidateServerRequest, ServerState::admin_validate_server, server_state))
369 .or(core_req!(AdminFileInfoRequest, ServerState::admin_file_info, server_state))
370 .or(core_req!(AdminRebuildIndexRequest, ServerState::admin_rebuild_index, server_state))
371 .or(core_req!(AdminSetUserTierRequest, ServerState::admin_set_user_tier, server_state))
372}
373
374pub fn build_info() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
375 warp::get()
376 .and(warp::path(&GetBuildInfoRequest::ROUTE[1..]))
377 .map(|| {
378 let span = span!(
379 Level::INFO,
380 "matched_request",
381 method = &GetBuildInfoRequest::METHOD.as_str(),
382 route = &GetBuildInfoRequest::ROUTE,
383 );
384 let _enter = span.enter();
385 let timer = router_service::HTTP_REQUEST_DURATION_HISTOGRAM
386 .with_label_values(&[GetBuildInfoRequest::ROUTE])
387 .start_timer();
388 let resp = get_build_info();
389 info!("request processed successfully");
390 let resp = warp::reply::json(&resp);
391 timer.observe_duration();
392 resp
393 })
394}
395
396pub fn get_metrics() -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone
397{
398 warp::get().and(warp::path("metrics")).map(|| {
399 let span = span!(Level::INFO, "matched_request", method = "GET", route = "/metrics",);
400 let _enter = span.enter();
401 match TextEncoder::new().encode_to_string(prometheus::gather().as_slice()) {
402 Ok(metrics) => {
403 info!("request processed successfully");
404 metrics
405 }
406 Err(err) => {
407 error!("Error preparing response for prometheus: {:?}", err);
408 String::new()
409 }
410 }
411 })
412}
413
414static STRIPE_WEBHOOK_ROUTE: &str = "stripe-webhooks";
415
416pub fn stripe_webhooks<S, A, G, D>(
417 server_state: &Arc<ServerState<S, A, G, D>>,
418) -> impl Filter<Extract = (impl warp::Reply + use<S, A, G, D>,), Error = warp::Rejection>
419+ Clone
420+ use<S, A, G, D>
421where
422 S: StripeClient,
423 A: AppStoreClient,
424 G: GooglePlayClient,
425 D: DocumentService,
426{
427 let cloned_state = server_state.clone();
428
429 warp::post()
430 .and(warp::path(STRIPE_WEBHOOK_ROUTE))
431 .and(warp::any().map(move || cloned_state.clone()))
432 .and(warp::body::bytes())
433 .and(warp::header::header("Stripe-Signature"))
434 .then(
435 |state: Arc<ServerState<S, A, G, D>>, request: Bytes, stripe_sig: HeaderValue| async move {
436 let span = span!(
437 Level::INFO,
438 "matched_request",
439 method = "POST",
440 route = format!("/{STRIPE_WEBHOOK_ROUTE}").as_str()
441 );
442 let _enter = span.enter();
443 info!("webhook routed");
444 let response = span
445 .in_scope(|| ServerState::stripe_webhooks(&state, request, stripe_sig))
446 .await;
447
448 match response {
449 Ok(_) => warp::reply::with_status("".to_string(), StatusCode::OK),
450 Err(e) => {
451 error!("{:?}", e);
452
453 let status_code = match e {
454 ServerError::ClientError(StripeWebhookError::VerificationError(_))
455 | ServerError::ClientError(StripeWebhookError::InvalidBody(_))
456 | ServerError::ClientError(StripeWebhookError::InvalidHeader(_))
457 | ServerError::ClientError(StripeWebhookError::ParseError(_)) => {
458 StatusCode::BAD_REQUEST
459 }
460 ServerError::InternalError(_) => StatusCode::INTERNAL_SERVER_ERROR,
461 };
462
463 warp::reply::with_status("".to_string(), status_code)
464 }
465 }
466 },
467 )
468}
469
470static PLAY_WEBHOOK_ROUTE: &str = "google_play_notification_webhook";
471
472pub fn google_play_notification_webhooks<S, A, G, D>(
473 server_state: &Arc<ServerState<S, A, G, D>>,
474) -> impl Filter<Extract = (impl warp::Reply + use<S, A, G, D>,), Error = warp::Rejection>
475+ Clone
476+ use<S, A, G, D>
477where
478 S: StripeClient,
479 A: AppStoreClient,
480 G: GooglePlayClient,
481 D: DocumentService,
482{
483 let cloned_state = server_state.clone();
484
485 warp::post()
486 .and(warp::path(PLAY_WEBHOOK_ROUTE))
487 .and(warp::any().map(move || cloned_state.clone()))
488 .and(warp::body::bytes())
489 .and(warp::query::query::<HashMap<String, String>>())
490 .then(
491 |state: Arc<ServerState<S, A, G, D>>,
492 request: Bytes,
493 query_parameters: HashMap<String, String>| async move {
494 let span = span!(
495 Level::INFO,
496 "matched_request",
497 method = "POST",
498 route = format!("/{PLAY_WEBHOOK_ROUTE}").as_str()
499 );
500 let _enter = span.enter();
501 info!("webhook routed");
502 let response = span
503 .in_scope(|| state.google_play_notification_webhooks(request, query_parameters))
504 .await;
505 match response {
506 Ok(_) => warp::reply::with_status("".to_string(), StatusCode::OK),
507 Err(e) => {
508 error!("{:?}", e);
509
510 let status_code = match e {
511 ServerError::ClientError(GooglePlayWebhookError::InvalidToken)
512 | ServerError::ClientError(
513 GooglePlayWebhookError::CannotRetrieveData,
514 )
515 | ServerError::ClientError(
516 GooglePlayWebhookError::CannotDecodePubSubData(_),
517 ) => StatusCode::BAD_REQUEST,
518 ServerError::ClientError(
519 GooglePlayWebhookError::CannotRetrieveUserInfo,
520 )
521 | ServerError::ClientError(
522 GooglePlayWebhookError::CannotRetrievePublicKey,
523 )
524 | ServerError::ClientError(GooglePlayWebhookError::CannotParseTime)
525 | ServerError::InternalError(_) => StatusCode::INTERNAL_SERVER_ERROR,
526 };
527
528 warp::reply::with_status("".to_string(), status_code)
529 }
530 }
531 },
532 )
533}
534
535static APP_STORE_WEBHOOK_ROUTE: &str = "app_store_notification_webhook";
536pub fn app_store_notification_webhooks<S, A, G, D>(
537 server_state: &Arc<ServerState<S, A, G, D>>,
538) -> impl Filter<Extract = (impl warp::Reply + use<S, A, G, D>,), Error = warp::Rejection>
539+ Clone
540+ use<S, A, G, D>
541where
542 S: StripeClient,
543 A: AppStoreClient,
544 G: GooglePlayClient,
545 D: DocumentService,
546{
547 let cloned_state = server_state.clone();
548
549 warp::post()
550 .and(warp::path(APP_STORE_WEBHOOK_ROUTE))
551 .and(warp::any().map(move || cloned_state.clone()))
552 .and(warp::body::bytes())
553 .then(|state: Arc<ServerState<S, A, G, D>>, body: Bytes| async move {
554 let span = span!(
555 Level::INFO,
556 "matched_request",
557 method = "POST",
558 route = format!("/{APP_STORE_WEBHOOK_ROUTE}").as_str()
559 );
560 let _enter = span.enter();
561 info!("webhook routed");
562 let response = span
563 .in_scope(|| state.app_store_notification_webhook(body))
564 .await;
565
566 match response {
567 Ok(_) => warp::reply::with_status("".to_string(), StatusCode::OK),
568 Err(e) => {
569 error!("{:?}", e);
570
571 let status_code = match e {
572 ServerError::ClientError(AppStoreNotificationError::InvalidJWS) => {
573 StatusCode::BAD_REQUEST
574 }
575 ServerError::InternalError(_) => StatusCode::INTERNAL_SERVER_ERROR,
576 };
577
578 warp::reply::with_status("".to_string(), status_code)
579 }
580 }
581 })
582}
583
584pub fn method(name: Method) -> impl Filter<Extract = (), Error = Rejection> + Clone {
585 warp::method()
586 .and(warp::any().map(move || name.clone()))
587 .and_then(|request: Method, intention: Method| async move {
588 if request == intention { Ok(()) } else { Err(reject::not_found()) }
589 })
590 .untuple_one()
591}
592
593pub fn deserialize_and_check<Req>(
594 config: &Config, request: Bytes, version: Option<String>, wire_format: WireFormat,
595 request_time: lb_rs::model::clock::Timestamp,
596) -> Result<RequestWrapper<Req>, ErrorWrapper<Req::Error>>
597where
598 Req: Request + DeserializeOwned + Serialize,
599{
600 handle_version_header::<Req>(config, &version)?;
601
602 let request = wire_format.deserialize(request.as_ref()).map_err(|err| {
603 warn!("Request parsing failure ({:?}): {}", wire_format, err);
604 ErrorWrapper::<Req::Error>::BadRequest
605 })?;
606
607 verify_auth(config, &request, request_time).map_err(|err| match err.kind {
608 LbErrKind::Sign(SignError::SignatureExpired(_))
609 | LbErrKind::Sign(SignError::SignatureInTheFuture(_)) => {
610 warn!("expired auth");
611 ErrorWrapper::<Req::Error>::ExpiredAuth
612 }
613 _ => {
614 warn!("invalid auth");
615 ErrorWrapper::<Req::Error>::InvalidAuth
616 }
617 })?;
618
619 Ok(request)
620}