feat: new matcher system instead of manually 'match'ing in servcie
This commit is contained in:
parent
f305cb5a85
commit
638b0376d8
15 changed files with 646 additions and 225 deletions
|
|
@ -1,186 +0,0 @@
|
|||
use std::{net::IpAddr, pin::Pin, sync::Arc};
|
||||
|
||||
use base64::{Engine, prelude::BASE64_STANDARD};
|
||||
use http_body_util::BodyExt;
|
||||
use hyper::{
|
||||
Method, Request, StatusCode,
|
||||
body::Incoming,
|
||||
service::Service,
|
||||
};
|
||||
use log::{debug, error, warn};
|
||||
use tokio::{net::TcpStream, sync::Mutex};
|
||||
|
||||
use crate::{
|
||||
config::{Client, Config},
|
||||
db::{BoxyDatabase, Endpoint},
|
||||
server::{custom_resp, default_response, GeneralResponse, TcpIntercept},
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ApiService {
|
||||
pub database: Arc<Mutex<&'static mut BoxyDatabase>>,
|
||||
pub config: Config,
|
||||
pub _address: Option<IpAddr>,
|
||||
}
|
||||
|
||||
|
||||
impl TcpIntercept for ApiService {
|
||||
fn stream(&mut self, stream: &TcpStream) {
|
||||
self._address = Some(stream.peer_addr().unwrap().ip());
|
||||
}
|
||||
}
|
||||
|
||||
impl ApiService {
|
||||
pub async fn new(database: Arc<Mutex<&'static mut BoxyDatabase>>, config: Config) -> Self {
|
||||
Self {
|
||||
database,
|
||||
config,
|
||||
_address: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Service<Request<Incoming>> for ApiService {
|
||||
type Response = GeneralResponse;
|
||||
type Error = hyper::Error;
|
||||
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
|
||||
|
||||
fn call(&self, req: Request<Incoming>) -> Self::Future {
|
||||
let database = self.database.clone();
|
||||
let config = self.config.clone();
|
||||
let address = self._address.unwrap();
|
||||
|
||||
Box::pin(async move {
|
||||
match *req.method() {
|
||||
Method::POST => match req.uri().path() {
|
||||
"/register" => {
|
||||
debug!("new api register request from {}", address);
|
||||
|
||||
let encoded_header =
|
||||
match req.headers().get(hyper::header::AUTHORIZATION) {
|
||||
None => {
|
||||
error!(
|
||||
"Authorization header not given for request from {address}",
|
||||
);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Invalid credentials.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
Some(x) => x,
|
||||
}
|
||||
.to_str()
|
||||
.unwrap();
|
||||
|
||||
debug!("authorization header: {}", encoded_header);
|
||||
|
||||
let auth_bytes = match BASE64_STANDARD.decode(&encoded_header[6..]) {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
error!(
|
||||
"Error while decoding authorization header from {address}: {e}",
|
||||
);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Invalid base64 string given.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
};
|
||||
|
||||
let auth_string = match String::from_utf8(auth_bytes) {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
error!(
|
||||
"Error while decoding authorization header from {address}: {e}",
|
||||
);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Invalid UTF-8 in authentication string.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
};
|
||||
|
||||
debug!("decoded auth string: {}", auth_string);
|
||||
|
||||
if !Client::verify(auth_string.clone(), config).await {
|
||||
warn!(
|
||||
"Authentication for string {} from {} failed.",
|
||||
auth_string, address
|
||||
);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::UNAUTHORIZED,
|
||||
"Invalid credentials.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
|
||||
let body = match String::from_utf8(
|
||||
req.collect()
|
||||
.await
|
||||
.unwrap()
|
||||
.to_bytes()
|
||||
.iter()
|
||||
.cloned()
|
||||
.collect::<Vec<u8>>(),
|
||||
) {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
error!(
|
||||
"Error while inferring UTF-8 string from {address}'s request body: {e}",
|
||||
);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Invalid UTF-8 in body.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
};
|
||||
|
||||
let json = match json::parse(body.as_str()) {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
error!("Error while parsing JSON body from {address}: {e}",);
|
||||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Invalid JSON in body.",
|
||||
)
|
||||
.await);
|
||||
}
|
||||
};
|
||||
|
||||
debug!("body: {}", body);
|
||||
|
||||
let mut endpoint = Endpoint::new(
|
||||
None,
|
||||
address,
|
||||
json["port"].as_u16().unwrap_or(8080),
|
||||
json["callback"].as_str().unwrap_or("/").to_string(),
|
||||
)
|
||||
.await;
|
||||
|
||||
endpoint
|
||||
.register(
|
||||
*database.lock().await,
|
||||
json["hostname"].as_str().unwrap().to_string(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
Ok(custom_resp(StatusCode::OK, "Success").await)
|
||||
}
|
||||
_ => Ok(default_response().await),
|
||||
},
|
||||
_ => Ok(default_response().await),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -33,7 +33,11 @@ impl Service<Request<Incoming>> for ControllerService {
|
|||
None => {
|
||||
error!("No host header given.");
|
||||
|
||||
return Ok(custom_resp(StatusCode::BAD_REQUEST, "No host header given.").await);
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"No host header given.".to_string(),
|
||||
)
|
||||
.await);
|
||||
}
|
||||
}
|
||||
.to_str()
|
||||
|
|
@ -49,9 +53,11 @@ impl Service<Request<Incoming>> for ControllerService {
|
|||
None => {
|
||||
error!("No endpoint found for request.");
|
||||
|
||||
return Ok(
|
||||
custom_resp(StatusCode::NOT_FOUND, "No endpoint found for host.").await,
|
||||
);
|
||||
return Ok(custom_resp(
|
||||
StatusCode::NOT_FOUND,
|
||||
"No endpoint found for host.".to_string(),
|
||||
)
|
||||
.await);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
|
|||
129
src/services/matcher.rs
Normal file
129
src/services/matcher.rs
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
use std::{net::IpAddr, pin::Pin, sync::Arc};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use base64::{Engine, prelude::BASE64_STANDARD};
|
||||
use http::request::Parts;
|
||||
use http_body_util::BodyExt;
|
||||
use hyper::{
|
||||
Method, Request, StatusCode,
|
||||
body::{self, Incoming},
|
||||
service::Service,
|
||||
};
|
||||
use log::{debug, error, warn};
|
||||
use tokio::{net::TcpStream, sync::Mutex};
|
||||
|
||||
use crate::{
|
||||
config::{Client, Config},
|
||||
db::{BoxyDatabase, Endpoint},
|
||||
server::{GeneralResponse, TcpIntercept, custom_resp, default_response, json_to_vec},
|
||||
};
|
||||
|
||||
// The routes itself
|
||||
#[async_trait]
|
||||
pub trait Route<T>
|
||||
where
|
||||
T: Matcher,
|
||||
{
|
||||
fn matcher(&self, m: &T, req: &Request<Incoming>) -> bool;
|
||||
async fn call(&self, m: &T, parts: Parts) -> Result<GeneralResponse, hyper::Error>;
|
||||
}
|
||||
|
||||
// Matcher, essentially just a router that contains routes and some other features
|
||||
#[async_trait]
|
||||
pub trait Matcher: Clone + Send + Sync + 'static {
|
||||
// Essentially a kind of "middleware", a universal matcher. If it doesn't match, it won't
|
||||
// route.
|
||||
async fn unimatch(
|
||||
&mut self,
|
||||
req: &Request<Incoming>,
|
||||
) -> (bool, Option<Result<GeneralResponse, hyper::Error>>);
|
||||
|
||||
// Return list of routes associated with self matcher
|
||||
fn retrieve(&self) -> Vec<Arc<dyn Route<Self> + Sync + Send>>;
|
||||
|
||||
// Wrap self into matcher service
|
||||
fn service(self) -> MatcherService<Self> {
|
||||
MatcherService::new(self)
|
||||
}
|
||||
|
||||
// Do something with TCP stream
|
||||
fn stream(&mut self, stream: &TcpStream) {}
|
||||
|
||||
// Body parser - made universal for api server cause lazy
|
||||
async fn body(&mut self, body: Incoming) -> Option<Result<GeneralResponse, hyper::Error>> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
// Wrapper service, wraps matcher into a service
|
||||
#[derive(Clone)]
|
||||
pub struct MatcherService<T>
|
||||
where
|
||||
T: Matcher,
|
||||
{
|
||||
inner: T,
|
||||
}
|
||||
|
||||
impl<T> MatcherService<T>
|
||||
where
|
||||
T: Matcher,
|
||||
{
|
||||
pub fn new(inner: T) -> Self {
|
||||
Self { inner }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Service<Request<Incoming>> for MatcherService<T>
|
||||
where
|
||||
T: Matcher,
|
||||
{
|
||||
type Response = GeneralResponse;
|
||||
type Error = hyper::Error;
|
||||
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
|
||||
|
||||
fn call(&self, req: Request<Incoming>) -> Self::Future {
|
||||
let mut matcher = self.inner.clone();
|
||||
|
||||
Box::pin(async move {
|
||||
let unimatched = matcher.unimatch(&req).await;
|
||||
|
||||
if !unimatched.0 {
|
||||
match unimatched.1 {
|
||||
Some(x) => {
|
||||
return x;
|
||||
}
|
||||
None => {
|
||||
return Ok(custom_resp(
|
||||
StatusCode::NOT_FOUND,
|
||||
"Could not match route".to_string(),
|
||||
)
|
||||
.await);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for r in matcher.retrieve() {
|
||||
if r.matcher(&matcher, &req) {
|
||||
let (parts, body) = req.into_parts();
|
||||
|
||||
if let Some(resp) = matcher.body(body).await {
|
||||
return resp;
|
||||
}
|
||||
|
||||
return r.call(&matcher, parts).await;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(default_response().await)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> TcpIntercept for MatcherService<T>
|
||||
where
|
||||
T: Matcher,
|
||||
{
|
||||
fn stream(&mut self, stream: &TcpStream) {
|
||||
self.inner.stream(stream);
|
||||
}
|
||||
}
|
||||
|
|
@ -29,7 +29,7 @@ impl Service<Request<Incoming>> for ProxyService {
|
|||
|
||||
return Ok(custom_resp(
|
||||
StatusCode::BAD_GATEWAY,
|
||||
"Unable to open connection to endpoint.",
|
||||
"Unable to open connection to endpoint.".to_string(),
|
||||
)
|
||||
.await);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue