proxmox-rest-server/src/h2service.rs

   1 use anyhow::{Error};
   2
   3 use std::collections::HashMap;
   4 use std::pin::Pin;
   5 use std::sync::Arc;
   6 use std::task::{Context, Poll};
   7
   8 use futures::*;
   9 use hyper::{Body, Request, Response, StatusCode};
  10
  11 use proxmox_router::{ApiResponseFuture, HttpError, Router, RpcEnvironment};
  12 use proxmox_router::http_err;
  13
  14 use crate::{normalize_uri_path, WorkerTask};
  15 use crate::formatter::*;
  16
  17 /// Hyper Service implementation to handle stateful H2 connections.
  18 ///
  19 /// We use this kind of service to handle backup protocol
  20 /// connections. State is stored inside the generic ``rpcenv``. Logs
  21 /// goes into the ``WorkerTask`` log.
  22 pub struct H2Service<E> {
  23     router: &'static Router,
  24     rpcenv: E,
  25     worker: Arc<WorkerTask>,
  26     debug: bool,
  27 }
  28
  29 impl <E: RpcEnvironment + Clone> H2Service<E> {
  30
  31     pub fn new(rpcenv: E, worker: Arc<WorkerTask>, router: &'static Router, debug: bool) -> Self {
  32         Self { rpcenv, worker, router, debug }
  33     }
  34
  35     pub fn debug<S: AsRef<str>>(&self, msg: S) {
  36         if self.debug { self.worker.log_message(msg); }
  37     }
  38
  39     fn handle_request(&self, req: Request<Body>) -> ApiResponseFuture {
  40
  41         let (parts, body) = req.into_parts();
  42
  43         let method = parts.method.clone();
  44
  45         let (path, components) = match normalize_uri_path(parts.uri.path()) {
  46             Ok((p,c)) => (p, c),
  47             Err(err) => return future::err(http_err!(BAD_REQUEST, "{}", err)).boxed(),
  48         };
  49
  50         self.debug(format!("{} {}", method, path));
  51
  52         let mut uri_param = HashMap::new();
  53
  54         let formatter = JSON_FORMATTER;
  55
  56         match self.router.find_method(&components, method, &mut uri_param) {
  57             None => {
  58                 let err = http_err!(NOT_FOUND, "Path '{}' not found.", path);
  59                 future::ok(formatter.format_error(err)).boxed()
  60             }
  61             Some(api_method) => {
  62                 crate::rest::handle_api_request(
  63                     self.rpcenv.clone(), api_method, formatter, parts, body, uri_param).boxed()
  64             }
  65         }
  66     }
  67
  68     fn log_response(worker: Arc<WorkerTask>, method: hyper::Method, path: &str, resp: &Response<Body>) {
  69
  70         let status = resp.status();
  71
  72         if !status.is_success() {
  73             let reason = status.canonical_reason().unwrap_or("unknown reason");
  74
  75             let mut message = "request failed";
  76             if let Some(data) = resp.extensions().get::<ErrorMessageExtension>() {
  77                 message = &data.0;
  78             }
  79
  80             worker.log_message(format!(
  81                 "{} {}: {} {}: {}",
  82                 method.as_str(),
  83                 path,
  84                 status.as_str(),
  85                 reason,
  86                 message
  87             ));
  88         }
  89     }
  90 }
  91
  92 impl <E: RpcEnvironment + Clone> tower_service::Service<Request<Body>> for H2Service<E> {
  93     type Response = Response<Body>;
  94     type Error = Error;
  95     #[allow(clippy::type_complexity)]
  96     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
  97
  98     fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
  99         Poll::Ready(Ok(()))
 100     }
 101
 102     fn call(&mut self, req: Request<Body>) -> Self::Future {
 103         let path = req.uri().path().to_owned();
 104         let method = req.method().clone();
 105         let worker = self.worker.clone();
 106
 107         self.handle_request(req)
 108             .map(move |result| match result {
 109                 Ok(res) => {
 110                     Self::log_response(worker, method, &path, &res);
 111                     Ok::<_, Error>(res)
 112                 }
 113                 Err(err) => {
 114                      if let Some(apierr) = err.downcast_ref::<HttpError>() {
 115                         let mut resp = Response::new(Body::from(apierr.message.clone()));
 116                         resp.extensions_mut().insert(ErrorMessageExtension(apierr.message.clone()));
 117                         *resp.status_mut() = apierr.code;
 118                         Self::log_response(worker, method, &path, &resp);
 119                         Ok(resp)
 120                     } else {
 121                         let mut resp = Response::new(Body::from(err.to_string()));
 122                         resp.extensions_mut().insert(ErrorMessageExtension(err.to_string()));
 123                         *resp.status_mut() = StatusCode::BAD_REQUEST;
 124                         Self::log_response(worker, method, &path, &resp);
 125                         Ok(resp)
 126                     }
 127                 }
 128             })
 129             .boxed()
 130     }
 131 }