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