Moved http service out of Registry into separate struct
This commit is contained in:
107
src/tunnel/mod.rs
Normal file
107
src/tunnel/mod.rs
Normal file
@@ -0,0 +1,107 @@
|
||||
mod registry;
|
||||
mod tui;
|
||||
|
||||
use registry::RegistryEntry;
|
||||
use std::sync::Arc;
|
||||
use tracing::trace;
|
||||
|
||||
use russh::server::Handle;
|
||||
use tokio::sync::{RwLock, RwLockReadGuard};
|
||||
|
||||
use crate::{stats::Stats, wrapper::Wrapper};
|
||||
|
||||
pub use registry::Registry;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum TunnelAccess {
|
||||
Private(String),
|
||||
Protected,
|
||||
Public,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TunnelInner {
|
||||
handle: Handle,
|
||||
internal_address: String,
|
||||
port: u32,
|
||||
access: Arc<RwLock<TunnelAccess>>,
|
||||
stats: Arc<Stats>,
|
||||
}
|
||||
|
||||
impl TunnelInner {
|
||||
pub async fn open(&self) -> Result<Wrapper, russh::Error> {
|
||||
trace!("Opening tunnel");
|
||||
self.stats.add_connection();
|
||||
let channel = self
|
||||
.handle
|
||||
.channel_open_forwarded_tcpip(
|
||||
&self.internal_address,
|
||||
self.port,
|
||||
&self.internal_address,
|
||||
self.port,
|
||||
)
|
||||
.await?;
|
||||
|
||||
Ok(Wrapper::new(channel.into_stream(), self.stats.clone()))
|
||||
}
|
||||
|
||||
pub async fn is_public(&self) -> bool {
|
||||
matches!(*self.access.read().await, TunnelAccess::Public)
|
||||
}
|
||||
|
||||
pub async fn get_access(&self) -> RwLockReadGuard<'_, TunnelAccess> {
|
||||
self.access.read().await
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Tunnel {
|
||||
inner: TunnelInner,
|
||||
|
||||
registry: Registry,
|
||||
registry_entry: RegistryEntry,
|
||||
}
|
||||
|
||||
impl Tunnel {
|
||||
pub async fn create(
|
||||
registry: &mut Registry,
|
||||
handle: Handle,
|
||||
internal_address: impl Into<String>,
|
||||
port: u32,
|
||||
access: TunnelAccess,
|
||||
) -> Self {
|
||||
let mut tunnel = Self {
|
||||
inner: TunnelInner {
|
||||
handle,
|
||||
internal_address: internal_address.into(),
|
||||
port,
|
||||
access: Arc::new(RwLock::new(access)),
|
||||
stats: Default::default(),
|
||||
},
|
||||
registry: registry.clone(),
|
||||
registry_entry: RegistryEntry::new(registry.clone()),
|
||||
};
|
||||
|
||||
registry.register(&mut tunnel).await;
|
||||
|
||||
tunnel
|
||||
}
|
||||
|
||||
pub async fn set_access(&self, access: TunnelAccess) {
|
||||
*self.inner.access.write().await = access;
|
||||
}
|
||||
|
||||
pub fn get_address(&self) -> Option<&String> {
|
||||
self.registry_entry.get_address()
|
||||
}
|
||||
|
||||
pub async fn set_name(&mut self, name: impl Into<String>) {
|
||||
let mut registry = self.registry.clone();
|
||||
registry.rename(self, name).await;
|
||||
}
|
||||
|
||||
pub async fn retry(&mut self) {
|
||||
let mut registry = self.registry.clone();
|
||||
registry.register(self).await;
|
||||
}
|
||||
}
|
||||
@@ -1,29 +1,12 @@
|
||||
use std::{
|
||||
collections::{HashMap, hash_map::Entry},
|
||||
ops::Deref,
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use bytes::Bytes;
|
||||
use http_body_util::{BodyExt as _, Empty, combinators::BoxBody};
|
||||
use hyper::{
|
||||
Request, Response, StatusCode,
|
||||
body::Incoming,
|
||||
client::conn::http1::Builder,
|
||||
header::{self, HOST},
|
||||
service::Service,
|
||||
};
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, error, trace, warn};
|
||||
use tracing::trace;
|
||||
|
||||
use crate::{
|
||||
Tunnel,
|
||||
animals::get_animal_name,
|
||||
auth::{AuthStatus, ForwardAuth},
|
||||
helper::response,
|
||||
tunnel::TunnelAccess,
|
||||
};
|
||||
use crate::{Tunnel, animals::get_animal_name};
|
||||
|
||||
use super::TunnelInner;
|
||||
|
||||
@@ -73,15 +56,13 @@ impl Drop for RegistryEntry {
|
||||
pub struct Registry {
|
||||
tunnels: Arc<RwLock<HashMap<String, TunnelInner>>>,
|
||||
domain: String,
|
||||
auth: ForwardAuth,
|
||||
}
|
||||
|
||||
impl Registry {
|
||||
pub fn new(domain: impl Into<String>, auth: ForwardAuth) -> Self {
|
||||
pub fn new(domain: impl Into<String>) -> Self {
|
||||
Self {
|
||||
tunnels: Arc::new(RwLock::new(HashMap::new())),
|
||||
domain: domain.into(),
|
||||
auth,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -142,120 +123,8 @@ impl Registry {
|
||||
tunnel.registry_entry.name = name.into();
|
||||
self.register(tunnel).await;
|
||||
}
|
||||
}
|
||||
|
||||
impl Service<Request<Incoming>> for Registry {
|
||||
type Response = Response<BoxBody<Bytes, hyper::Error>>;
|
||||
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 {
|
||||
trace!("{:#?}", req);
|
||||
|
||||
let Some(authority) = req
|
||||
.uri()
|
||||
.authority()
|
||||
.as_ref()
|
||||
.map(|a| a.to_string())
|
||||
.or_else(|| {
|
||||
req.headers()
|
||||
.get(HOST)
|
||||
.and_then(|h| h.to_str().ok().map(|s| s.to_owned()))
|
||||
})
|
||||
else {
|
||||
let resp = response(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Missing or invalid authority or host header",
|
||||
);
|
||||
|
||||
return Box::pin(async { Ok(resp) });
|
||||
};
|
||||
|
||||
debug!(authority, "Tunnel request");
|
||||
|
||||
let s = self.clone();
|
||||
Box::pin(async move {
|
||||
let Some(entry) = s.tunnels.read().await.get(&authority).cloned() else {
|
||||
debug!(tunnel = authority, "Unknown tunnel");
|
||||
let resp = response(StatusCode::NOT_FOUND, "Unknown tunnel");
|
||||
|
||||
return Ok(resp);
|
||||
};
|
||||
|
||||
if !matches!(entry.access.read().await.deref(), TunnelAccess::Public) {
|
||||
let user = match s.auth.check_auth(req.method(), req.headers()).await {
|
||||
Ok(AuthStatus::Authenticated(user)) => user,
|
||||
Ok(AuthStatus::Unauthenticated(location)) => {
|
||||
let resp = Response::builder()
|
||||
.status(StatusCode::FOUND)
|
||||
.header(header::LOCATION, location)
|
||||
.body(
|
||||
Empty::new()
|
||||
// NOTE: I have NO idea why this is able to convert from Innfallible to hyper::Error
|
||||
.map_err(|never| match never {})
|
||||
.boxed(),
|
||||
)
|
||||
.expect("configuration should be valid");
|
||||
|
||||
return Ok(resp);
|
||||
}
|
||||
Ok(AuthStatus::Unauthorized) => {
|
||||
let resp = response(
|
||||
StatusCode::FORBIDDEN,
|
||||
"You do not have permission to access this tunnel",
|
||||
);
|
||||
|
||||
return Ok(resp);
|
||||
}
|
||||
Err(err) => {
|
||||
error!("Unexpected error during authentication: {err}");
|
||||
let resp = response(
|
||||
StatusCode::FORBIDDEN,
|
||||
"Unexpected error during authentication",
|
||||
);
|
||||
|
||||
return Ok(resp);
|
||||
}
|
||||
};
|
||||
|
||||
trace!("Tunnel is getting accessed by {user:?}");
|
||||
|
||||
if let TunnelAccess::Private(owner) = entry.access.read().await.deref() {
|
||||
if !user.is(owner) {
|
||||
let resp = response(
|
||||
StatusCode::FORBIDDEN,
|
||||
"You do not have permission to access this tunnel",
|
||||
);
|
||||
|
||||
return Ok(resp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let io = match entry.open().await {
|
||||
Ok(io) => io,
|
||||
Err(err) => {
|
||||
warn!(tunnel = authority, "Failed to open tunnel: {err}");
|
||||
let resp = response(StatusCode::INTERNAL_SERVER_ERROR, "Failed to open tunnel");
|
||||
|
||||
return Ok(resp);
|
||||
}
|
||||
};
|
||||
|
||||
let (mut sender, conn) = Builder::new()
|
||||
.preserve_header_case(true)
|
||||
.title_case_headers(true)
|
||||
.handshake(io)
|
||||
.await?;
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(err) = conn.await {
|
||||
warn!(runnel = authority, "Connection failed: {err}");
|
||||
}
|
||||
});
|
||||
|
||||
let resp = sender.send_request(req).await?;
|
||||
Ok(resp.map(|b| b.boxed()))
|
||||
})
|
||||
pub async fn get(&self, address: &str) -> Option<TunnelInner> {
|
||||
self.tunnels.read().await.get(address).cloned()
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user