Compare commits

...

3 Commits

Author SHA1 Message Date
Brian Picciano
c1659fab2a Got gemini proxy working, via a custom tokio_rustls branch 2023-07-24 19:06:01 +02:00
Brian Picciano
c8176c819f Got basic gemini listening working. Proxying does not yet work, nor does serving the origin 2023-07-22 11:31:55 +02:00
Brian Picciano
2d1e237735 A bit of module structure tidying 2023-07-21 14:43:39 +02:00
16 changed files with 340 additions and 98 deletions

View File

@ -21,6 +21,10 @@ service:
value: hi value: hi
- name: user-agent - name: user-agent
value: "" value: ""
gemini:
proxied_domains:
mediocregopher.com:
url: gemini://127.0.0.1:1965
passphrase: foobar passphrase: foobar
dns_records: dns_records:
- kind: A - kind: A

5
Cargo.lock generated
View File

@ -2963,9 +2963,8 @@ dependencies = [
[[package]] [[package]]
name = "tokio-rustls" name = "tokio-rustls"
version = "0.24.0" version = "0.24.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "git+https://code.betamike.com/micropelago/tokio-rustls.git?branch=transparent-acceptor#18fd688b335430e17e054e15ff7d6ce073db2419"
checksum = "e0d409377ff5b1e3ca6437aa86c1eb7d40c134bfec254e44c830defa92669db5"
dependencies = [ dependencies = [
"rustls", "rustls",
"tokio", "tokio",

View File

@ -20,7 +20,7 @@ serde_json = "1.0.96"
trust-dns-client = "0.22.0" trust-dns-client = "0.22.0"
mockall = "0.11.4" mockall = "0.11.4"
thiserror = "1.0.40" thiserror = "1.0.40"
tokio = { version = "1.28.1", features = [ "full" ]} tokio = { version = "1.28.1", features = [ "full", "net" ]}
signal-hook = "0.3.15" signal-hook = "0.3.15"
futures = "0.3.28" futures = "0.3.28"
signal-hook-tokio = { version = "0.3.1", features = [ "futures-v0_3" ]} signal-hook-tokio = { version = "0.3.1", features = [ "futures-v0_3" ]}
@ -45,3 +45,6 @@ serde_yaml = "0.9.22"
rand = "0.8.5" rand = "0.8.5"
reqwest = "0.11.18" reqwest = "0.11.18"
hyper-reverse-proxy = "0.5.1" hyper-reverse-proxy = "0.5.1"
[patch.crates-io]
tokio-rustls = { git = "https://code.betamike.com/micropelago/tokio-rustls.git", branch = "transparent-acceptor" }

View File

@ -44,6 +44,8 @@
pkgs.stdenv.cc pkgs.stdenv.cc
pkgs.openssl pkgs.openssl
toolchain toolchain
pkgs.nmap # ncat
]; ];
shellHook = '' shellHook = ''
source $(pwd)/.env.dev source $(pwd)/.env.dev

View File

@ -1,7 +1,6 @@
use crate::domain::{self, acme, checker, store}; use crate::domain::{self, acme, checker, store};
use crate::error::unexpected::{self, Mappable}; use crate::error::unexpected::{self, Mappable};
use crate::origin; use crate::{origin, task_stack, util};
use crate::util;
use std::sync; use std::sync;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
@ -187,7 +186,7 @@ impl ManagerImpl {
DomainStore: store::Store + Send + Sync + 'static, DomainStore: store::Store + Send + Sync + 'static,
AcmeManager: acme::manager::Manager + Send + Sync + 'static, AcmeManager: acme::manager::Manager + Send + Sync + 'static,
>( >(
task_stack: &mut util::TaskStack<unexpected::Error>, task_stack: &mut task_stack::TaskStack<unexpected::Error>,
origin_store: OriginStore, origin_store: OriginStore,
domain_store: DomainStore, domain_store: DomainStore,
domain_checker: checker::DNSChecker, domain_checker: checker::DNSChecker,

View File

@ -4,8 +4,10 @@
pub mod config; pub mod config;
pub mod domain; pub mod domain;
pub mod error;
pub mod origin; pub mod origin;
pub mod service; pub mod service;
pub mod task_stack;
pub mod token; pub mod token;
pub mod util;
mod error;
mod util;

View File

@ -128,7 +128,7 @@ async fn main() {
None None
}; };
let mut task_stack = domani::util::TaskStack::new(); let mut task_stack = domani::task_stack::TaskStack::new();
let domain_manager = domani::domain::manager::ManagerImpl::new( let domain_manager = domani::domain::manager::ManagerImpl::new(
&mut task_stack, &mut task_stack,
@ -142,6 +142,12 @@ async fn main() {
&mut task_stack, &mut task_stack,
domain_manager.clone(), domain_manager.clone(),
domain_manager.clone(), domain_manager.clone(),
config.service.clone(),
);
let _ = domani::service::gemini::Service::new(
&mut task_stack,
domain_manager.clone(),
config.service, config.service,
); );

View File

@ -1,5 +1,5 @@
mod config; mod config;
pub mod gemini;
pub mod http; pub mod http;
mod util;
pub use config::*; pub use config::*;

View File

@ -24,7 +24,7 @@ impl From<ConfigDNSRecord> for domain::checker::DNSRecord {
} }
} }
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize, Clone)]
pub struct Config { pub struct Config {
pub passphrase: String, pub passphrase: String,
pub dns_records: Vec<ConfigDNSRecord>, pub dns_records: Vec<ConfigDNSRecord>,
@ -32,4 +32,6 @@ pub struct Config {
pub primary_domain: domain::Name, pub primary_domain: domain::Name,
#[serde(default)] #[serde(default)]
pub http: service::http::Config, pub http: service::http::Config,
#[serde(default)]
pub gemini: service::gemini::Config,
} }

144
src/service/gemini.rs Normal file
View File

@ -0,0 +1,144 @@
mod config;
pub use config::*;
use crate::error::unexpected::{self, Mappable};
use crate::{domain, service, task_stack};
use std::sync;
use tokio_util::sync::CancellationToken;
pub struct Service {
cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
config: service::Config,
}
#[derive(thiserror::Error, Debug)]
enum HandleConnError {
#[error("client error: {0}")]
ClientError(String),
#[error(transparent)]
Unexpected(#[from] unexpected::Error),
}
impl Service {
pub fn new(
task_stack: &mut task_stack::TaskStack<unexpected::Error>,
cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
config: service::Config,
) -> sync::Arc<Service> {
let service = sync::Arc::new(Service {
cert_resolver,
config,
});
task_stack.push_spawn(|canceller| listen(service.clone(), canceller));
service
}
async fn proxy_conn<IO>(
&self,
proxied_domain: &ConfigProxiedDomain,
mut conn: IO,
) -> unexpected::Result<()>
where
IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let mut proxy_conn = tokio::net::TcpStream::connect(&proxied_domain.url.addr)
.await
.map_unexpected_while(|| {
format!("failed to connect to proxy {}", proxied_domain.url.url,)
})?;
_ = tokio::io::copy_bidirectional(&mut conn, &mut proxy_conn).await;
Ok(())
}
async fn handle_conn(
&self,
conn: tokio::net::TcpStream,
_tls_config: sync::Arc<rustls::ServerConfig>,
) -> Result<(), HandleConnError> {
let acceptor =
tokio_rustls::TransparentConfigAcceptor::new(rustls::server::Acceptor::default(), conn);
futures::pin_mut!(acceptor);
match acceptor.as_mut().await {
Ok(start) => {
let client_hello = start.client_hello();
let domain = client_hello.server_name().ok_or_else(|| {
HandleConnError::ClientError("missing SNI in ClientHello".to_string())
})?;
let domain: domain::Name = domain.parse().map_err(|e| {
HandleConnError::ClientError(format!(
"parsing domain {domain}, provided in SNI: {e}"
))
})?;
// If the domain should be proxied, then proxy it
if let Some(proxied_domain) = self.config.gemini.proxied_domains.get(&domain) {
let conn = start.into_original_stream();
self.proxy_conn(proxied_domain, conn).await?;
return Ok(());
}
return Err(HandleConnError::ClientError(format!(
"unknown domain {domain}"
)));
}
Err(err) => {
return Err(unexpected::Error::from(
format!("failed to accept TLS connection: {err}").as_str(),
)
.into())
}
}
}
}
async fn listen(
service: sync::Arc<Service>,
canceller: CancellationToken,
) -> unexpected::Result<()> {
let tls_config = sync::Arc::new(
rustls::server::ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth() // TODO maybe this isn't right?
.with_cert_resolver(service.cert_resolver.clone()),
);
log::info!(
"Listening on gemini://{}:{}",
&service.config.primary_domain.clone(),
&service.config.gemini.gemini_addr.port(),
);
let listener = tokio::net::TcpListener::bind(service.config.gemini.gemini_addr)
.await
.or_unexpected_while("binding tcp socket")?;
loop {
let (conn, addr) = tokio::select! {
res = listener.accept() => res.or_unexpected_while("accepting connection")?,
_ = canceller.cancelled() => return Ok(()),
};
let service = service.clone();
let tls_config = tls_config.clone();
tokio::spawn(async move {
match service.handle_conn(conn, tls_config).await {
Ok(_) => (),
Err(HandleConnError::ClientError(e)) => {
log::warn!("Bad request from connection {addr}: {e}")
}
Err(HandleConnError::Unexpected(e)) => {
log::error!("Server error handling connection {addr}: {e}")
}
}
});
}
}

View File

@ -0,0 +1,79 @@
use crate::domain;
use crate::error::unexpected::{self, Mappable};
use serde::{Deserialize, Serialize};
use serde_with::{serde_as, TryFromInto};
use std::{collections, net, str::FromStr};
fn default_gemini_addr() -> net::SocketAddr {
net::SocketAddr::from_str("[::]:3965").unwrap()
}
#[derive(Deserialize, Serialize, Clone)]
pub struct ConfigProxiedDomainUrl {
pub url: String,
pub addr: String,
}
impl From<ConfigProxiedDomainUrl> for String {
fn from(url: ConfigProxiedDomainUrl) -> Self {
url.url
}
}
impl TryFrom<String> for ConfigProxiedDomainUrl {
type Error = unexpected::Error;
fn try_from(url: String) -> Result<Self, Self::Error> {
// use http's implementation, should be the same
let parsed = http::Uri::from_str(url.as_str())
.map_unexpected_while(|| format!("parsing proxy url {url}"))?;
let scheme = parsed.scheme().map_unexpected_while(|| {
format!("expected a scheme of gemini in the proxy url {url}")
})?;
if scheme != "gemini" {
return Err(unexpected::Error::from(
format!("scheme of proxy url {url} should be 'gemini'",).as_str(),
));
}
match parsed.authority() {
None => Err(unexpected::Error::from(
format!("proxy url {url} should have a host",).as_str(),
)),
Some(authority) => {
let port = authority.port().map(|p| p.as_u16()).unwrap_or(1965);
Ok(ConfigProxiedDomainUrl {
url: url,
addr: format!("{}:{port}", authority.host()),
})
}
}
}
}
#[serde_as]
#[derive(Deserialize, Serialize, Clone)]
pub struct ConfigProxiedDomain {
#[serde_as(as = "TryFromInto<String>")]
pub url: ConfigProxiedDomainUrl,
}
#[derive(Deserialize, Serialize, Clone)]
pub struct Config {
#[serde(default = "default_gemini_addr")]
pub gemini_addr: net::SocketAddr,
pub proxied_domains: collections::HashMap<domain::Name, ConfigProxiedDomain>,
}
impl Default for Config {
fn default() -> Self {
Self {
gemini_addr: default_gemini_addr(),
proxied_domains: Default::default(),
}
}
}

View File

@ -2,6 +2,7 @@ mod config;
mod proxy; mod proxy;
mod tasks; mod tasks;
mod tpl; mod tpl;
mod util;
pub use config::*; pub use config::*;
@ -13,7 +14,7 @@ use std::str::FromStr;
use std::{future, net, sync}; use std::{future, net, sync};
use crate::error::unexpected; use crate::error::unexpected;
use crate::{domain, service, util}; use crate::{domain, service, task_stack};
pub struct Service { pub struct Service {
domain_manager: sync::Arc<dyn domain::manager::Manager>, domain_manager: sync::Arc<dyn domain::manager::Manager>,
@ -23,7 +24,7 @@ pub struct Service {
} }
pub fn new( pub fn new(
task_stack: &mut util::TaskStack<unexpected::Error>, task_stack: &mut task_stack::TaskStack<unexpected::Error>,
domain_manager: sync::Arc<dyn domain::manager::Manager>, domain_manager: sync::Arc<dyn domain::manager::Manager>,
cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>, cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
config: service::Config, config: service::Config,
@ -64,7 +65,7 @@ struct DomainInitArgs {
domain: domain::Name, domain: domain::Name,
#[serde(flatten)] #[serde(flatten)]
flat_domain_settings: service::util::FlatDomainSettings, url_encoded_domain_settings: util::UrlEncodedDomainSettings,
} }
#[derive(Deserialize)] #[derive(Deserialize)]
@ -73,7 +74,7 @@ struct DomainSyncArgs {
passphrase: String, passphrase: String,
#[serde(flatten)] #[serde(flatten)]
flat_domain_settings: service::util::FlatDomainSettings, url_encoded_domain_settings: util::UrlEncodedDomainSettings,
} }
impl<'svc> Service { impl<'svc> Service {
@ -267,7 +268,7 @@ impl<'svc> Service {
#[derive(Serialize)] #[derive(Serialize)]
struct Data<'a> { struct Data<'a> {
domain: domain::Name, domain: domain::Name,
flat_domain_settings: service::util::FlatDomainSettings, url_encoded_domain_settings: util::UrlEncodedDomainSettings,
dns_records: &'a [service::ConfigDNSRecord], dns_records: &'a [service::ConfigDNSRecord],
challenge_token: String, challenge_token: String,
@ -275,7 +276,7 @@ impl<'svc> Service {
dns_records_have_cname: bool, dns_records_have_cname: bool,
} }
let settings: domain::Settings = match args.flat_domain_settings.try_into() { let settings: domain::Settings = match args.url_encoded_domain_settings.try_into() {
Ok(settings) => settings, Ok(settings) => settings,
Err(e) => { Err(e) => {
return self return self
@ -298,7 +299,7 @@ impl<'svc> Service {
_ => false, _ => false,
}); });
let flat_domain_settings = match settings.try_into() { let url_encoded_domain_settings = match settings.try_into() {
Ok(s) => s, Ok(s) => s,
Err(e) => { Err(e) => {
return self return self
@ -310,7 +311,7 @@ impl<'svc> Service {
"/domain_init.html", "/domain_init.html",
Data { Data {
domain: args.domain, domain: args.domain,
flat_domain_settings, url_encoded_domain_settings,
dns_records: &self.config.dns_records, dns_records: &self.config.dns_records,
challenge_token: settings_hash, challenge_token: settings_hash,
@ -325,7 +326,7 @@ impl<'svc> Service {
return self.render_error_page(401, "Incorrect passphrase"); return self.render_error_page(401, "Incorrect passphrase");
} }
let settings: domain::Settings = match args.flat_domain_settings.try_into() { let settings: domain::Settings = match args.url_encoded_domain_settings.try_into() {
Ok(settings) => settings, Ok(settings) => settings,
Err(e) => { Err(e) => {
return self return self

View File

@ -7,10 +7,10 @@ use serde_with::{serde_as, TryFromInto};
use std::{collections, net, str::FromStr}; use std::{collections, net, str::FromStr};
fn default_http_addr() -> net::SocketAddr { fn default_http_addr() -> net::SocketAddr {
net::SocketAddr::from_str("[::]:3030").unwrap() net::SocketAddr::from_str("[::]:3080").unwrap()
} }
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize, Clone)]
pub enum ConfigFormMethod { pub enum ConfigFormMethod {
GET, GET,
POST, POST,
@ -60,7 +60,7 @@ impl TryFrom<String> for ConfigProxiedDomainUrl {
fn try_from(url: String) -> Result<Self, Self::Error> { fn try_from(url: String) -> Result<Self, Self::Error> {
let parsed = http::Uri::from_str(url.as_str()) let parsed = http::Uri::from_str(url.as_str())
.or_unexpected_while("parsing proxy url {proxy_url}")?; .map_unexpected_while(|| format!("parsing proxy url {url}"))?;
let scheme = parsed let scheme = parsed
.scheme() .scheme()
@ -76,7 +76,7 @@ impl TryFrom<String> for ConfigProxiedDomainUrl {
} }
} }
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize, Clone)]
pub struct ConfigProxiedDomainRequestHeader { pub struct ConfigProxiedDomainRequestHeader {
pub name: String, pub name: String,
pub value: String, pub value: String,
@ -129,7 +129,7 @@ impl TryFrom<Vec<ConfigProxiedDomainRequestHeader>> for ConfigProxiedDomainReque
} }
#[serde_as] #[serde_as]
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize, Clone)]
pub struct ConfigProxiedDomain { pub struct ConfigProxiedDomain {
#[serde_as(as = "TryFromInto<String>")] #[serde_as(as = "TryFromInto<String>")]
pub url: ConfigProxiedDomainUrl, pub url: ConfigProxiedDomainUrl,
@ -139,7 +139,7 @@ pub struct ConfigProxiedDomain {
pub request_headers: ConfigProxiedDomainRequestHeaders, pub request_headers: ConfigProxiedDomainRequestHeaders,
} }
#[derive(Deserialize, Serialize)] #[derive(Deserialize, Serialize, Clone)]
pub struct Config { pub struct Config {
#[serde(default = "default_http_addr")] #[serde(default = "default_http_addr")]
pub http_addr: net::SocketAddr, pub http_addr: net::SocketAddr,

View File

@ -7,7 +7,7 @@ use crate::{domain, error::unexpected, origin};
#[serde_as] #[serde_as]
#[derive(Serialize, Deserialize, Default, Debug)] #[derive(Serialize, Deserialize, Default, Debug)]
pub struct FlatDomainSettings { pub struct UrlEncodedDomainSettings {
domain_setting_origin_descr_kind: String, domain_setting_origin_descr_kind: String,
domain_setting_origin_descr_git_url: Option<String>, domain_setting_origin_descr_git_url: Option<String>,
@ -20,10 +20,10 @@ pub struct FlatDomainSettings {
domain_setting_add_path_prefix: Option<String>, domain_setting_add_path_prefix: Option<String>,
} }
impl TryFrom<FlatDomainSettings> for domain::Settings { impl TryFrom<UrlEncodedDomainSettings> for domain::Settings {
type Error = String; type Error = String;
fn try_from(v: FlatDomainSettings) -> Result<Self, Self::Error> { fn try_from(v: UrlEncodedDomainSettings) -> Result<Self, Self::Error> {
let origin_descr = match v.domain_setting_origin_descr_kind.as_str() { let origin_descr = match v.domain_setting_origin_descr_kind.as_str() {
"git" => Ok(origin::Descr::Git { "git" => Ok(origin::Descr::Git {
url: v url: v
@ -44,11 +44,11 @@ impl TryFrom<FlatDomainSettings> for domain::Settings {
} }
} }
impl TryFrom<domain::Settings> for FlatDomainSettings { impl TryFrom<domain::Settings> for UrlEncodedDomainSettings {
type Error = unexpected::Error; type Error = unexpected::Error;
fn try_from(v: domain::Settings) -> Result<Self, Self::Error> { fn try_from(v: domain::Settings) -> Result<Self, Self::Error> {
let mut res = FlatDomainSettings::default(); let mut res = UrlEncodedDomainSettings::default();
match v.origin_descr { match v.origin_descr {
origin::Descr::Git { url, branch_name } => { origin::Descr::Git { url, branch_name } => {

66
src/task_stack.rs Normal file
View File

@ -0,0 +1,66 @@
use crate::util::BoxFuture;
use std::error;
use tokio_util::sync::CancellationToken;
pub struct TaskStack<E>
where
E: error::Error + Send + 'static,
{
wait_group: Vec<BoxFuture<'static, Result<(), E>>>,
}
impl<E> TaskStack<E>
where
E: error::Error + Send + 'static,
{
pub fn new() -> TaskStack<E> {
TaskStack {
wait_group: Vec::new(),
}
}
/// push adds the given Future to the stack, to be executed once stop is called.
pub fn push<Fut>(&mut self, f: Fut)
where
Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
{
self.wait_group.push(Box::pin(f))
}
/// push_spawn will spawn the given closure in a tokio task. Once the CancellationToken is
/// cancelled the closure is expected to return.
pub fn push_spawn<F, Fut>(&mut self, mut f: F)
where
Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
F: FnMut(CancellationToken) -> Fut,
{
let canceller = CancellationToken::new();
let handle = tokio::spawn(f(canceller.clone()));
self.push(async move {
canceller.cancel();
handle.await.expect("failed to join task")
});
}
/// stop will process all operations which have been pushed onto the stack in the reverse order
/// they were pushed.
pub async fn stop(mut self) -> Result<(), E> {
// reverse wait_group in place, so we stop the most recently added first. Since this method
// consumes self this is fine.
self.wait_group.reverse();
for fut in self.wait_group {
fut.await?;
}
Ok(())
}
}
impl<E> Default for TaskStack<E>
where
E: error::Error + Send + 'static,
{
fn default() -> Self {
Self::new()
}
}

View File

@ -1,6 +1,4 @@
use std::{error, fs, io, path, pin}; use std::{fs, io, path, pin};
use tokio_util::sync::CancellationToken;
pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> { pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
match fs::File::open(path) { match fs::File::open(path) {
@ -15,66 +13,3 @@ pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
pub type BoxByteStream = futures::stream::BoxStream<'static, io::Result<Vec<u8>>>; pub type BoxByteStream = futures::stream::BoxStream<'static, io::Result<Vec<u8>>>;
pub type BoxFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>; pub type BoxFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>;
pub struct TaskStack<E>
where
E: error::Error + Send + 'static,
{
wait_group: Vec<BoxFuture<'static, Result<(), E>>>,
}
impl<E> TaskStack<E>
where
E: error::Error + Send + 'static,
{
pub fn new() -> TaskStack<E> {
TaskStack {
wait_group: Vec::new(),
}
}
/// push adds the given Future to the stack, to be executed once stop is called.
pub fn push<Fut>(&mut self, f: Fut)
where
Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
{
self.wait_group.push(Box::pin(f))
}
/// push_spawn will spawn the given closure in a tokio task. Once the CancellationToken is
/// cancelled the closure is expected to return.
pub fn push_spawn<F, Fut>(&mut self, mut f: F)
where
Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
F: FnMut(CancellationToken) -> Fut,
{
let canceller = CancellationToken::new();
let handle = tokio::spawn(f(canceller.clone()));
self.push(async move {
canceller.cancel();
handle.await.expect("failed to join task")
});
}
/// stop will process all operations which have been pushed onto the stack in the reverse order
/// they were pushed.
pub async fn stop(mut self) -> Result<(), E> {
// reverse wait_group in place, so we stop the most recently added first. Since this method
// consumes self this is fine.
self.wait_group.reverse();
for fut in self.wait_group {
fut.await?;
}
Ok(())
}
}
impl<E> Default for TaskStack<E>
where
E: error::Error + Send + 'static,
{
fn default() -> Self {
Self::new()
}
}