[proxmox.git] / proxmox-rest-server / src / 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_router::http_err;
use proxmox_router::{ApiResponseFuture, HttpError, Router, RpcEnvironment};

use crate::formatter::*;
use crate::{normalize_uri_path, 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_message(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) => crate::rest::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_message(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
2bda552b	1	use anyhow::Error;
319e4255 DM	2
319e4255 DM	3	use std::collections::HashMap;
2a16daaf	4	use std::pin::Pin;
319e4255	5	use std::sync::Arc;
3a1a7cbd	6	use std::task::{Context, Poll};
319e4255 DM	7
	8	use futures::*;
	9	use hyper::{Body, Request, Response, StatusCode};
	10
09046671	11	use proxmox_router::http_err;
2bda552b	12	use proxmox_router::{ApiResponseFuture, HttpError, Router, RpcEnvironment};
4da0705c	13
85ec987a	14	use crate::formatter::*;
2bda552b	15	use crate::{normalize_uri_path, WorkerTask};
dc28aa1a	16
319e4255 DM	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
2bda552b	29	impl<E: RpcEnvironment + Clone> H2Service<E> {
319e4255	30	pub fn new(rpcenv: E, worker: Arc<WorkerTask>, router: &'static Router, debug: bool) -> Self {
2bda552b TL	31	Self {
	32	rpcenv,
	33	worker,
	34	router,
	35	debug,
	36	}
319e4255 DM	37	}
	38
	39	pub fn debug<S: AsRef<str>>(&self, msg: S) {
2bda552b TL	40	if self.debug {
	41	self.worker.log_message(msg);
	42	}
319e4255 DM	43	}
319e4255 DM	44
3e2c8347	45	fn handle_request(&self, req: Request<Body>) -> ApiResponseFuture {
319e4255 DM	46	let (parts, body) = req.into_parts();
	47
	48	let method = parts.method.clone();
	49
cc674416	50	let (path, components) = match normalize_uri_path(parts.uri.path()) {
2bda552b	51	Ok((p, c)) => (p, c),
df9109d4	52	Err(err) => return future::err(http_err!(BAD_REQUEST, "{}", err)).boxed(),
319e4255 DM	53	};
	54
	55	self.debug(format!("{} {}", method, path));
	56
	57	let mut uri_param = HashMap::new();
	58
962553d2	59	let formatter = JSON_FORMATTER;
319e4255 DM	60
319e4255 DM	61	match self.router.find_method(&components, method, &mut uri_param) {
3de9361d	62	None => {
df9109d4	63	let err = http_err!(NOT_FOUND, "Path '{}' not found.", path);
962553d2	64	future::ok(formatter.format_error(err)).boxed()
319e4255	65	}
2bda552b TL	66	Some(api_method) => crate::rest::handle_api_request(
	67	self.rpcenv.clone(),
	68	api_method,
	69	formatter,
	70	parts,
	71	body,
	72	uri_param,
	73	)
	74	.boxed(),
319e4255 DM	75	}
	76	}
	77
2bda552b TL	78	fn log_response(
	79	worker: Arc<WorkerTask>,
	80	method: hyper::Method,
	81	path: &str,
	82	resp: &Response<Body>,
	83	) {
319e4255 DM	84	let status = resp.status();
	85
	86	if !status.is_success() {
	87	let reason = status.canonical_reason().unwrap_or("unknown reason");
	88
	89	let mut message = "request failed";
	90	if let Some(data) = resp.extensions().get::<ErrorMessageExtension>() {
	91	message = &data.0;
	92	}
	93
020c8e69 DM	94	worker.log_message(format!(
	95	"{} {}: {} {}: {}",
	96	method.as_str(),
	97	path,
	98	status.as_str(),
	99	reason,
	100	message
	101	));
319e4255 DM	102	}
	103	}
	104	}
	105
2bda552b	106	impl<E: RpcEnvironment + Clone> tower_service::Service<Request<Body>> for H2Service<E> {
3a1a7cbd	107	type Response = Response<Body>;
319e4255	108	type Error = Error;
2a16daaf FG	109	#[allow(clippy::type_complexity)]
2a16daaf FG	110	type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
319e4255	111
3a1a7cbd WB	112	fn poll_ready(&mut self, _cx: &mut Context) -> Poll<Result<(), Self::Error>> {
	113	Poll::Ready(Ok(()))
	114	}
	115
	116	fn call(&mut self, req: Request<Body>) -> Self::Future {
319e4255 DM	117	let path = req.uri().path().to_owned();
	118	let method = req.method().clone();
	119	let worker = self.worker.clone();
	120
28f3b0df	121	self.handle_request(req)
3a1a7cbd	122	.map(move \|result\| match result {
319e4255 DM	123	Ok(res) => {
	124	Self::log_response(worker, method, &path, &res);
	125	Ok::<_, Error>(res)
	126	}
	127	Err(err) => {
2bda552b	128	if let Some(apierr) = err.downcast_ref::<HttpError>() {
319e4255	129	let mut resp = Response::new(Body::from(apierr.message.clone()));
2bda552b TL	130	resp.extensions_mut()
2bda552b TL	131	.insert(ErrorMessageExtension(apierr.message.clone()));
319e4255 DM	132	*resp.status_mut() = apierr.code;
	133	Self::log_response(worker, method, &path, &resp);
	134	Ok(resp)
	135	} else {
	136	let mut resp = Response::new(Body::from(err.to_string()));
2bda552b TL	137	resp.extensions_mut()
2bda552b TL	138	.insert(ErrorMessageExtension(err.to_string()));
319e4255 DM	139	*resp.status_mut() = StatusCode::BAD_REQUEST;
	140	Self::log_response(worker, method, &path, &resp);
	141	Ok(resp)
	142	}
	143	}
3a1a7cbd WB	144	})
3a1a7cbd WB	145	.boxed()
319e4255 DM	146	}
319e4255 DM	147	}