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

extern crate apitest;

use failure::*;
use std::collections::HashMap;

//use std::collections::HashMap;
use lazy_static::lazy_static;

//use apitest::json_schema::*;
use apitest::api_info::*;
use apitest::json_schema::*;

//use serde_derive::{Serialize, Deserialize};
use serde_json::{json, Value};

use futures::future::*;
//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::{Method, Body, Request, Response, Server, StatusCode};
use hyper::rt::{Future, Stream};
use hyper::service::service_fn;
use hyper::header;

//use std::time::{Duration, Instant};

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(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| format_err!("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(format_err!("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_async_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);
             */

            (info.async_handler)(params, info)
        });

    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 handle_static_file_download(filename: PathBuf) ->  BoxFut {

    let response = tokio::fs::metadata(filename.clone())
        .map_err(|err| format_err!("File access problems: {}", err))
        .and_then(|metadata| {
            println!("TEST METADATA {:?} {}", metadata, metadata.len());

            if metadata.len() < 1024*8 {
                println!("SMALL SIZED FILE");
                Either::A(File::open(filename)
                    .map_err(|err| format_err!("File open failed: {}", err))
                    .and_then(|file| {
                        let buf: Vec<u8> = Vec::new();
                        tokio::io::read_to_end(file, buf)
                            .map_err(|err| format_err!("File read failed: {}", err))
                            .and_then(|data| Ok(Response::new(data.1.into())))
                    }))

            } else {
                Either::B(
                    File::open(filename)
                        .map_err(|err| format_err!("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())
                    }))
            }
        });

    return Box::new(response);
}

fn handle_request(req: Request<Body>) -> BoxFut {

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

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

    // normalize path
    let components: Vec<&str> = path.split('/').filter(|x| !x.is_empty()).collect();
    let comp_len = components.len();
    let path = components.iter().fold(String::new(), |mut acc, chunk| {
        acc.push('/');
        acc.push_str(chunk);
        acc
    });

    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(info) = ROUTER.find_method(&components[2..]) {
                println!("FOUND INFO");
                let api_method_opt = match method {
                    Method::GET => &info.get,
                    Method::PUT => &info.put,
                    Method::POST => &info.post,
                    Method::DELETE => &info.delete,
                    _ => &None,
                };
                let api_method = match api_method_opt {
                    Some(m) => m,
                    _ => http_error_future!(NOT_FOUND, format!("No such method '{}'.", method)),
                };

                // fixme: handle auth

                //return handle_sync_api_request(api_method, parts, body);
                return handle_async_api_request(api_method, parts, body);
            }
        }
    }

    // not Auth for accessing files!
    if let Some(filename) = CACHED_DIRS.get(&path) {

        println!("SERVER STATIC FILE {:?}", path);
        return handle_static_file_download(filename.clone());
    }

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

// add default dirs which includes jquery and bootstrap
// my $base = '/usr/share/libpve-http-server-perl';
// add_dirs($self->{dirs}, '/css/' => "$base/css/");
// add_dirs($self->{dirs}, '/js/' => "$base/js/");
// add_dirs($self->{dirs}, '/fonts/' => "$base/fonts/");

//use std::io;
use std::fs;
use std::path::{Path, PathBuf};

fn add_dirs(cache: &mut HashMap<String, PathBuf>, alias: String, path: &Path) -> Result<(), Error> {

    if path.is_dir() {
        for direntry in fs::read_dir(path)? {
            let entry = direntry?;
            let entry_path = entry.path();
            let file_type = entry.file_type()?;
            if let Some(file_name) = entry_path.file_name() {
                let newalias = alias.clone() + &String::from(file_name.to_string_lossy()); // fixme
                if file_type.is_dir() {
                    add_dirs(cache, newalias, entry_path.as_path())?;
                } else if file_type.is_file() {
                    cache.insert(newalias, entry_path);
                }
            }
        }
    }
    Ok(())
 }

