1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
use serde::Deserialize;
use hyper::{Request, Body, Method, Uri};
pub mod parse;
use parse::UrlQueryParams;
#[derive(Default, Deserialize, Debug, Clone)]
pub struct ProxyConfig {
pub bind_addr: Option<String>, // 127.0.0.1:9292
pub upstream_addr: Option<String>, // 127.0.0.1:9200
pub allow_all_indices: Option<bool>,
pub index: Vec<IndexConfig>
}
#[derive(Deserialize, Debug, Clone)]
pub struct IndexConfig {
pub name: String,
}
impl ProxyConfig {
pub fn allow_index(&self, name: &str) -> bool {
if self.allow_all_indices == Some(true) {
return true
}
for index in &self.index {
if index.name == name {
return true
}
}
false
}
}
#[derive(Debug)]
pub enum ProxyError {
Malformed(String),
ParseError(String),
NotAllowed(String),
NotSupported(String),
NotFound(String),
}
pub async fn filter_request(req: Request<Body>, config: &ProxyConfig) -> Result<Request<Body>, ProxyError> {
let (parts, body) = req.into_parts();
// split path into at most 3 chunks
let mut req_path = parts.uri.path();
if req_path.starts_with("/") {
req_path = &req_path[1..];
}
let path_chunks: Vec<&str> = req_path.split("/").collect();
if path_chunks.len() > 3 {
return Err(ProxyError::NotSupported("only request paths with up to three segments allowed".to_string()))
}
let params: UrlQueryParams = serde_urlencoded::from_str(parts.uri.query().unwrap_or("")).unwrap();
// this is sort of like a router
let body = match (&parts.method, path_chunks.as_slice()) {
(&Method::GET, [""]) | (&Method::HEAD, [""]) => {
Body::empty()
},
(&Method::POST, ["_search", "scroll"]) | (&Method::DELETE, ["_search", "scroll"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
filter_scroll_request(¶ms, &whole_body, config)?
},
(&Method::GET, [index, "_search"]) | (&Method::POST, [index, "_search"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
filter_search_request(index, ¶ms, &whole_body, config)?
},
(&Method::GET, [index, "_count"]) | (&Method::POST, [index, "_count"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
filter_search_request(index, ¶ms, &whole_body, config)?
},
(&Method::GET, [index, "_doc", key]) | (&Method::GET, [index, "_source", key]) => {
filter_read_request(index, path_chunks[1], key, ¶ms, config)?
},
_ => Err(ProxyError::NotSupported("unknown endpoint".to_string()))?,
};
let upstream_query = serde_urlencoded::to_string(params).unwrap();
let upstream_query_and_params = if upstream_query.len() > 0 {
format!("{}?{}", req_path, upstream_query)
} else {
req_path.to_string()
};
let upstream_uri = Uri::builder()
.scheme("http")
.authority(config.upstream_addr.as_ref().unwrap_or(&"localhost:9200".to_string()).as_str())
.path_and_query(upstream_query_and_params.as_str())
.build()
.unwrap();
let upstream_req = Request::builder()
.uri(upstream_uri)
.method(&parts.method)
.body(body)
.unwrap();
Ok(upstream_req)
}
pub fn filter_scroll_request(_params: &UrlQueryParams, _body: &[u8], _config: &ProxyConfig) -> Result<Body, ProxyError> {
// XXX
// TODO: check that scroll_id is not "_all"
//let _parsed: ScrollBody = serde_json::from_str(&body).unwrap();
Err(ProxyError::NotSupported("not yet implemented".to_string()))
}
pub fn filter_read_request(index: &str, _endpoint: &str, _key: &str, _params: &UrlQueryParams, config: &ProxyConfig) -> Result<Body, ProxyError>{
if !config.allow_index(index) {
return Err(ProxyError::NotAllowed(format!("index doesn't exist or isn't proxied: {}", index)));
}
// XXX: no body needed?
Ok(Body::empty())
}
pub fn filter_search_request(index: &str, _params: &UrlQueryParams, body: &[u8], config: &ProxyConfig) -> Result<Body, ProxyError> {
if !config.allow_index(index) {
return Err(ProxyError::NotAllowed(format!("index doesn't exist or isn't proxied: {}", index)));
}
// XXX: more checks
if body.len() > 0 {
let parsed: parse::ScrollBody = serde_json::from_slice(body).unwrap();
Ok(Body::from(serde_json::to_string(&parsed).unwrap()))
} else {
Ok(Body::empty())
}
}
|
