Goatworks/RoadSign
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

♻️ Migrated to actix rs

Browse Source
This commit is contained in:
LittleSheep
2024-02-13 00:01:39 +08:00
parent ead748a508
commit b7d4a54d62
65 changed files with 7039 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/main.rs
View File

@@ -1,16 +1,14 @@
pub mod auth;
mod config;
mod proxies;
mod sideload;
pub mod warden;
use actix_web::{App, HttpServer, web};
use awc::Client;
use lazy_static::lazy_static;
use poem::{listener::TcpListener, EndpointExt, Route, Server};
use poem_openapi::OpenApiService;
use proxies::RoadInstance;
use tokio::sync::Mutex;
use tracing::{error, info, Level};
use crate::proxies::route;
lazy_static! {
@@ -20,9 +18,6 @@ lazy_static! {
#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
// Setting up logging
if std::env::var_os("RUST_LOG").is_none() {
std::env::set_var("RUST_LOG", "poem=debug");
}
tracing_subscriber::fmt()
.with_max_level(Level::DEBUG)
.init();
@@ -44,36 +39,17 @@ async fn main() -> Result<(), std::io::Error> {
};
// Proxies
let proxies_server = Server::new(TcpListener::bind(
let proxies_server = HttpServer::new(|| {
App::new()
.app_data(web::Data::new(Client::default()))
.route("/", web::to(route::handle))
}).bind(
config::C
.read()
.await
.get_string("listen.proxies")
.unwrap_or("0.0.0.0:80".to_string()),
))
.run(route::handle);
// Sideload
let sideload = OpenApiService::new(sideload::SideloadApi, "Sideload API", "1.0")
.server("http://localhost:3000/cgi");
let sideload_server = Server::new(TcpListener::bind(
config::C
.read()
.await
.get_string("listen.sideload")
.unwrap_or("0.0.0.0:81".to_string()),
))
.run(
Route::new().nest("/cgi", sideload).with(auth::BasicAuth {
username: "RoadSign".to_string(),
password: config::C
.read()
.await
.get_string("secret")
.unwrap_or("password".to_string()),
}),
);
.unwrap_or("0.0.0.0:80".to_string())
)?.run();
// Process manager
{
@@ -85,7 +61,7 @@ async fn main() -> Result<(), std::io::Error> {
app.warden.start().await;
}
tokio::try_join!(proxies_server, sideload_server)?;
tokio::try_join!(proxies_server)?;
Ok(())
}

52
src/proxies/browser.rs
View File

@@ -1,52 +0,0 @@
use std::fmt::Write;
pub struct DirectoryTemplate<'a> {
pub path: &'a str,
pub files: Vec<FileRef>,
}
impl<'a> DirectoryTemplate<'a> {
pub fn render(&self) -> String {
let mut s = format!(
r#"
<html>
<head>
<title>Index of {}</title>
</head>
<body>
<h1>Index of /{}</h1>
<ul>"#,
self.path, self.path
);
for file in &self.files {
if file.is_dir {
let _ = write!(
s,
r#"<li><a href="{}">{}/</a></li>"#,
file.url, file.filename
);
} else {
let _ = write!(
s,
r#"<li><a href="{}">{}</a></li>"#,
file.url, file.filename
);
}
}
s.push_str(
r#"</ul>
</body>
</html>"#,
);
s
}
}
pub struct FileRef {
pub url: String,
pub filename: String,
pub is_dir: bool,
}

9
src/proxies/config.rs
View File

@@ -75,15 +75,6 @@ impl Destination {
.collect::<Vec<_>>()[0]
}
pub fn get_websocket_uri(&self) -> Result<String, ()> {
let parts = self.uri.as_str().splitn(2, "://").collect::<Vec<_>>();
let url = parts.get(1).unwrap_or(&"");
match self.get_protocol() {
"http" | "https" => Ok(url.replace("http", "ws")),
_ => Err(()),
}
}
pub fn get_hypertext_uri(&self) -> Result<String, ()> {
match self.get_protocol() {
"http" => Ok("http://".to_string() + self.get_host()),

7
src/proxies/mod.rs
View File

@@ -1,5 +1,5 @@
use http::Method;
use poem::http::{HeaderMap, Uri};
use actix_web::http::header::HeaderMap;
use actix_web::http::{Method, Uri};
use regex::Regex;
use wildmatch::WildMatch;
@@ -10,7 +10,6 @@ use self::{
metrics::RoadMetrics,
};
pub mod browser;
pub mod config;
pub mod loader;
pub mod metrics;
@@ -38,7 +37,7 @@ impl RoadInstance {
pub fn filter(
&self,
uri: &Uri,
method: Method,
method: &Method,
headers: &HeaderMap,
) -> Option<(&Region, &Location)> {
self.regions.iter().find_map(|region| {

219
src/proxies/responder.rs
View File

@@ -1,117 +1,50 @@
use futures_util::{SinkExt, StreamExt};
use http::{header, request::Builder, HeaderMap, Method, StatusCode, Uri};
use lazy_static::lazy_static;
use poem::{
web::{websocket::WebSocket, StaticFileRequest},
Body, Error, FromRequest, IntoResponse, Request, Response,
};
use futures_util::{SinkExt};
use std::{
ffi::OsStr,
path::{Path, PathBuf},
sync::Arc,
};
use tokio::sync::RwLock;
use tokio_tungstenite::connect_async;
use super::browser::{DirectoryTemplate, FileRef};
lazy_static! {
pub static ref CLIENT: reqwest::Client = reqwest::Client::new();
}
pub async fn repond_websocket(req: Builder, ws: WebSocket) -> Response {
ws.on_upgrade(move |socket| async move {
let (mut clientsink, mut clientstream) = socket.split();
// Start connection to server
let (serversocket, _) = connect_async(req.body(()).unwrap()).await.unwrap();
let (mut serversink, mut serverstream) = serversocket.split();
let client_live = Arc::new(RwLock::new(true));
let server_live = client_live.clone();
tokio::spawn(async move {
while let Some(Ok(msg)) = clientstream.next().await {
if (serversink.send(msg.into()).await).is_err() {
break;
};
if !*client_live.read().await {
break;
};
}
*client_live.write().await = false;
});
// Relay server messages to the client
tokio::spawn(async move {
while let Some(Ok(msg)) = serverstream.next().await {
if (clientsink.send(msg.into()).await).is_err() {
break;
};
if !*server_live.read().await {
break;
};
}
*server_live.write().await = false;
});
})
.into_response()
}
use actix_files::{NamedFile};
use actix_proxy::IntoHttpResponse;
use actix_web::{HttpRequest, HttpResponse, web};
use actix_web::http::Method;
use awc::Client;
pub async fn respond_hypertext(
uri: String,
ori: &Uri,
req: &Request,
method: Method,
body: Body,
headers: &HeaderMap,
) -> Result<Response, Error> {
let ip = req.remote_addr().to_string();
req: HttpRequest,
client: web::Data<Client>,
) -> Result<HttpResponse, HttpResponse> {
let ip = req.peer_addr().unwrap().ip().to_string();
let proto = req.uri().scheme_str().unwrap();
let host = req.uri().host().unwrap();
let mut headers = headers.clone();
headers.insert("Server", "RoadSign".parse().unwrap());
headers.insert("X-Forward-For", ip.parse().unwrap());
headers.insert("X-Forwarded-Proto", proto.parse().unwrap());
headers.insert("X-Forwarded-Host", host.parse().unwrap());
headers.insert("X-Real-IP", ip.parse().unwrap());
let mut headers = req.headers().clone();
headers.insert("Server".parse().unwrap(), "RoadSign".parse().unwrap());
headers.insert("X-Forward-For".parse().unwrap(), ip.parse().unwrap());
headers.insert("X-Forwarded-Proto".parse().unwrap(), proto.parse().unwrap());
headers.insert("X-Forwarded-Host".parse().unwrap(), host.parse().unwrap());
headers.insert("X-Real-IP".parse().unwrap(), ip.parse().unwrap());
headers.insert(
"Forwarded",
"Forwarded".parse().unwrap(),
format!("by={};for={};host={};proto={}", ip, ip, host, proto)
.parse()
.unwrap(),
);
let res = CLIENT
.request(method, uri + ori.path() + ori.query().unwrap_or(""))
.headers(headers.clone())
.body(body.into_bytes().await.unwrap())
.send()
.await;
let res = client.get(uri).send().await;
match res {
return match res {
Ok(result) => {
let mut res = Response::default();
res.extensions().clone_from(&result.extensions());
result.headers().iter().for_each(|(key, val)| {
res.headers_mut().insert(key, val.to_owned());
});
res.headers_mut()
.insert("Server", "RoadSign".parse().unwrap());
res.set_status(result.status());
res.set_version(result.version());
res.set_body(result.bytes().await.unwrap());
let mut res = result.into_http_response();
res.headers_mut().insert("Server".parse().unwrap(), "RoadSign".parse().unwrap());
Ok(res)
}
Err(error) => Err(Error::from_string(
error.to_string(),
error.status().unwrap_or(StatusCode::BAD_GATEWAY),
)),
}
Err(error) => {
Err(HttpResponse::BadGateway()
.body(format!("Something went wrong... {:}", error)))
}
};
}
pub struct StaticResponderConfig {
@@ -126,14 +59,11 @@ pub struct StaticResponderConfig {
pub async fn respond_static(
cfg: StaticResponderConfig,
method: Method,
req: &Request,
) -> Result<Response, Error> {
if method != Method::GET {
return Err(Error::from_string(
"This destination only support GET request.",
StatusCode::METHOD_NOT_ALLOWED,
));
req: HttpRequest,
) -> Result<HttpResponse, HttpResponse> {
if req.method() != Method::GET {
return Err(HttpResponse::MethodNotAllowed()
.body("This destination only support GET request."));
}
let path = req
@@ -142,9 +72,12 @@ pub async fn respond_static(
.trim_start_matches('/')
.trim_end_matches('/');
let path = percent_encoding::percent_decode_str(path)
.decode_utf8()
.map_err(|_| Error::from_status(StatusCode::NOT_FOUND))?;
let path = match percent_encoding::percent_decode_str(path).decode_utf8() {
Ok(val) => val,
Err(_) => {
return Err(HttpResponse::NotFound().body("Not found."));
}
};
let base_path = cfg.uri.parse::<PathBuf>().unwrap();
let mut file_path = base_path.clone();
@@ -159,7 +92,8 @@ pub async fn respond_static(
}
if !file_path.starts_with(cfg.uri) {
return Err(Error::from_status(StatusCode::FORBIDDEN));
return Err(HttpResponse::Forbidden()
.body("Unexpected path."));
}
if !file_path.exists() {
@@ -172,87 +106,30 @@ pub async fn respond_static(
file_path.pop();
file_path.push((file_name + &suffix).as_str());
if file_path.is_file() {
return Ok(StaticFileRequest::from_request_without_body(req)
.await?
.create_response(&file_path, cfg.utf8)?
.into_response());
return Ok(NamedFile::open(file_path).unwrap().into_response(&req));
}
}
if let Some(file) = cfg.fallback {
let fallback_path = base_path.join(file);
if fallback_path.is_file() {
return Ok(StaticFileRequest::from_request_without_body(req)
.await?
.create_response(&fallback_path, cfg.utf8)?
.into_response());
return Ok(NamedFile::open(fallback_path).unwrap().into_response(&req));
}
}
return Err(Error::from_status(StatusCode::NOT_FOUND));
return Err(HttpResponse::NotFound().body("Not found."));
}
if file_path.is_file() {
Ok(StaticFileRequest::from_request_without_body(req)
.await?
.create_response(&file_path, cfg.utf8)?
.into_response())
return if file_path.is_file() {
Ok(NamedFile::open(file_path).unwrap().into_response(&req))
} else {
if cfg.with_slash
&& !req.original_uri().path().ends_with('/')
&& (cfg.index.is_some() || cfg.browse)
{
let redirect_to = format!("{}/", req.original_uri().path());
return Ok(Response::builder()
.status(StatusCode::FOUND)
.header(header::LOCATION, redirect_to)
.finish());
}
if let Some(index_file) = &cfg.index {
let index_path = file_path.join(index_file);
if index_path.is_file() {
return Ok(StaticFileRequest::from_request_without_body(req)
.await?
.create_response(&index_path, cfg.utf8)?
.into_response());
return Ok(NamedFile::open(index_path).unwrap().into_response(&req));
}
}
if cfg.browse {
let read_dir = file_path
.read_dir()
.map_err(|_| Error::from_status(StatusCode::FORBIDDEN))?;
let mut template = DirectoryTemplate {
path: &path,
files: Vec::new(),
};
for res in read_dir {
let entry = res.map_err(|_| Error::from_status(StatusCode::FORBIDDEN))?;
if let Some(filename) = entry.file_name().to_str() {
let mut base_url = req.original_uri().path().to_string();
if !base_url.ends_with('/') {
base_url.push('/');
}
let filename_url = percent_encoding::percent_encode(
filename.as_bytes(),
percent_encoding::NON_ALPHANUMERIC,
);
template.files.push(FileRef {
url: format!("{base_url}{filename_url}"),
filename: filename.to_string(),
is_dir: entry.path().is_dir(),
});
}
}
let html = template.render();
Ok(Response::builder()
.header(header::CONTENT_TYPE, mime::TEXT_HTML_UTF_8.as_ref())
.body(Body::from_string(html)))
} else {
Err(Error::from_status(StatusCode::NOT_FOUND))
}
}
Err(HttpResponse::NotFound().body("Not found."))
};
}

90
src/proxies/route.rs
View File

@@ -1,10 +1,5 @@
use http::Method;
use poem::{
handler,
http::{HeaderMap, StatusCode, Uri},
web::websocket::WebSocket,
Body, Error, FromRequest, IntoResponse, Request, Response, Result,
};
use actix_web::{HttpRequest, HttpResponse, web};
use awc::Client;
use rand::seq::SliceRandom;
use crate::{
@@ -15,22 +10,13 @@ use crate::{
ROAD,
};
#[handler]
pub async fn handle(
req: &Request,
uri: &Uri,
headers: &HeaderMap,
method: Method,
body: Body,
) -> Result<impl IntoResponse, Error> {
pub async fn handle(req: HttpRequest, client: web::Data<Client>) -> HttpResponse {
let readable_app = ROAD.lock().await;
let (region, location) = match readable_app.filter(uri, method.clone(), headers) {
let (region, location) = match readable_app.filter(req.uri(), req.method(), req.headers()) {
Some(val) => val,
None => {
return Err(Error::from_string(
"There are no region be able to respone this request.",
StatusCode::NOT_FOUND,
))
return HttpResponse::NotFound()
.body("There are no region be able to respone this request.");
}
};
@@ -41,58 +27,31 @@ pub async fn handle(
async fn forward(
end: &Destination,
req: &Request,
ori: &Uri,
headers: &HeaderMap,
method: Method,
body: Body,
) -> Result<Response, Error> {
// Handle websocket
if let Ok(ws) = WebSocket::from_request_without_body(req).await {
// Get uri
let Ok(uri) = end.get_websocket_uri() else {
return Err(Error::from_string(
"This destination was not support websockets.",
StatusCode::NOT_IMPLEMENTED,
));
};
// Build request
let mut ws_req = http::Request::builder().uri(&uri);
for (key, value) in headers.iter() {
ws_req = ws_req.header(key, value);
}
// Start the websocket connection
return Ok(responder::repond_websocket(ws_req, ws).await);
}
req: HttpRequest,
client: web::Data<Client>,
) -> Result<HttpResponse, HttpResponse> {
// Handle normal web request
match end.get_type() {
DestinationType::Hypertext => {
let Ok(uri) = end.get_hypertext_uri() else {
return Err(Error::from_string(
"This destination was not support web requests.",
StatusCode::NOT_IMPLEMENTED,
));
return Err(HttpResponse::NotImplemented()
.body("This destination was not support web requests."));
};
responder::respond_hypertext(uri, ori, req, method, body, headers).await
responder::respond_hypertext(uri, req, client).await
}
DestinationType::StaticFiles => {
let Ok(cfg) = end.get_static_config() else {
return Err(Error::from_string(
"This destination was not support static files.",
StatusCode::NOT_IMPLEMENTED,
));
return Err(HttpResponse::NotImplemented()
.body("This destination was not support static files."));
};
responder::respond_static(cfg, method, req).await
responder::respond_static(cfg, req).await
}
_ => {
return Err(HttpResponse::NotImplemented()
.body("Unsupported destination protocol."));
}
_ => Err(Error::from_string(
"Unsupported destination protocol.",
StatusCode::NOT_IMPLEMENTED,
)),
}
}
@@ -100,23 +59,22 @@ pub async fn handle(
let loc = location.clone();
let end = destination.clone();
match forward(&end, req, uri, headers, method, body).await {
return match forward(&end, req, client).await {
Ok(resp) => {
tokio::spawn(async move {
let writable_app = &mut ROAD.lock().await;
writable_app.metrics.add_success_request(reg, loc, end);
});
Ok(resp)
resp
}
Err(err) => {
let message = format!("{:}", err);
Err(resp) => {
tokio::spawn(async move {
let writable_app = &mut ROAD.lock().await;
writable_app
.metrics
.add_faliure_request(reg, loc, end, message);
.add_faliure_request(reg, loc, end, "TODO".to_owned());
});
Err(err)
resp
}
}
}

19
src/sideload/mod.rs
View File

@@ -1,19 +0,0 @@
use poem_openapi::OpenApi;
pub mod overview;
pub mod regions;
pub struct SideloadApi;
#[OpenApi]
impl SideloadApi {
#[oai(path = "/", method = "get")]
async fn index(&self) -> overview::OverviewResponse {
overview::index().await
}
#[oai(path = "/regions", method = "get")]
async fn regions_index(&self) -> regions::RegionResponse {
regions::index().await
}
}

75
src/sideload/overview.rs
View File

@@ -1,75 +0,0 @@
use poem_openapi::{payload::Json, ApiResponse, Object};
use crate::{
proxies::{
config::{Destination, Location},
metrics::RoadTrace,
},
ROAD,
};
#[derive(ApiResponse)]
pub enum OverviewResponse {
/// Return the overview data.
#[oai(status = 200)]
Ok(Json<OverviewData>),
}
#[derive(Debug, Object, Clone, PartialEq)]
pub struct OverviewData {
/// Loaded regions count
#[oai(read_only)]
regions: usize,
/// Loaded locations count
#[oai(read_only)]
locations: usize,
/// Loaded destnations count
#[oai(read_only)]
destinations: usize,
/// Recent requests count
requests_count: u64,
/// Recent requests success count
faliures_count: u64,
/// Recent requests falied count
successes_count: u64,
/// Recent requests success rate
success_rate: f64,
/// Recent successes
recent_successes: Vec<RoadTrace>,
/// Recent errors
recent_errors: Vec<RoadTrace>,
}
pub async fn index() -> OverviewResponse {
let locked_app = ROAD.lock().await;
let regions = locked_app.regions.clone();
let locations = regions
.iter()
.flat_map(|item| item.locations.clone())
.collect::<Vec<Location>>();
let destinations = locations
.iter()
.flat_map(|item| item.destinations.clone())
.collect::<Vec<Destination>>();
OverviewResponse::Ok(Json(OverviewData {
regions: regions.len(),
locations: locations.len(),
destinations: destinations.len(),
requests_count: locked_app.metrics.requests_count,
successes_count: locked_app.metrics.requests_count - locked_app.metrics.failures_count,
faliures_count: locked_app.metrics.failures_count,
success_rate: locked_app.metrics.get_success_rate(),
recent_successes: locked_app
.metrics
.recent_successes
.clone()
.into_iter()
.collect::<Vec<_>>(),
recent_errors: locked_app
.metrics
.recent_errors
.clone()
.into_iter()
.collect::<Vec<_>>(),
}))
}

25
src/sideload/regions.rs
View File

@@ -1,25 +0,0 @@
use poem_openapi::{payload::Json, ApiResponse};
use crate::{proxies::config::Region, ROAD};
#[derive(ApiResponse)]
pub enum RegionResponse {
/// Return the region data.
#[oai(status = 200)]
Ok(Json<Region>),
/// Return the list of region data.
#[oai(status = 200)]
OkMany(Json<Vec<Region>>),
/// Return the region data after created.
#[oai(status = 201)]
Created(Json<Region>),
/// Return was not found.
#[oai(status = 404)]
NotFound,
}
pub async fn index() -> RegionResponse {
let locked_app = ROAD.lock().await;
RegionResponse::OkMany(Json(locked_app.regions.clone()))
}