fn initialize_directory_cache() -> HashMap<String, PathBuf> {

    let mut basedirs = HashMap::new();

    basedirs.insert("novnc", Path::new("/usr/share/novnc-pve"));
    basedirs.insert("extjs", Path::new("/usr/share/javascript/extjs"));
    basedirs.insert("fontawesome", Path::new("/usr/share/fonts-font-awesome"));
    basedirs.insert("xtermjs", Path::new("/usr/share/pve-xtermjs"));
    basedirs.insert("widgettoolkit", Path::new("/usr/share/javascript/proxmox-widget-toolkit"));

    let mut cache = HashMap::new();

    if let Err(err) = add_dirs(&mut cache, "/pve2/ext6/".into(), basedirs["extjs"]) {
        eprintln!("directory cache init error: {}", err);
    }

    cache
}

lazy_static!{
    static ref CACHED_DIRS: HashMap<String, PathBuf> = initialize_directory_cache();
}

lazy_static!{
    static ref ROUTER: MethodInfo = apitest::api3::router();
}

fn main() {
    println!("Fast Static Type Definitions 1");

    let count = CACHED_DIRS.iter().count();
    println!("Dircache contains {} entries.", count);

    let addr = ([127, 0, 0, 1], 8007).into();

    let new_svc = || {
        service_fn(|req| {
            // clumsy way to convert failure::Error to Response
            handle_request(req).then(|result| -> Result<Response<Body>, String> {
                match result {
                    Ok(res) => Ok(res),
                    Err(err) => Ok(error_response!(BAD_REQUEST, err.to_string())),
                }
            })
        })
    };

    let server = Server::bind(&addr)
        .serve(new_svc)
        .map_err(|e| eprintln!("server error: {}", e));

    // Run this server for... forever!
    hyper::rt::run(server);
}
Commit	Line	Data
763220ce	1	extern crate apitest;
d8d978eb	2
6d77fb40	3	use failure::*;
ef1f0e65	4	use std::collections::HashMap;
d11f14f7	5
c819ec8d	6	//use std::collections::HashMap;
504b3597	7	use lazy_static::lazy_static;
28e47cea	8
504b3597	9	//use apitest::json_schema::*;
886e5ce8	10	use apitest::api_info::*;
6d77fb40	11	use apitest::json_schema::*;
886e5ce8	12
504b3597	13	//use serde_derive::{Serialize, Deserialize};
805aec15 DM	14	use serde_json::{json, Value};
805aec15 DM	15
579fbe7d DM	16	use futures::future::*;
	17	//use tokio::prelude::*;
	18	//use tokio::timer::Delay;
	19	use tokio::fs::File;
	20	use tokio_codec;
	21	//use bytes::{BytesMut, BufMut};
22f0adf2	22
c819ec8d	23	//use hyper::body::Payload;
b82472c0	24	use hyper::http::request::Parts;
28e47cea	25	use hyper::{Method, Body, Request, Response, Server, StatusCode};
b82472c0 DM	26	use hyper::rt::{Future, Stream};
b82472c0 DM	27	use hyper::service::service_fn;
3cdec2a0	28	use hyper::header;
b82472c0	29
579fbe7d	30	//use std::time::{Duration, Instant};
b82472c0	31
260c1ee8	32	type BoxFut = Box<Future<Item = Response<Body>, Error = failure::Error> + Send>;
d8d978eb	33
260c1ee8	34	macro_rules! error_response {
28e47cea DM	35	($status:ident, $msg:expr) => {{
	36	let mut resp = Response::new(Body::from($msg));
	37	*resp.status_mut() = StatusCode::$status;
260c1ee8	38	resp
b82472c0 DM	39	}}
b82472c0 DM	40	}
260c1ee8	41
b82472c0 DM	42	macro_rules! http_error_future {
b82472c0 DM	43	($status:ident, $msg:expr) => {{
260c1ee8	44	let resp = error_response!($status, $msg);
805aec15	45	return Box::new(ok(resp));
28e47cea DM	46	}}
	47	}
	48
805aec15	49	fn get_request_parameters_async<'a>(
c819ec8d DM	50	info: &'a ApiMethod,
	51	parts: Parts,
	52	req_body: Body,
