[proxmox-backup.git] / src / api_server.rs

use std::collections::HashMap;
use std::path::{PathBuf};
use std::sync::Arc;

use failure::*;
use serde_json::{json, Value};

use crate::json_schema::*;
use crate::api_info::*;


use futures::future::{self, Either};
//use tokio::prelude::*;
//use tokio::timer::Delay;
use tokio::fs::File;
use tokio_codec;
//use bytes::{BytesMut, BufMut};

//use hyper::body::Payload;
use hyper::http::request::Parts;
use hyper::{Body, Request, Method, Response, StatusCode};
use hyper::service::{Service, NewService};
use hyper::rt::{Future, Stream};
use hyper::header;


pub struct ApiConfig {
    basedir: PathBuf,
    router: &'static MethodInfo,
    aliases: HashMap<String, PathBuf>,
}

impl ApiConfig {

    pub fn new<B: Into<PathBuf>>(basedir: B, router: &'static MethodInfo) -> Self {
        Self {
            basedir: basedir.into(),
            router: router,
            aliases: HashMap::new(),
        }
    }

    pub fn find_method(&self, components: &[&str], method: Method) -> Option<&'static ApiMethod> {

        if let Some(info) = self.router.find_route(components) {
            println!("FOUND INFO");
            let opt_api_method = match method {
                Method::GET => &info.get,
                Method::PUT => &info.put,
                Method::POST => &info.post,
                Method::DELETE => &info.delete,
                _ => &None,
            };
            if let Some(api_method) = opt_api_method {
                return Some(&api_method);
            }
        }
        None
    }

    pub fn find_alias(&self, components: &[&str]) -> PathBuf {

        let mut prefix = String::new();
        let mut filename = self.basedir.clone();
        let comp_len = components.len();
        if comp_len >= 1 {
            prefix.push_str(components[0]);
            if let Some(subdir) = self.aliases.get(&prefix) {
                filename.push(subdir);
                for i in 1..comp_len { filename.push(components[i]) }
            }
        }
        filename
    }

    pub fn add_alias<S, P>(&mut self, alias: S, path: P)
        where S: Into<String>,
              P: Into<PathBuf>,
    {
        self.aliases.insert(alias.into(), path.into());
    }
}

type BoxFut = Box<Future<Item = Response<Body>, Error = failure::Error> + Send>;

macro_rules! error_response {
    ($status:ident, $msg:expr) => {{
        let mut resp = Response::new(Body::from($msg));
        *resp.status_mut() = StatusCode::$status;
        resp
    }}
}

macro_rules! http_error_future {
    ($status:ident, $msg:expr) => {{
        let resp = error_response!($status, $msg);
        return Box::new(future::ok(resp));
    }}
}

fn get_request_parameters_async<'a>(
    info: &'a ApiMethod,
    parts: Parts,
    req_body: Body,
) -> Box<Future<Item = Value, Error = failure::Error> + Send + 'a>
{
    let resp = req_body
        .map_err(|err| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("Promlems reading request body: {}", err))))
        .fold(Vec::new(), |mut acc, chunk| {
            if acc.len() + chunk.len() < 64*1024 { //fimxe: max request body size?
                acc.extend_from_slice(&*chunk);
                Ok(acc)
            }
            else { Err(Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("Request body too large")))) }
        })
        .and_then(move |body| {

            let bytes = String::from_utf8(body.to_vec())?; // why copy??

            println!("GOT BODY {:?}", bytes);

            let mut test_required = true;

            let mut params = json!({});

            if bytes.len() > 0 {
                params = parse_query_string(&bytes, &info.parameters, true)?;
                test_required = false;
            }

            if let Some(query_str) = parts.uri.query() {
                let query_params = parse_query_string(query_str, &info.parameters, test_required)?;

                for (k, v) in query_params.as_object().unwrap() {
                    params[k] = v.clone(); // fixme: why clone()??
                }
            }

            println!("GOT PARAMS {}", params);
            Ok(params)
        });

    Box::new(resp)
}

