[proxmox-backup.git] / src / server / h2service.rs

use anyhow::{Error};

use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use futures::*;
use hyper::{Body, Request, Response, StatusCode};

use proxmox::api::{ApiResponseFuture, HttpError, Router, RpcEnvironment};
use proxmox::http_err;

use proxmox_rest_server::normalize_uri_path;
use proxmox_rest_server::formatter::*;

use crate::server::WorkerTask;

/// Hyper Service implementation to handle stateful H2 connections.
///
/// We use this kind of service to handle backup protocol
/// connections. State is stored inside the generic ``rpcenv``. Logs
/// goes into the ``WorkerTask`` log.
pub struct H2Service<E> {
    router: &'static Router,
    rpcenv: E,
    worker: Arc<WorkerTask>,
    debug: bool,
}

impl <E: RpcEnvironment + Clone> H2Service<E> {

    pub fn new(rpcenv: E, worker: Arc<WorkerTask>, router: &'static Router, debug: bool) -> Self {
        Self { rpcenv, worker, router, debug }
    }

    pub fn debug<S: AsRef<str>>(&self, msg: S) {
        if self.debug { self.worker.log(msg); }
    }

    fn handle_request(&self, req: Request<Body>) -> ApiResponseFuture {

        let (parts, body) = req.into_parts();

        let method = parts.method.clone();

        let (path, components) = match normalize_uri_path(parts.uri.path()) {
            Ok((p,c)) => (p, c),
            Err(err) => return future::err(http_err!(BAD_REQUEST, "{}", err)).boxed(),
        };

        self.debug(format!("{} {}", method, path));

        let mut uri_param = HashMap::new();

        let formatter = &JSON_FORMATTER;

        match self.router.find_method(&components, method, &mut uri_param) {
            None => {
                let err = http_err!(NOT_FOUND, "Path '{}' not found.", path);
                future::ok((formatter.format_error)(err)).boxed()
            }
            Some(api_method) => {
                proxmox_rest_server::handle_api_request(
                    self.rpcenv.clone(), api_method, formatter, parts, body, uri_param).boxed()
            }
        }
    }

    fn log_response(worker: Arc<WorkerTask>, method: hyper::Method, path: &str, resp: &Response<Body>) {

        let status = resp.status();

        if !status.is_success() {
            let reason = status.canonical_reason().unwrap_or("unknown reason");

            let mut message = "request failed";
            if let Some(data) = resp.extensions().get::<ErrorMessageExtension>() {
                message = &data.0;
            }

            worker.log(format!("{} {}: {} {}: {}", method.as_str(), path, status.as_str(), reason, message));
        }
    }
}

impl <E: RpcEnvironment + Clone> tower_service::Service<Request<Body>> for H2Service<E> {
    type Response = Response<Body>;
    type Error = Error;
    #[allow(clippy::type_complexity)]
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Request<Body>) -> Self::Future {
        let path = req.uri().path().to_owned();
        let method = req.method().clone();
        let worker = self.worker.clone();

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