805aec15	53	) -> Box<Future<Item = Value, Error = failure::Error> + Send + 'a>
b82472c0	54	{
cec9f02e	55	let resp = req_body
260c1ee8	56	.map_err(\|err\| format_err!("Promlems reading request body: {}", err))
cec9f02e DM	57	.fold(Vec::new(), \|mut acc, chunk\| {
	58	if acc.len() + chunk.len() < 64*1024 { //fimxe: max request body size?
	59	acc.extend_from_slice(&*chunk);
	60	ok(acc)
	61	} else {
	62	err(format_err!("Request body too large"))
	63	}
	64	})
260c1ee8	65	.and_then(move \|body\| {
b82472c0	66
260c1ee8	67	let bytes = String::from_utf8(body.to_vec())?; // why copy??
b82472c0	68
260c1ee8	69	println!("GOT BODY {:?}", bytes);
b82472c0	70
260c1ee8	71	let mut test_required = true;
b82472c0	72
260c1ee8	73	let mut params = json!({});
c819ec8d	74
260c1ee8 DM	75	if bytes.len() > 0 {
	76	params = parse_query_string(&bytes, &info.parameters, true)?;
	77	test_required = false;
	78	}
c819ec8d	79
260c1ee8 DM	80	if let Some(query_str) = parts.uri.query() {
260c1ee8 DM	81	let query_params = parse_query_string(query_str, &info.parameters, test_required)?;
c819ec8d	82
260c1ee8 DM	83	for (k, v) in query_params.as_object().unwrap() {
	84	params[k] = v.clone(); // fixme: why clone()??
	85	}
b82472c0	86	}
b82472c0	87
805aec15 DM	88	println!("GOT PARAMS {}", params);
	89	Ok(params)
	90	});
	91
	92	Box::new(resp)
	93	}
	94
	95	fn handle_async_api_request<'a>(
	96	info: &'a ApiMethod,
	97	parts: Parts,
	98	req_body: Body,
	99	) -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'a>
	100	{
	101	let params = get_request_parameters_async(info, parts, req_body);
	102
	103	let resp = params
	104	.and_then(move \|params\| {
	105
	106	println!("GOT PARAMS {}", params);
	107
	108	/*
	109	let when = Instant::now() + Duration::from_millis(3000);
	110	let task = Delay::new(when).then(\|_\| {
	111	println!("A LAZY TASK");
	112	ok(())
	113	});
	114
	115	tokio::spawn(task);
	116	*/
	117
	118	(info.async_handler)(params, info)
	119	});
	120
	121	Box::new(resp)
	122	}
	123
	124	fn handle_sync_api_request<'a>(
	125	info: &'a ApiMethod,
	126	parts: Parts,
	127	req_body: Body,
	128	) -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'a>
	129	{
	130	let params = get_request_parameters_async(info, parts, req_body);
	131
	132	let resp = params
	133	.and_then(move \|params\| {
	134
260c1ee8 DM	135	println!("GOT PARAMS {}", params);
260c1ee8 DM	136
805aec15 DM	137	/*
	138	let when = Instant::now() + Duration::from_millis(3000);
	139	let task = Delay::new(when).then(\|_\| {
	140	println!("A LAZY TASK");
	141	ok(())
	142	});
	143
	144	tokio::spawn(task);
	145	*/
	146
260c1ee8	147	let res = (info.handler)(params, info)?;
c819ec8d	148
260c1ee8	149	Ok(res)
c819ec8d	150
805aec15 DM	151	}).then(\|result\| {
	152	match result {
	153	Ok(ref value) => {
	154	let json_str = value.to_string();
c819ec8d	155
805aec15 DM	156	Ok(Response::builder()
805aec15 DM	157	.status(StatusCode::OK)
3cdec2a0	158	.header(header::CONTENT_TYPE, "application/json")
805aec15 DM	159	.body(Body::from(json_str))?)
	160	}
	161	Err(err) => Ok(error_response!(BAD_REQUEST, err.to_string()))
260c1ee8	162	}
805aec15	163	});
b82472c0 DM	164
	165	Box::new(resp)
	166	}
	167