fn handle_sync_api_request<'a>(
    info: &'a ApiMethod,
    parts: Parts,
    req_body: Body,
) -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'a>
{
    let params = get_request_parameters_async(info, parts, req_body);

    let resp = params
        .and_then(move |params| {

            println!("GOT PARAMS {}", params);

            /*
            let when = Instant::now() + Duration::from_millis(3000);
            let task = Delay::new(when).then(|_| {
                println!("A LAZY TASK");
                ok(())
            });

            tokio::spawn(task);
             */

            let res = (info.handler)(params, info)?;

            Ok(res)

        }).then(|result| {
            match result {
                Ok(ref value) => {
                    let json_str = value.to_string();

                    Ok(Response::builder()
                       .status(StatusCode::OK)
                       .header(header::CONTENT_TYPE, "application/json")
                       .body(Body::from(json_str))?)
                }
                Err(err) => Ok(error_response!(BAD_REQUEST, err.to_string()))
            }
        });

    Box::new(resp)
}

fn simple_static_file_download(filename: PathBuf) ->  BoxFut {

    Box::new(File::open(filename)
        .map_err(|err| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File open failed: {}", err))))
        .and_then(|file| {
            let buf: Vec<u8> = Vec::new();
            tokio::io::read_to_end(file, buf)
                .map_err(|err| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File read failed: {}", err))))
                .and_then(|data| Ok(Response::new(data.1.into())))
        }))
}

fn chuncked_static_file_download(filename: PathBuf) ->  BoxFut {

    Box::new(File::open(filename)
        .map_err(|err| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File open failed: {}", err))))
        .and_then(|file| {
            let payload = tokio_codec::FramedRead::new(file, tokio_codec::BytesCodec::new()).
                map(|bytes| {
                    //sigh - howto avoid copy here? or the whole map() ??
                    hyper::Chunk::from(bytes.to_vec())
                });
            let body = Body::wrap_stream(payload);
            // fixme: set content type and other headers
            Ok(Response::builder()
               .status(StatusCode::OK)
               .body(body)
               .unwrap())
        }))
}

fn handle_static_file_download(filename: PathBuf) ->  BoxFut {

    let response = tokio::fs::metadata(filename.clone())
        .map_err(|err| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File access problems: {}", err))))
        .and_then(|metadata| {
            if metadata.len() < 1024*32 {
                Either::A(simple_static_file_download(filename))
            } else {
                Either::B(chuncked_static_file_download(filename))
             }
        });

    return Box::new(response);
}

pub fn handle_request(api: Arc<ApiConfig>, req: Request<Body>) -> BoxFut {

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

    let method = parts.method.clone();
    let path = parts.uri.path();

    // normalize path
    // do not allow ".", "..", or hidden files ".XXXX"
    // also remove empty path components

    let items = path.split('/');
    let mut path = String::new();
    let mut components = vec![];

    for name in items {
        if name.is_empty() { continue; }
        if name.starts_with(".") {
            http_error_future!(BAD_REQUEST, "Path contains illegal components.\n");
        }
        path.push('/');
        path.push_str(name);
        components.push(name);
    }

    let comp_len = components.len();

    println!("REQUEST {} {}", method, path);
    println!("COMPO {:?}", components);

    if comp_len >= 1 && components[0] == "api3" {
        println!("GOT API REQUEST");
        if comp_len >= 2 {
            let format = components[1];
            if format != "json" {
                http_error_future!(BAD_REQUEST, format!("Unsupported output format '{}'.", format))
            }

            if let Some(api_method) = api.find_method(&components[2..], method) {
                // fixme: handle auth
                return handle_sync_api_request(api_method, parts, body);
            }
        }
    } else {
        // not Auth for accessing files!

        let filename = api.find_alias(&components);
        return handle_static_file_download(filename);
    }

    http_error_future!(NOT_FOUND, "Path not found.")
    //Box::new(ok(Response::new(Body::from("RETURN WEB GUI\n"))))
}

pub struct RestServer {
    pub api_config: Arc<ApiConfig>,
}

impl RestServer {

    pub fn new(api_config: ApiConfig) -> Self {
        Self { api_config: Arc::new(api_config) }
    }
}

impl NewService for RestServer
{
    type ReqBody = Body;
    type ResBody = Body;
    type Error = hyper::Error;
    type InitError = hyper::Error;
    type Service = ApiService;
    type Future = Box<Future<Item = Self::Service, Error = Self::InitError> + Send>;
    fn new_service(&self) -> Self::Future {
        Box::new(future::ok(ApiService { api_config: self.api_config.clone() }))
    }
}

pub struct ApiService {
    pub api_config: Arc<ApiConfig>,
}

impl Service for ApiService {
    type ReqBody = Body;
    type ResBody = Body;
    type Error = hyper::Error;
    type Future = Box<Future<Item = Response<Body>, Error = Self::Error> + Send>;

    fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {

        Box::new(handle_request(self.api_config.clone(), req).then(|result| {
            match result {
                Ok(res) => Ok::<_, hyper::Error>(res),
                Err(err) => {
                    if let Some(apierr) = err.downcast_ref::<ApiError>() {
                        let mut resp = Response::new(Body::from(apierr.message.clone()));
                        *resp.status_mut() = apierr.code;
                        Ok(resp)
                    } else {
                        let mut resp = Response::new(Body::from(err.to_string()));
                        *resp.status_mut() = StatusCode::BAD_REQUEST;
                        Ok(resp)
                    }
                }
            }
        }))
    }
}
Commit	Line	Data
1a53be14 DM	1	use std::collections::HashMap;
1a53be14 DM	2	use std::path::{PathBuf};
9bc17e8d DM	3	use std::sync::Arc;
	4
	5	use failure::*;
	6	use serde_json::{json, Value};
	7
	8	use crate::json_schema::*;
1a53be14	9	use crate::api_info::*;
1a53be14	10
1a53be14	11
9bc17e8d DM	12	use futures::future::{self, Either};
	13	//use tokio::prelude::*;
	14	//use tokio::timer::Delay;
	15	use tokio::fs::File;
	16	use tokio_codec;
	17	//use bytes::{BytesMut, BufMut};
	18
	19	//use hyper::body::Payload;
	20	use hyper::http::request::Parts;
	21	use hyper::{Body, Request, Method, Response, StatusCode};
	22	use hyper::service::{Service, NewService};
	23	use hyper::rt::{Future, Stream};
	24	use hyper::header;
	25
	26
	27	pub struct ApiConfig {
1a53be14 DM	28	basedir: PathBuf,
	29	router: &'static MethodInfo,
	30	aliases: HashMap<String, PathBuf>,
	31	}
	32
9bc17e8d	33	impl ApiConfig {
1a53be14 DM	34
1a53be14 DM	35	pub fn new<B: Into<PathBuf>>(basedir: B, router: &'static MethodInfo) -> Self {
9bc17e8d	36	Self {
1a53be14 DM	37	basedir: basedir.into(),
	38	router: router,
	39	aliases: HashMap::new(),
	40	}
	41	}
	42
	43	pub fn find_method(&self, components: &[&str], method: Method) -> Option<&'static ApiMethod> {
	44
	45	if let Some(info) = self.router.find_route(components) {
	46	println!("FOUND INFO");
	47	let opt_api_method = match method {
	48	Method::GET => &info.get,
	49	Method::PUT => &info.put,
	50	Method::POST => &info.post,
	51	Method::DELETE => &info.delete,
	52	_ => &None,
	53	};
	54	if let Some(api_method) = opt_api_method {
	55	return Some(&api_method);
	56	}
	57	}
	58	None
	59	}
	60
	61	pub fn find_alias(&self, components: &[&str]) -> PathBuf {
	62
	63	let mut prefix = String::new();
	64	let mut filename = self.basedir.clone();
	65	let comp_len = components.len();
	66	if comp_len >= 1 {
	67	prefix.push_str(components[0]);
	68	if let Some(subdir) = self.aliases.get(&prefix) {
	69	filename.push(subdir);
	70	for i in 1..comp_len { filename.push(components[i]) }
	71	}
	72	}
	73	filename
	74	}
	75
	76	pub fn add_alias<S, P>(&mut self, alias: S, path: P)
	77	where S: Into<String>,
	78	P: Into<PathBuf>,
	79	{
	80	self.aliases.insert(alias.into(), path.into());
	81	}
	82	}
9bc17e8d DM	83
	84	type BoxFut = Box<Future<Item = Response<Body>, Error = failure::Error> + Send>;
	85
	86	macro_rules! error_response {
	87	($status:ident, $msg:expr) => {{
	88	let mut resp = Response::new(Body::from($msg));
	89	*resp.status_mut() = StatusCode::$status;
	90	resp
	91	}}
	92	}
	93
	94	macro_rules! http_error_future {
	95	($status:ident, $msg:expr) => {{
	96	let resp = error_response!($status, $msg);
	97	return Box::new(future::ok(resp));
	98	}}
	99	}
	100
	101	fn get_request_parameters_async<'a>(
	102	info: &'a ApiMethod,
	103	parts: Parts,
	104	req_body: Body,
	105	) -> Box<Future<Item = Value, Error = failure::Error> + Send + 'a>
	106	{
	107	let resp = req_body
	108	.map_err(\|err\| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("Promlems reading request body: {}", err))))
	109	.fold(Vec::new(), \|mut acc, chunk\| {
	110	if acc.len() + chunk.len() < 64*1024 { //fimxe: max request body size?
	111	acc.extend_from_slice(&*chunk);
	112	Ok(acc)
	113	}
	114	else { Err(Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("Request body too large")))) }
	115	})
	116	.and_then(move \|body\| {
	117
	118	let bytes = String::from_utf8(body.to_vec())?; // why copy??
	119
	120	println!("GOT BODY {:?}", bytes);
	121
	122	let mut test_required = true;
	123
	124	let mut params = json!({});
	125
	126	if bytes.len() > 0 {
	127	params = parse_query_string(&bytes, &info.parameters, true)?;
	128	test_required = false;
	129	}
	130
	131	if let Some(query_str) = parts.uri.query() {
	132	let query_params = parse_query_string(query_str, &info.parameters, test_required)?;
	133
	134	for (k, v) in query_params.as_object().unwrap() {
	135	params[k] = v.clone(); // fixme: why clone()??
	136	}
	137	}
	138
	139	println!("GOT PARAMS {}", params);
	140	Ok(params)
	141	});
	142
	143	Box::new(resp)
	144	}
	145
	146	fn handle_sync_api_request<'a>(