579fbe7d DM	168	fn handle_static_file_download(filename: PathBuf) -> BoxFut {
579fbe7d DM	169
78d0783b DM	170	let response = tokio::fs::metadata(filename.clone())
	171	.map_err(\|err\| format_err!("File access problems: {}", err))
	172	.and_then(\|metadata\| {
	173	println!("TEST METADATA {:?} {}", metadata, metadata.len());
	174
	175	if metadata.len() < 1024*8 {
	176	println!("SMALL SIZED FILE");
	177	Either::A(File::open(filename)
	178	.map_err(\|err\| format_err!("File open failed: {}", err))
	179	.and_then(\|file\| {
	180	let buf: Vec<u8> = Vec::new();
	181	tokio::io::read_to_end(file, buf)
	182	.map_err(\|err\| format_err!("File read failed: {}", err))
	183	.and_then(\|data\| Ok(Response::new(data.1.into())))
	184	}))
	185
	186	} else {
	187	Either::B(
	188	File::open(filename)
	189	.map_err(\|err\| format_err!("File open failed: {}", err))
	190	.and_then(\|file\| {
	191	let payload = tokio_codec::FramedRead::new(file, tokio_codec::BytesCodec::new()).
	192	map(\|bytes\| {
	193	//sigh - howto avoid copy here? or the whole map() ??
	194	hyper::Chunk::from(bytes.to_vec())
	195	});
	196	let body = Body::wrap_stream(payload);
	197	// fixme: set content type and other headers
	198	Ok(Response::builder()
	199	.status(StatusCode::OK)
	200	.body(body)
	201	.unwrap())
	202	}))
	203	}
579fbe7d DM	204	});
	205
	206	return Box::new(response);
	207	}
	208
b82472c0 DM	209	fn handle_request(req: Request<Body>) -> BoxFut {
	210
	211	let (parts, body) = req.into_parts();
28e47cea	212
b82472c0 DM	213	let method = parts.method.clone();
b82472c0 DM	214	let path = parts.uri.path();
886e5ce8	215
579fbe7d	216	// normalize path
28e47cea DM	217	let components: Vec<&str> = path.split('/').filter(\|x\| !x.is_empty()).collect();
28e47cea DM	218	let comp_len = components.len();
579fbe7d DM	219	let path = components.iter().fold(String::new(), \|mut acc, chunk\| {
	220	acc.push('/');
	221	acc.push_str(chunk);
	222	acc
	223	});
886e5ce8 DM	224
886e5ce8 DM	225	println!("REQUEST {} {}", method, path);
28e47cea DM	226	println!("COMPO {:?}", components);
	227
	228	if comp_len >= 1 && components[0] == "api3" {
	229	println!("GOT API REQUEST");
	230	if comp_len >= 2 {
	231	let format = components[1];
	232	if format != "json" {
6639c14b	233	http_error_future!(BAD_REQUEST, format!("Unsupported output format '{}'.", format))
28e47cea DM	234	}
28e47cea DM	235
0dde2f04	236	if let Some(info) = ROUTER.find_method(&components[2..]) {
28e47cea DM	237	println!("FOUND INFO");
28e47cea DM	238	let api_method_opt = match method {
b82472c0 DM	239	Method::GET => &info.get,
	240	Method::PUT => &info.put,
	241	Method::POST => &info.post,
	242	Method::DELETE => &info.delete,
0dde2f04	243	_ => &None,
28e47cea	244	};
bfcba4fd	245	let api_method = match api_method_opt {
28e47cea	246	Some(m) => m,
6639c14b	247	_ => http_error_future!(NOT_FOUND, format!("No such method '{}'.", method)),
28e47cea DM	248	};
28e47cea DM	249
4beaacb6	250	// fixme: handle auth
28e47cea	251
805aec15 DM	252	//return handle_sync_api_request(api_method, parts, body);
805aec15 DM	253	return handle_async_api_request(api_method, parts, body);
28e47cea DM	254	}
	255	}
	256	}
886e5ce8	257
579fbe7d DM	258	// not Auth for accessing files!
	259	if let Some(filename) = CACHED_DIRS.get(&path) {
	260
	261	println!("SERVER STATIC FILE {:?}", path);
	262	return handle_static_file_download(filename.clone());
	263	}
	264
	265	http_error_future!(NOT_FOUND, "Path not found.")
	266	//Box::new(ok(Response::new(Body::from("RETURN WEB GUI\n"))))
886e5ce8 DM	267	}
886e5ce8 DM	268
ef1f0e65 DM	269	// add default dirs which includes jquery and bootstrap
	270	// my $base = '/usr/share/libpve-http-server-perl';
	271	// add_dirs($self->{dirs}, '/css/' => "$base/css/");
	272	// add_dirs($self->{dirs}, '/js/' => "$base/js/");
	273	// add_dirs($self->{dirs}, '/fonts/' => "$base/fonts/");
	274