147	info: &'a ApiMethod,
148	parts: Parts,
149	req_body: Body,
150	) -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'a>
151	{
152	let params = get_request_parameters_async(info, parts, req_body);
153
154	let resp = params
155	.and_then(move \|params\| {
156
157	println!("GOT PARAMS {}", params);
158
159	/*
160	let when = Instant::now() + Duration::from_millis(3000);
161	let task = Delay::new(when).then(\|_\| {
162	println!("A LAZY TASK");
163	ok(())
164	});
165
166	tokio::spawn(task);
167	*/
168
169	let res = (info.handler)(params, info)?;
170
171	Ok(res)
172
173	}).then(\|result\| {
174	match result {
175	Ok(ref value) => {
176	let json_str = value.to_string();
177
178	Ok(Response::builder()
179	.status(StatusCode::OK)
180	.header(header::CONTENT_TYPE, "application/json")
181	.body(Body::from(json_str))?)
182	}
183	Err(err) => Ok(error_response!(BAD_REQUEST, err.to_string()))
184	}
185	});
186
187	Box::new(resp)
188	}
189
190	fn simple_static_file_download(filename: PathBuf) -> BoxFut {
191
192	Box::new(File::open(filename)
193	.map_err(\|err\| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File open failed: {}", err))))
194	.and_then(\|file\| {
195	let buf: Vec<u8> = Vec::new();
196	tokio::io::read_to_end(file, buf)
197	.map_err(\|err\| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File read failed: {}", err))))
198	.and_then(\|data\| Ok(Response::new(data.1.into())))
199	}))
200	}
201
202	fn chuncked_static_file_download(filename: PathBuf) -> BoxFut {
203
204	Box::new(File::open(filename)
205	.map_err(\|err\| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File open failed: {}", err))))
206	.and_then(\|file\| {
207	let payload = tokio_codec::FramedRead::new(file, tokio_codec::BytesCodec::new()).
208	map(\|bytes\| {
209	//sigh - howto avoid copy here? or the whole map() ??
210	hyper::Chunk::from(bytes.to_vec())
211	});
212	let body = Body::wrap_stream(payload);
213	// fixme: set content type and other headers
214	Ok(Response::builder()
215	.status(StatusCode::OK)
216	.body(body)
217	.unwrap())
218	}))
219	}
220
221	fn handle_static_file_download(filename: PathBuf) -> BoxFut {
222
223	let response = tokio::fs::metadata(filename.clone())
224	.map_err(\|err\| Error::from(ApiError::new(StatusCode::BAD_REQUEST, format!("File access problems: {}", err))))
225	.and_then(\|metadata\| {
226	if metadata.len() < 1024*32 {
227	Either::A(simple_static_file_download(filename))
228	} else {
229	Either::B(chuncked_static_file_download(filename))
230	}
231	});
232
233	return Box::new(response);
234	}
235
236	pub fn handle_request(api: Arc<ApiConfig>, req: Request<Body>) -> BoxFut {
237
238	let (parts, body) = req.into_parts();
239
240	let method = parts.method.clone();
241	let path = parts.uri.path();
242
243	// normalize path
244	// do not allow ".", "..", or hidden files ".XXXX"
245	// also remove empty path components
246
247	let items = path.split('/');
248	let mut path = String::new();
249	let mut components = vec![];
250
251	for name in items {
252	if name.is_empty() { continue; }
253	if name.starts_with(".") {
254	http_error_future!(BAD_REQUEST, "Path contains illegal components.\n");
255	}
256	path.push('/');
257	path.push_str(name);
258	components.push(name);
259	}
260
261	let comp_len = components.len();
262
263	println!("REQUEST {} {}", method, path);
264	println!("COMPO {:?}", components);
265
266	if comp_len >= 1 && components[0] == "api3" {
267	println!("GOT API REQUEST");
268	if comp_len >= 2 {
269	let format = components[1];
270	if format != "json" {
271	http_error_future!(BAD_REQUEST, format!("Unsupported output format '{}'.", format))
272	}
273
274	if let Some(api_method) = api.find_method(&components[2..], method) {
275	// fixme: handle auth
276	return handle_sync_api_request(api_method, parts, body);
277	}
278	}
279	} else {
280	// not Auth for accessing files!
281
282	let filename = api.find_alias(&components);
283	return handle_static_file_download(filename);
284	}
285
286	http_error_future!(NOT_FOUND, "Path not found.")
287	//Box::new(ok(Response::new(Body::from("RETURN WEB GUI\n"))))
288	}
289
290	pub struct RestServer {
291	pub api_config: Arc<ApiConfig>,
292	}
293
294	impl RestServer {
295
296	pub fn new(api_config: ApiConfig) -> Self {
297	Self { api_config: Arc::new(api_config) }
298	}
299	}
300
301	impl NewService for RestServer
302	{
303	type ReqBody = Body;
304	type ResBody = Body;
305	type Error = hyper::Error;
306	type InitError = hyper::Error;
307	type Service = ApiService;
308	type Future = Box<Future<Item = Self::Service, Error = Self::InitError> + Send>;
309	fn new_service(&self) -> Self::Future {
310	Box::new(future::ok(ApiService { api_config: self.api_config.clone() }))
311	}
312	}
313
314	pub struct ApiService {
315	pub api_config: Arc<ApiConfig>,
316	}
317
318	impl Service for ApiService {
319	type ReqBody = Body;
320	type ResBody = Body;
321	type Error = hyper::Error;
322	type Future = Box<Future<Item = Response<Body>, Error = Self::Error> + Send>;
323
324	fn call(&mut self, req: Request<Self::ReqBody>) -> Self::Future {
325
326	Box::new(handle_request(self.api_config.clone(), req).then(\|result\| {
327	match result {
328	Ok(res) => Ok::<_, hyper::Error>(res),
329	Err(err) => {
330	if let Some(apierr) = err.downcast_ref::<ApiError>() {
331	let mut resp = Response::new(Body::from(apierr.message.clone()));
332	*resp.status_mut() = apierr.code;
333	Ok(resp)
334	} else {
335	let mut resp = Response::new(Body::from(err.to_string()));
336	*resp.status_mut() = StatusCode::BAD_REQUEST;
337	Ok(resp)
338	}
339	}
340	}
341	}))
342	}
343	}