579fbe7d DM	275	//use std::io;
579fbe7d DM	276	use std::fs;
ef1f0e65 DM	277	use std::path::{Path, PathBuf};
	278
	279	fn add_dirs(cache: &mut HashMap<String, PathBuf>, alias: String, path: &Path) -> Result<(), Error> {
	280
	281	if path.is_dir() {
	282	for direntry in fs::read_dir(path)? {
	283	let entry = direntry?;
	284	let entry_path = entry.path();
	285	let file_type = entry.file_type()?;
	286	if let Some(file_name) = entry_path.file_name() {
	287	let newalias = alias.clone() + &String::from(file_name.to_string_lossy()); // fixme
	288	if file_type.is_dir() {
	289	add_dirs(cache, newalias, entry_path.as_path())?;
	290	} else if file_type.is_file() {
	291	cache.insert(newalias, entry_path);
	292	}
	293	}
	294	}
	295	}
	296	Ok(())
	297	}
	298
	299	fn initialize_directory_cache() -> HashMap<String, PathBuf> {
	300
	301	let mut basedirs = HashMap::new();
	302
	303	basedirs.insert("novnc", Path::new("/usr/share/novnc-pve"));
	304	basedirs.insert("extjs", Path::new("/usr/share/javascript/extjs"));
	305	basedirs.insert("fontawesome", Path::new("/usr/share/fonts-font-awesome"));
	306	basedirs.insert("xtermjs", Path::new("/usr/share/pve-xtermjs"));
	307	basedirs.insert("widgettoolkit", Path::new("/usr/share/javascript/proxmox-widget-toolkit"));
	308
	309	let mut cache = HashMap::new();
	310
579fbe7d DM	311	if let Err(err) = add_dirs(&mut cache, "/pve2/ext6/".into(), basedirs["extjs"]) {
	312	eprintln!("directory cache init error: {}", err);
	313	}
ef1f0e65 DM	314
	315	cache
	316	}
	317
	318	lazy_static!{
	319	static ref CACHED_DIRS: HashMap<String, PathBuf> = initialize_directory_cache();
	320	}
	321
504b3597	322	lazy_static!{
0dde2f04	323	static ref ROUTER: MethodInfo = apitest::api3::router();
504b3597 DM	324	}
504b3597 DM	325
d8d978eb DM	326	fn main() {
	327	println!("Fast Static Type Definitions 1");
	328
579fbe7d	329	let count = CACHED_DIRS.iter().count();
ef1f0e65 DM	330	println!("Dircache contains {} entries.", count);
ef1f0e65 DM	331
886e5ce8 DM	332	let addr = ([127, 0, 0, 1], 8007).into();
	333
	334	let new_svc = \|\| {
260c1ee8 DM	335	service_fn(\|req\| {
	336	// clumsy way to convert failure::Error to Response
	337	handle_request(req).then(\|result\| -> Result<Response<Body>, String> {
	338	match result {
	339	Ok(res) => Ok(res),
6639c14b	340	Err(err) => Ok(error_response!(BAD_REQUEST, err.to_string())),
260c1ee8 DM	341	}
	342	})
	343	})
886e5ce8 DM	344	};
	345
	346	let server = Server::bind(&addr)
	347	.serve(new_svc)
	348	.map_err(\|e\| eprintln!("server error: {}", e));
	349
	350	// Run this server for... forever!
	351	hyper::rt::run(server);
d8d978eb	352	}