micropelago/domani
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: micropelago:f2374cded5e3a67d181c309522cc3649672af524
Branches Tags
micropelago:main
..
compare: micropelago:420f1ff42ae970fa610fc0200b3f2106320aa4f7
Branches Tags
micropelago:main

No commits in common. "f2374cded5e3a67d181c309522cc3649672af524" and "420f1ff42ae970fa610fc0200b3f2106320aa4f7" have entirely different histories.
f2374cded5 ... 420f1ff42a

24 changed files with 887 additions and 848 deletions
Show Stats Expand all files Collapse all files

4
TODO
View File

@ -1,3 +1 @@
- make domain_manager implement rusttls cert resolver
- Try to switch from Arc to Box where possible
- maybe build TaskSet into some kind of defer-like replacement
- clean up main a lot

1
src/domain/acme.rs
View File

@ -1,5 +1,4 @@
pub mod manager;
pub mod resolver;
pub mod store;
mod private_key;

53
src/domain/acme/manager.rs
View File

@ -7,24 +7,39 @@ const LETS_ENCRYPT_URL: &str = "https://acme-v02.api.letsencrypt.org/directory";
pub type GetHttp01ChallengeKeyError = acme::store::GetHttp01ChallengeKeyError;
#[mockall::automock]
pub trait Manager: Sync + Send {
fn sync_domain<'mgr>(
&'mgr self,
domain: domain::Name,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>;
#[mockall::automock(
type SyncDomainFuture=future::Ready<Result<(), unexpected::Error>>;
)]
pub trait Manager {
type SyncDomainFuture<'mgr>: future::Future<Output = Result<(), unexpected::Error>>
+ Send
+ Unpin
+ 'mgr
where
Self: 'mgr;
fn sync_domain(&self, domain: domain::Name) -> Self::SyncDomainFuture<'_>;
fn get_http01_challenge_key(&self, token: &str) -> Result<String, GetHttp01ChallengeKeyError>;
}
struct ManagerImpl {
store: sync::Arc<dyn acme::store::Store>,
pub trait BoxedManager: Manager + Send + Sync + Clone + 'static {}
struct ManagerImpl<Store>
where
Store: acme::store::BoxedStore,
{
store: Store,
account: sync::Arc<acme2::Account>,
}
pub async fn new(
store: sync::Arc<dyn acme::store::Store>,
pub async fn new<Store>(
store: Store,
contact_email: &str,
) -> Result<sync::Arc<dyn Manager>, unexpected::Error> {
) -> Result<impl BoxedManager, unexpected::Error>
where
Store: acme::store::BoxedStore,
{
let dir = acme2::DirectoryBuilder::new(LETS_ENCRYPT_URL.to_string())
.build()
.await
@ -66,12 +81,16 @@ pub async fn new(
Ok(sync::Arc::new(ManagerImpl { store, account }))
}
impl Manager for ManagerImpl {
fn sync_domain<'mgr>(
&'mgr self,
domain: domain::Name,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>
{
impl<Store> BoxedManager for sync::Arc<ManagerImpl<Store>> where Store: acme::store::BoxedStore {}
impl<Store> Manager for sync::Arc<ManagerImpl<Store>>
where
Store: acme::store::BoxedStore,
{
type SyncDomainFuture<'mgr> = pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>
where Self: 'mgr;
fn sync_domain(&self, domain: domain::Name) -> Self::SyncDomainFuture<'_> {
Box::pin(async move {
// if there's an existing cert, and its expiry (determined by the soonest value of
// not_after amongst its parts) is later than 30 days from now, then we consider it to be

44
src/domain/acme/resolver.rs
View File

@ -1,44 +0,0 @@
use crate::domain::acme::store;
use crate::error::unexpected::Mappable;
use std::sync;
struct CertResolver(sync::Arc<dyn store::Store>);
pub fn new(
store: sync::Arc<dyn store::Store>,
) -> sync::Arc<dyn rustls::server::ResolvesServerCert> {
return sync::Arc::new(CertResolver(store));
}
impl rustls::server::ResolvesServerCert for CertResolver {
fn resolve(
&self,
client_hello: rustls::server::ClientHello<'_>,
) -> Option<sync::Arc<rustls::sign::CertifiedKey>> {
let domain = client_hello.server_name()?;
match self.0.get_certificate(domain) {
Err(store::GetCertificateError::NotFound) => {
log::warn!("No cert found for domain {domain}");
Ok(None)
}
Err(store::GetCertificateError::Unexpected(err)) => Err(err),
Ok((key, cert)) => {
match rustls::sign::any_supported_type(&key.into()).or_unexpected() {
Err(err) => Err(err),
Ok(key) => Ok(Some(sync::Arc::new(rustls::sign::CertifiedKey {
cert: cert.into_iter().map(|cert| cert.into()).collect(),
key,
ocsp: None,
sct_list: None,
}))),
}
}
}
.unwrap_or_else(|err| {
log::error!("Unexpected error getting cert for domain {domain}: {err}");
None
})
}
}

59
src/domain/acme/store.rs
View File

@ -38,7 +38,7 @@ pub enum GetCertificateError {
}
#[mockall::automock]
pub trait Store: Sync + Send {
pub trait Store {
fn set_account_key(&self, k: &PrivateKey) -> Result<(), unexpected::Error>;
fn get_account_key(&self) -> Result<PrivateKey, GetAccountKeyError>;
@ -60,6 +60,8 @@ pub trait Store: Sync + Send {
) -> Result<(PrivateKey, Vec<Certificate>), GetCertificateError>;
}
pub trait BoxedStore: Store + Send + Sync + Clone + 'static {}
#[derive(Debug, Serialize, Deserialize)]
struct StoredPKeyCert {
private_key: PrivateKey,
@ -70,7 +72,12 @@ struct FSStore {
dir_path: path::PathBuf,
}
pub fn new(dir_path: &path::Path) -> Result<sync::Arc<dyn Store>, unexpected::Error> {
#[derive(Clone)]
struct BoxedFSStore(sync::Arc<FSStore>);
pub fn new(
dir_path: &path::Path,
) -> Result<impl BoxedStore + rustls::server::ResolvesServerCert, unexpected::Error> {
vec![
dir_path,
dir_path.join("http01_challenge_keys").as_ref(),
@ -82,14 +89,14 @@ pub fn new(dir_path: &path::Path) -> Result<sync::Arc<dyn Store>, unexpected::Er
})
.try_collect()?;
Ok(sync::Arc::new(FSStore {
Ok(BoxedFSStore(sync::Arc::new(FSStore {
dir_path: dir_path.into(),
}))
})))
}
impl FSStore {
impl BoxedFSStore {
fn account_key_path(&self) -> path::PathBuf {
self.dir_path.join("account.key")
self.0.dir_path.join("account.key")
}
fn http01_challenge_key_path(&self, token: &str) -> path::PathBuf {
@ -99,18 +106,20 @@ impl FSStore {
.expect("token successfully hashed");
let n = h.finalize().encode_hex::<String>();
self.dir_path.join("http01_challenge_keys").join(n)
self.0.dir_path.join("http01_challenge_keys").join(n)
}
fn certificate_path(&self, domain: &str) -> path::PathBuf {
let mut domain = domain.to_string();
domain.push_str(".json");
self.dir_path.join("certificates").join(domain)
self.0.dir_path.join("certificates").join(domain)
}
}
impl Store for FSStore {
impl BoxedStore for BoxedFSStore {}
impl Store for BoxedFSStore {
fn set_account_key(&self, k: &PrivateKey) -> Result<(), unexpected::Error> {
let path = self.account_key_path();
{
@ -223,6 +232,38 @@ impl Store for FSStore {
}
}
impl rustls::server::ResolvesServerCert for BoxedFSStore {
fn resolve(
&self,
client_hello: rustls::server::ClientHello<'_>,
) -> Option<sync::Arc<rustls::sign::CertifiedKey>> {
let domain = client_hello.server_name()?;
match self.get_certificate(domain) {
Err(GetCertificateError::NotFound) => {
log::warn!("No cert found for domain {domain}");
Ok(None)
}
Err(GetCertificateError::Unexpected(err)) => Err(err),
Ok((key, cert)) => {
match rustls::sign::any_supported_type(&key.into()).or_unexpected() {
Err(err) => Err(err),
Ok(key) => Ok(Some(sync::Arc::new(rustls::sign::CertifiedKey {
cert: cert.into_iter().map(|cert| cert.into()).collect(),
key,
ocsp: None,
sct_list: None,
}))),
}
}
}
.unwrap_or_else(|err| {
log::error!("Unexpected error getting cert for domain {domain}: {err}");
None
})
}
}
#[cfg(test)]
mod tests {
use super::*;

23
src/domain/config.rs
View File

@ -38,18 +38,23 @@ pub enum SetError {
Unexpected(#[from] unexpected::Error),
}
/// Used in the return from all_domains from Store.
pub type AllDomainsResult<T> = Result<T, unexpected::Error>;
#[mockall::automock]
pub trait Store: Sync + Send {
pub trait Store {
fn get(&self, domain: &domain::Name) -> Result<Config, GetError>;
fn set(&self, domain: &domain::Name, config: &Config) -> Result<(), SetError>;
fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error>;
fn all_domains(&self) -> AllDomainsResult<Vec<AllDomainsResult<domain::Name>>>;
}
pub trait BoxedStore: Store + Send + Sync + Clone {}
struct FSStore {
dir_path: PathBuf,
}
pub fn new(dir_path: &Path) -> io::Result<sync::Arc<dyn Store>> {
pub fn new(dir_path: &Path) -> io::Result<impl BoxedStore> {
fs::create_dir_all(dir_path)?;
Ok(sync::Arc::new(FSStore {
dir_path: dir_path.into(),
@ -66,7 +71,9 @@ impl FSStore {
}
}
impl Store for FSStore {
impl BoxedStore for sync::Arc<FSStore> {}
impl Store for sync::Arc<FSStore> {
fn get(&self, domain: &domain::Name) -> Result<Config, GetError> {
let path = self.config_file_path(domain);
let config_file = fs::File::open(path.as_path()).map_err(|e| match e.kind() {
@ -96,11 +103,11 @@ impl Store for FSStore {
Ok(())
}
fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error> {
fs::read_dir(&self.dir_path)
fn all_domains(&self) -> AllDomainsResult<Vec<AllDomainsResult<domain::Name>>> {
Ok(fs::read_dir(&self.dir_path)
.or_unexpected()?
.map(
|dir_entry_res: io::Result<fs::DirEntry>| -> Result<domain::Name, unexpected::Error> {
|dir_entry_res: io::Result<fs::DirEntry>| -> AllDomainsResult<domain::Name> {
let domain = dir_entry_res.or_unexpected()?.file_name();
let domain = domain.to_str().ok_or(unexpected::Error::from(
"couldn't convert os string to &str",
@ -110,7 +117,7 @@ impl Store for FSStore {
.map_unexpected_while(|| format!("parsing {domain} as domain name"))
},
)
.try_collect()
.collect())
}
}

206
src/domain/manager.rs
View File

@ -1,9 +1,8 @@
use crate::domain::{self, acme, checker, config};
use crate::error::unexpected::{self, Mappable};
use crate::error::unexpected::{self, Intoable, Mappable};
use crate::origin;
use std::{future, pin, sync};
use tokio_util::sync::CancellationToken;
#[derive(thiserror::Error, Debug)]
pub enum GetConfigError {
@ -116,109 +115,112 @@ impl From<config::SetError> for SyncWithConfigError {
pub type GetAcmeHttp01ChallengeKeyError = acme::manager::GetHttp01ChallengeKeyError;
#[mockall::automock]
pub trait Manager: Sync + Send {
pub type AllDomainsResult<T> = config::AllDomainsResult<T>;
#[mockall::automock(
type Origin=origin::MockOrigin;
type SyncWithConfigFuture=future::Ready<Result<(), SyncWithConfigError>>;
type SyncAllOriginsErrorsIter=Vec<unexpected::Error>;
)]
pub trait Manager {
type Origin<'mgr>: origin::Origin + 'mgr
where
Self: 'mgr;
type SyncWithConfigFuture<'mgr>: future::Future<Output = Result<(), SyncWithConfigError>>
+ Send
+ Unpin
+ 'mgr
where
Self: 'mgr;
type SyncAllOriginsErrorsIter<'mgr>: IntoIterator<Item = unexpected::Error> + 'mgr
where
Self: 'mgr;
fn get_config(&self, domain: &domain::Name) -> Result<config::Config, GetConfigError>;
fn get_origin(
fn get_origin(&self, domain: &domain::Name) -> Result<Self::Origin<'_>, GetOriginError>;
fn sync_with_config(
&self,
domain: &domain::Name,
) -> Result<sync::Arc<dyn origin::Origin>, GetOriginError>;
fn sync_cert<'mgr>(
&'mgr self,
domain: domain::Name,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>;
fn sync_with_config<'mgr>(
&'mgr self,
domain: domain::Name,
config: config::Config,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>;
) -> Self::SyncWithConfigFuture<'_>;
fn sync_all_origins(&self) -> Result<Self::SyncAllOriginsErrorsIter<'_>, unexpected::Error>;
fn get_acme_http01_challenge_key(
&self,
token: &str,
) -> Result<String, GetAcmeHttp01ChallengeKeyError>;
fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error>;
fn all_domains(&self) -> AllDomainsResult<Vec<AllDomainsResult<domain::Name>>>;
}
pub struct ManagerImpl {
origin_store: sync::Arc<dyn origin::store::Store>,
domain_config_store: sync::Arc<dyn config::Store>,
pub trait BoxedManager: Manager + Send + Sync + Clone {}
struct ManagerImpl<OriginStore, DomainConfigStore, AcmeManager>
where
OriginStore: origin::store::BoxedStore,
DomainConfigStore: config::BoxedStore,
AcmeManager: acme::manager::BoxedManager,
{
origin_store: OriginStore,
domain_config_store: DomainConfigStore,
domain_checker: checker::DNSChecker,
acme_manager: Option<sync::Arc<dyn acme::manager::Manager>>,
canceller: CancellationToken,
origin_sync_handler: tokio::task::JoinHandle<()>,
acme_manager: Option<AcmeManager>,
}
fn sync_origins(origin_store: &dyn origin::store::Store) {
match origin_store.all_descrs() {
Ok(iter) => iter.into_iter(),
Err(err) => {
log::error!("Error fetching origin descriptors: {err}");
return;
}
}
.for_each(|descr| {
if let Err(err) = origin_store.sync(descr.clone(), origin::store::Limits {}) {
log::error!("Failed to sync store for {:?}: {err}", descr);
return;
}
});
}
pub fn new(
origin_store: sync::Arc<dyn origin::store::Store>,
domain_config_store: sync::Arc<dyn config::Store>,
pub fn new<OriginStore, DomainConfigStore, AcmeManager>(
origin_store: OriginStore,
domain_config_store: DomainConfigStore,
domain_checker: checker::DNSChecker,
acme_manager: Option<sync::Arc<dyn acme::manager::Manager>>,
) -> ManagerImpl {
let canceller = CancellationToken::new();
let origin_sync_handler = {
let origin_store = origin_store.clone();
let canceller = canceller.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(20 * 60));
loop {
tokio::select! {
_ = interval.tick() => sync_origins(origin_store.as_ref()),
_ = canceller.cancelled() => return,
}
}
})
};
ManagerImpl {
acme_manager: Option<AcmeManager>,
) -> impl BoxedManager
where
OriginStore: origin::store::BoxedStore,
DomainConfigStore: config::BoxedStore,
AcmeManager: acme::manager::BoxedManager,
{
sync::Arc::new(ManagerImpl {
origin_store,
domain_config_store,
domain_checker,
acme_manager,
canceller,
origin_sync_handler,
}
})
}
impl ManagerImpl {
pub fn stop(self) -> tokio::task::JoinHandle<()> {
self.canceller.cancel();
self.origin_sync_handler
}
impl<OriginStore, DomainConfigStore, AcmeManager> BoxedManager
for sync::Arc<ManagerImpl<OriginStore, DomainConfigStore, AcmeManager>>
where
OriginStore: origin::store::BoxedStore,
DomainConfigStore: config::BoxedStore,
AcmeManager: acme::manager::BoxedManager,
{
}
impl Manager for ManagerImpl {
impl<OriginStore, DomainConfigStore, AcmeManager> Manager
for sync::Arc<ManagerImpl<OriginStore, DomainConfigStore, AcmeManager>>
where
OriginStore: origin::store::BoxedStore,
DomainConfigStore: config::BoxedStore,
AcmeManager: acme::manager::BoxedManager,
{
type Origin<'mgr> = OriginStore::Origin<'mgr>
where Self: 'mgr;
type SyncWithConfigFuture<'mgr> = pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>
where Self: 'mgr;
type SyncAllOriginsErrorsIter<'mgr> = Box<dyn Iterator<Item = unexpected::Error> + 'mgr>
where Self: 'mgr;
fn get_config(&self, domain: &domain::Name) -> Result<config::Config, GetConfigError> {
Ok(self.domain_config_store.get(domain)?)
}
fn get_origin(
&self,
domain: &domain::Name,
) -> Result<sync::Arc<dyn origin::Origin>, GetOriginError> {
fn get_origin(&self, domain: &domain::Name) -> Result<Self::Origin<'_>, GetOriginError> {
let config = self.domain_config_store.get(domain)?;
let origin = self
.origin_store
@ -228,26 +230,11 @@ impl Manager for ManagerImpl {
Ok(origin)
}
fn sync_cert<'mgr>(
&'mgr self,
domain: domain::Name,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>
{
Box::pin(async move {
if let Some(ref acme_manager) = self.acme_manager {
acme_manager.sync_domain(domain.clone()).await?;
}
Ok(())
})
}
fn sync_with_config<'mgr>(
&'mgr self,
fn sync_with_config(
&self,
domain: domain::Name,
config: config::Config,
) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>
{
) -> Self::SyncWithConfigFuture<'_> {
Box::pin(async move {
let config_hash = config
.hash()
@ -262,12 +249,39 @@ impl Manager for ManagerImpl {
self.domain_config_store.set(&domain, &config)?;
self.sync_cert(domain).await?;
if let Some(ref acme_manager) = self.acme_manager {
acme_manager.sync_domain(domain.clone()).await?;
}
Ok(())
})
}
fn sync_all_origins(&self) -> Result<Self::SyncAllOriginsErrorsIter<'_>, unexpected::Error> {
let iter = self
.origin_store
.all_descrs()
.or_unexpected_while("fetching all origin descrs")?
.into_iter();
Ok(Box::from(iter.filter_map(|descr| {
if let Err(err) = descr {
return Some(err.into_unexpected());
}
let descr = descr.unwrap();
if let Err(err) = self
.origin_store
.sync(descr.clone(), origin::store::Limits {})
{
return Some(err.into_unexpected_while(format!("syncing store {:?}", descr)));
}
None
})))
}
fn get_acme_http01_challenge_key(
&self,
token: &str,
@ -279,7 +293,7 @@ impl Manager for ManagerImpl {
Err(GetAcmeHttp01ChallengeKeyError::NotFound)
}
fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error> {
fn all_domains(&self) -> AllDomainsResult<Vec<AllDomainsResult<domain::Name>>> {
self.domain_config_store.all_domains()
}
}

1
src/lib.rs
View File

@ -1,4 +1,3 @@
#![feature(result_option_inspect)]
#![feature(iterator_try_collect)]
pub mod domain;

232
src/main.rs
View File

@ -1,10 +1,19 @@
#![feature(result_option_inspect)]
use clap::Parser;
use futures::stream::futures_unordered::FuturesUnordered;
use futures::stream::StreamExt;
use signal_hook_tokio::Signals;
use tokio::select;
use tokio::time;
use std::convert::Infallible;
use std::net::SocketAddr;
use std::str::FromStr;
use std::{path, sync};
use std::{future, path, sync};
use domiply::domain::acme::manager::Manager as AcmeManager;
use domiply::domain::manager::Manager;
#[derive(Parser, Debug)]
#[command(version)]
@ -59,10 +68,14 @@ struct Cli {
}
#[derive(Clone)]
struct HTTPSParams {
struct HTTPSParams<DomainAcmeStore, DomainAcmeManager>
where
DomainAcmeStore: domiply::domain::acme::store::BoxedStore,
DomainAcmeManager: domiply::domain::acme::manager::BoxedManager,
{
https_listen_addr: SocketAddr,
domain_acme_store: sync::Arc<dyn domiply::domain::acme::store::Store>,
domain_acme_manager: sync::Arc<dyn domiply::domain::acme::manager::Manager>,
domain_acme_store: DomainAcmeStore,
domain_acme_manager: DomainAcmeManager,
}
#[tokio::main]
@ -78,6 +91,7 @@ async fn main() {
)
.init();
let mut wait_group = FuturesUnordered::new();
let canceller = tokio_util::sync::CancellationToken::new();
{
@ -145,31 +159,201 @@ async fn main() {
https_params.as_ref().map(|p| p.domain_acme_manager.clone()),
);
let domain_manager = sync::Arc::new(domain_manager);
wait_group.push({
let domain_manager = domain_manager.clone();
let canceller = canceller.clone();
{
let (http_service, http_service_task_set) = domiply::service::http::new(
domain_manager.clone(),
config.domain_checker_target_a,
config.passphrase,
config.http_listen_addr.clone(),
config.http_domain.clone(),
https_params.map(|p| domiply::service::http::HTTPSParams {
listen_addr: p.https_listen_addr,
cert_resolver: domiply::domain::acme::resolver::new(p.domain_acme_store),
}),
);
tokio::spawn(async move {
let mut interval = time::interval(time::Duration::from_secs(20 * 60));
canceller.cancelled().await;
loop {
select! {
_ = interval.tick() => (),
_ = canceller.cancelled() => return,
}
domiply::service::http::stop(http_service, http_service_task_set).await;
let errors_iter = domain_manager.sync_all_origins();
if let Err(err) = errors_iter {
log::error!("Got error calling sync_all_origins: {err}");
continue;
}
errors_iter
.unwrap()
.into_iter()
.for_each(|err| log::error!("syncing failed: {err}"));
}
})
});
let service = domiply::service::new(
domain_manager.clone(),
config.domain_checker_target_a,
config.passphrase,
config.http_domain.clone(),
);
let service = sync::Arc::new(service);
wait_group.push({
let http_domain = config.http_domain.clone();
let canceller = canceller.clone();
let service = service.clone();
let make_service = hyper::service::make_service_fn(move |_| {
let service = service.clone();
// Create a `Service` for responding to the request.
let service = hyper::service::service_fn(move |req| {
domiply::service::handle_request(service.clone(), req)
});
// Return the service to hyper.
async move { Ok::<_, Infallible>(service) }
});
tokio::spawn(async move {
let addr = config.http_listen_addr;
log::info!(
"Listening on http://{}:{}",
http_domain.as_str(),
addr.port()
);
let server = hyper::Server::bind(&addr).serve(make_service);
let graceful = server.with_graceful_shutdown(async {
canceller.cancelled().await;
});
if let Err(e) = graceful.await {
panic!("server error: {}", e);
};
})
});
if let Some(https_params) = https_params {
// Periodically refresh all domain certs, including the http_domain passed in the Cli opts
wait_group.push({
let https_params = https_params.clone();
let domain_manager = domain_manager.clone();
let http_domain = config.http_domain.clone();
let canceller = canceller.clone();
tokio::spawn(async move {
let mut interval = time::interval(time::Duration::from_secs(60 * 60));
loop {
select! {
_ = interval.tick() => (),
_ = canceller.cancelled() => return,
}
_ = https_params
.domain_acme_manager
.sync_domain(http_domain.clone())
.await
.inspect_err(|err| {
log::error!(
"Error while getting cert for {}: {err}",
http_domain.as_str()
)
});
let domains_iter = domain_manager.all_domains();
if let Err(err) = domains_iter {
log::error!("Got error calling all_domains: {err}");
continue;
}
for domain in domains_iter.unwrap().into_iter() {
match domain {
Ok(domain) => {
let _ = https_params
.domain_acme_manager
.sync_domain(domain.clone())
.await
.inspect_err(|err| {
log::error!(
"Error while getting cert for {}: {err}",
domain.as_str(),
)
});
}
Err(err) => log::error!("Error iterating through domains: {err}"),
};
}
}
})
});
// HTTPS server
wait_group.push({
let https_params = https_params;
let http_domain = config.http_domain.clone();
let canceller = canceller.clone();
let service = service.clone();
let make_service = hyper::service::make_service_fn(move |_| {
let service = service.clone();
// Create a `Service` for responding to the request.
let service = hyper::service::service_fn(move |req| {
domiply::service::handle_request(service.clone(), req)
});
// Return the service to hyper.
async move { Ok::<_, Infallible>(service) }
});
tokio::spawn(async move {
let canceller = canceller.clone();
let server_config: tokio_rustls::TlsAcceptor = sync::Arc::new(
rustls::server::ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth()
.with_cert_resolver(sync::Arc::from(https_params.domain_acme_store)),
)
.into();
let addr = https_params.https_listen_addr;
let addr_incoming = hyper::server::conn::AddrIncoming::bind(&addr)
.expect("https listen socket creation failed");
let incoming =
tls_listener::TlsListener::new(server_config, addr_incoming).filter(|conn| {
if let Err(err) = conn {
log::error!("Error accepting TLS connection: {:?}", err);
future::ready(false)
} else {
future::ready(true)
}
});
let incoming = hyper::server::accept::from_stream(incoming);
log::info!(
"Listening on https://{}:{}",
http_domain.as_str(),
addr.port()
);
let server = hyper::Server::builder(incoming).serve(make_service);
let graceful = server.with_graceful_shutdown(async {
canceller.cancelled().await;
});
if let Err(e) = graceful.await {
panic!("server error: {}", e);
};
})
})
}
sync::Arc::into_inner(domain_manager)
.unwrap()
.stop()
.await
.expect("domain manager failed to shutdown cleanly");
while wait_group.next().await.is_some() {}
log::info!("Graceful shutdown complete");
}

25
src/origin/store.rs
View File

@ -1,6 +1,5 @@
use crate::error::unexpected;
use crate::origin;
use std::sync;
pub mod git;
@ -39,13 +38,29 @@ pub enum AllDescrsError {
Unexpected(#[from] unexpected::Error),
}
#[mockall::automock]
/// Used in the return from all_descrs from Store.
pub type AllDescrsResult<T> = Result<T, AllDescrsError>;
#[mockall::automock(
type Origin=origin::MockOrigin;
type AllDescrsIter=Vec<AllDescrsResult<origin::Descr>>;
)]
/// Describes a storage mechanism for Origins. Each Origin is uniquely identified by its Descr.
pub trait Store: Sync + Send {
pub trait Store {
type Origin<'store>: origin::Origin + 'store
where
Self: 'store;
type AllDescrsIter<'store>: IntoIterator<Item = AllDescrsResult<origin::Descr>> + 'store
where
Self: 'store;
/// If the origin is of a kind which can be updated, sync will pull down the latest version of
/// the origin into the storage.
fn sync(&self, descr: origin::Descr, limits: Limits) -> Result<(), SyncError>;
fn get(&self, descr: origin::Descr) -> Result<sync::Arc<dyn origin::Origin>, GetError>;
fn all_descrs(&self) -> Result<Vec<origin::Descr>, AllDescrsError>;
fn get(&self, descr: origin::Descr) -> Result<Self::Origin<'_>, GetError>;
fn all_descrs(&self) -> AllDescrsResult<Self::AllDescrsIter<'_>>;
}
pub trait BoxedStore: Store + Send + Sync + Clone {}

42
src/origin/store/git.rs
View File

@ -4,14 +4,13 @@ use crate::origin::{self, store};
use std::path::{Path, PathBuf};
use std::{collections, fs, io, sync};
#[derive(Clone)]
struct Origin {
descr: origin::Descr,
repo: sync::Arc<gix::ThreadSafeRepository>,
repo: gix::ThreadSafeRepository,
tree_object_id: gix::ObjectId,
}
impl origin::Origin for Origin {
impl origin::Origin for sync::Arc<Origin> {
fn descr(&self) -> &origin::Descr {
&self.descr
}
@ -71,7 +70,7 @@ struct Store {
origins: sync::RwLock<collections::HashMap<origin::Descr, sync::Arc<Origin>>>,
}
pub fn new(dir_path: PathBuf) -> io::Result<sync::Arc<dyn super::Store>> {
pub fn new(dir_path: PathBuf) -> io::Result<impl super::BoxedStore> {
fs::create_dir_all(&dir_path)?;
Ok(sync::Arc::new(Store {
dir_path,
@ -97,7 +96,7 @@ impl Store {
&self,
repo: gix::Repository,
descr: origin::Descr,
) -> Result<Origin, GetOriginError> {
) -> Result<sync::Arc<Origin>, GetOriginError> {
let origin::Descr::Git {
ref branch_name, ..
} = descr;
@ -119,11 +118,11 @@ impl Store {
.map_unexpected_while(|| format!("parsing {commit_object_id} as commit"))?
.tree();
Ok(Origin {
Ok(sync::Arc::from(Origin {
descr,
repo: sync::Arc::new(repo.into()),
repo: repo.into(),
tree_object_id,
})
}))
}
fn sync_inner(
@ -208,7 +207,15 @@ impl Store {
}
}
impl super::Store for Store {
impl super::BoxedStore for sync::Arc<Store> {}
impl super::Store for sync::Arc<Store> {
type Origin<'store> = sync::Arc<Origin>
where Self: 'store;
type AllDescrsIter<'store> = Box<dyn Iterator<Item = store::AllDescrsResult<origin::Descr>> + 'store>
where Self: 'store;
fn sync(&self, descr: origin::Descr, limits: store::Limits) -> Result<(), store::SyncError> {
// attempt to lock this descr for syncing, doing so within a new scope so the mutex
// isn't actually being held for the whole method duration.
@ -248,12 +255,12 @@ impl super::Store for Store {
})?;
let mut origins = self.origins.write().unwrap();
(*origins).insert(descr, sync::Arc::new(origin));
(*origins).insert(descr, origin);
Ok(())
}
fn get(&self, descr: origin::Descr) -> Result<sync::Arc<dyn origin::Origin>, store::GetError> {
fn get(&self, descr: origin::Descr) -> Result<Self::Origin<'_>, store::GetError> {
{
let origins = self.origins.read().unwrap();
if let Some(origin) = origins.get(&descr) {
@ -280,18 +287,16 @@ impl super::Store for Store {
GetOriginError::Unexpected(e) => store::GetError::Unexpected(e),
})?;
let origin = sync::Arc::new(origin.clone());
let mut origins = self.origins.write().unwrap();
(*origins).insert(descr, origin.clone());
Ok(origin)
}
fn all_descrs(&self) -> Result<Vec<origin::Descr>, store::AllDescrsError> {
fs::read_dir(&self.dir_path).or_unexpected()?.map(
|dir_entry_res: io::Result<fs::DirEntry>| -> Result<origin::Descr, store::AllDescrsError> {
fn all_descrs(&self) -> store::AllDescrsResult<Self::AllDescrsIter<'_>> {
Ok(Box::from(
fs::read_dir(&self.dir_path).or_unexpected()?.map(
|dir_entry_res: io::Result<fs::DirEntry>| -> store::AllDescrsResult<origin::Descr> {
let descr_id: String = dir_entry_res
.or_unexpected()?
.file_name()
@ -317,7 +322,8 @@ impl super::Store for Store {
Ok(descr)
},
).try_collect()
),
))
}
}

421
src/service.rs
View File

@ -1,2 +1,421 @@
pub mod http;
use hyper::{Body, Method, Request, Response};
use serde::{Deserialize, Serialize};
use std::convert::Infallible;
use std::future::Future;
use std::net;
use std::str::FromStr;
use std::sync;
use crate::origin::Origin;
use crate::{domain, origin};
pub mod http_tpl;
mod util;
type SvcResponse = Result<Response<hyper::body::Body>, String>;
#[derive(Clone)]
pub struct Service<'svc, DomainManager>
where
DomainManager: domain::manager::BoxedManager,
{
domain_manager: DomainManager,
target_a: net::Ipv4Addr,
passphrase: String,
http_domain: domain::Name,
handlebars: handlebars::Handlebars<'svc>,
}
pub fn new<'svc, DomainManager>(
domain_manager: DomainManager,
target_a: net::Ipv4Addr,
passphrase: String,
http_domain: domain::Name,
) -> Service<'svc, DomainManager>
where
DomainManager: domain::manager::BoxedManager,
{
Service {
domain_manager,
target_a,
passphrase,
http_domain,
handlebars: self::http_tpl::get(),
}
}
#[derive(Serialize)]
struct BasePresenter<'a, T> {
page_name: &'a str,
data: T,
}
#[derive(Deserialize)]
struct DomainGetArgs {
domain: domain::Name,
}
#[derive(Deserialize)]
struct DomainInitArgs {
domain: domain::Name,
}
#[derive(Deserialize)]
struct DomainSyncArgs {
domain: domain::Name,
passphrase: String,
}
impl<'svc, DomainManager> Service<'svc, DomainManager>
where
DomainManager: domain::manager::BoxedManager,
{
fn serve_string(&self, status_code: u16, path: &'_ str, body: Vec<u8>) -> SvcResponse {
let content_type = mime_guess::from_path(path)
.first_or_octet_stream()
.to_string();
match Response::builder()
.status(status_code)
.header("Content-Type", content_type)
.body(body.into())
{
Ok(res) => Ok(res),
Err(err) => Err(format!("failed to build {}: {}", path, err)),
}
}
//// TODO make this use an io::Write, rather than SvcResponse
fn render<T>(&self, status_code: u16, name: &'_ str, value: T) -> SvcResponse
where
T: Serialize,
{
let rendered = match self.handlebars.render(name, &value) {
Ok(res) => res,
Err(handlebars::RenderError {
template_name: None,
..
}) => return self.render_error_page(404, "Static asset not found"),
Err(err) => {
return self.render_error_page(500, format!("template error: {err}").as_str())
}
};
self.serve_string(status_code, name, rendered.into())
}
fn render_error_page(&'svc self, status_code: u16, e: &'_ str) -> SvcResponse {
#[derive(Serialize)]
struct Response<'a> {
error_msg: &'a str,
}
self.render(
status_code,
"/base.html",
BasePresenter {
page_name: "/error.html",
data: &Response { error_msg: e },
},
)
}
fn render_page<T>(&self, name: &'_ str, data: T) -> SvcResponse
where
T: Serialize,
{
self.render(
200,
"/base.html",
BasePresenter {
page_name: name,
data,
},
)
}
fn serve_origin(&self, domain: domain::Name, path: &'_ str) -> SvcResponse {
let mut path_owned;
let path = match path.ends_with('/') {
true => {
path_owned = String::from(path);
path_owned.push_str("index.html");
path_owned.as_str()
}
false => path,
};
let origin = match self.domain_manager.get_origin(&domain) {
Ok(o) => o,
Err(domain::manager::GetOriginError::NotFound) => {
return self.render_error_page(404, "Domain not found")
}
Err(domain::manager::GetOriginError::Unexpected(e)) => {
return self.render_error_page(500, format!("failed to fetch origin: {e}").as_str())
}
};
let mut buf = Vec::<u8>::new();
match origin.read_file_into(path, &mut buf) {
Ok(_) => self.serve_string(200, path, buf),
Err(origin::ReadFileIntoError::FileNotFound) => {
self.render_error_page(404, "File not found")
}
Err(origin::ReadFileIntoError::Unexpected(e)) => {
self.render_error_page(500, format!("failed to fetch file {path}: {e}").as_str())
}
}
}
async fn with_query_req<'a, F, In, Out>(&self, req: &'a Request<Body>, f: F) -> SvcResponse
where
In: Deserialize<'a>,
F: FnOnce(In) -> Out,
Out: Future<Output = SvcResponse>,
{
let query = req.uri().query().unwrap_or("");
match serde_urlencoded::from_str::<In>(query) {
Ok(args) => f(args).await,
Err(err) => Err(format!("failed to parse query args: {}", err)),
}
}
fn domain_get(&self, args: DomainGetArgs) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domain: domain::Name,
config: Option<domain::config::Config>,
}
let config = match self.domain_manager.get_config(&args.domain) {
Ok(config) => Some(config),
Err(domain::manager::GetConfigError::NotFound) => None,
Err(domain::manager::GetConfigError::Unexpected(e)) => {
return self
.render_error_page(500, format!("retrieving configuration: {}", e).as_str());
}
};
self.render_page(
"/domain.html",
Response {
domain: args.domain,
config,
},
)
}
fn domain_init(&self, args: DomainInitArgs, domain_config: util::FlatConfig) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domain: domain::Name,
flat_config: util::FlatConfig,
target_a: net::Ipv4Addr,
challenge_token: String,
}
let config: domain::config::Config = match domain_config.try_into() {
Ok(Some(config)) => config,
Ok(None) => return self.render_error_page(400, "domain config is required"),
Err(e) => {
return self.render_error_page(400, format!("invalid domain config: {e}").as_str())
}
};
let config_hash = match config.hash() {
Ok(hash) => hash,
Err(e) => {
return self
.render_error_page(500, format!("failed to hash domain config: {e}").as_str())
}
};
self.render_page(
"/domain_init.html",
Response {
domain: args.domain,
flat_config: config.into(),
target_a: self.target_a,
challenge_token: config_hash,
},
)
}
async fn domain_sync(
&self,
args: DomainSyncArgs,
domain_config: util::FlatConfig,
) -> SvcResponse {
if args.passphrase != self.passphrase.as_str() {
return self.render_error_page(401, "Incorrect passphrase");
}
let config: domain::config::Config = match domain_config.try_into() {
Ok(Some(config)) => config,
Ok(None) => return self.render_error_page(400, "domain config is required"),
Err(e) => {
return self.render_error_page(400, format!("invalid domain config: {e}").as_str())
}
};
let sync_result = self
.domain_manager
.sync_with_config(args.domain.clone(), config)
.await;
#[derive(Serialize)]
struct Response {
domain: domain::Name,
error_msg: Option<String>,
}
let error_msg = match sync_result {
Ok(_) => None,
Err(domain::manager::SyncWithConfigError::InvalidURL) => Some("Fetching the git repository failed, please double check that you input the correct URL.".to_string()),
Err(domain::manager::SyncWithConfigError::InvalidBranchName) => Some("The git repository does not have a branch of the given name, please double check that you input the correct name.".to_string()),
Err(domain::manager::SyncWithConfigError::AlreadyInProgress) => Some("The configuration of your domain is still in progress, please refresh in a few minutes.".to_string()),
Err(domain::manager::SyncWithConfigError::TargetANotSet) => Some("The A record is not set correctly on the domain. Please double check that you put the correct value on the record. If the value is correct, then most likely the updated records have not yet propagated. In this case you can refresh in a few minutes to try again.".to_string()),
Err(domain::manager::SyncWithConfigError::ChallengeTokenNotSet) => Some("The TXT record is not set correctly on the domain. Please double check that you put the correct value on the record. If the value is correct, then most likely the updated records have not yet propagated. In this case you can refresh in a few minutes to try again.".to_string()),
Err(domain::manager::SyncWithConfigError::Unexpected(e)) => Some(format!("An unexpected error occurred: {e}")),
};
let response = Response {
domain: args.domain,
error_msg,
};
self.render_page("/domain_sync.html", response)
}
pub fn domains(&self) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domains: Vec<String>,
}
let domains = match self.domain_manager.all_domains() {
Ok(domains) => domains,
Err(e) => {
return self.render_error_page(500, format!("failed get all domains: {e}").as_str())
}
};
let domains: Vec<domain::Name> = match domains.into_iter().try_collect() {
Ok(domains) => domains,
Err(e) => {
return self.render_error_page(500, format!("failed get all domains: {e}").as_str())
}
};
let mut domains: Vec<String> = domains
.into_iter()
.map(|domain| domain.as_str().to_string())
.collect();
domains.sort();
self.render_page("/domains.html", Response { domains })
}
}
pub async fn handle_request<DomainManager>(
svc: sync::Arc<Service<'_, DomainManager>>,
req: Request<Body>,
) -> Result<Response<Body>, Infallible>
where
DomainManager: domain::manager::BoxedManager,
{
match handle_request_inner(svc, req).await {
Ok(res) => Ok(res),
Err(err) => panic!("unexpected error {err}"),
}
}
fn strip_port(host: &str) -> &str {
match host.rfind(':') {
None => host,
Some(i) => &host[..i],
}
}
pub async fn handle_request_inner<DomainManager>(
svc: sync::Arc<Service<'_, DomainManager>>,
req: Request<Body>,
) -> SvcResponse
where
DomainManager: domain::manager::BoxedManager,
{
let maybe_host = match (
req.headers()
.get("Host")
.and_then(|v| v.to_str().ok())
.map(strip_port),
req.uri().host().map(strip_port),
) {
(Some(h), _) if h != svc.http_domain.as_str() => Some(h),
(_, Some(h)) if h != svc.http_domain.as_str() => Some(h),
_ => None,
}
.and_then(|h| domain::Name::from_str(h).ok());
let method = req.method();
let path = req.uri().path();
// Serving acme challenges always takes priority. We serve them from the same store no matter
// the domain, presumably they are cryptographically random enough that it doesn't matter.
if method == Method::GET && path.starts_with("/.well-known/acme-challenge/") {
let token = path.trim_start_matches("/.well-known/acme-challenge/");
if let Ok(key) = svc.domain_manager.get_acme_http01_challenge_key(token) {
let body: hyper::Body = key.into();
return match Response::builder().status(200).body(body) {
Ok(res) => Ok(res),
Err(err) => Err(format!(
"failed to write acme http-01 challenge key: {}",
err
)),
};
}
}
// If a managed domain was given then serve that from its origin
if let Some(domain) = maybe_host {
return svc.serve_origin(domain, req.uri().path());
}
// Serve main domiply site
if method == Method::GET && path.starts_with("/static/") {
return svc.render(200, path, ());
}
match (method, path) {
(&Method::GET, "/") | (&Method::GET, "/index.html") => svc.render_page("/index.html", ()),
(&Method::GET, "/domain.html") => {
svc.with_query_req(&req, |args: DomainGetArgs| async { svc.domain_get(args) })
.await
}
(&Method::GET, "/domain_init.html") => {
svc.with_query_req(&req, |args: DomainInitArgs| async {
svc.with_query_req(&req, |config: util::FlatConfig| async {
svc.domain_init(args, config)
})
.await
})
.await
}
(&Method::GET, "/domain_sync.html") => {
svc.with_query_req(&req, |args: DomainSyncArgs| async {
svc.with_query_req(&req, |config: util::FlatConfig| async {
svc.domain_sync(args, config).await
})
.await
})
.await
}
(&Method::GET, "/domains.html") => svc.domains(),
_ => svc.render_error_page(404, "Page not found!"),
}
}

441
src/service/http.rs
View File

@ -1,441 +0,0 @@
mod tasks;
mod tpl;
use hyper::{Body, Method, Request, Response};
use serde::{Deserialize, Serialize};
use std::convert::Infallible;
use std::str::FromStr;
use std::{future, net, sync};
use crate::error::unexpected;
use crate::{domain, origin, service, util};
type SvcResponse = Result<Response<hyper::body::Body>, String>;
pub struct Service {
domain_manager: sync::Arc<dyn domain::manager::Manager>,
target_a: net::Ipv4Addr,
passphrase: String,
http_domain: domain::Name,
handlebars: handlebars::Handlebars<'static>,
}
pub struct HTTPSParams {
pub listen_addr: net::SocketAddr,
pub cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
}
pub fn new(
domain_manager: sync::Arc<dyn domain::manager::Manager>,
target_a: net::Ipv4Addr,
passphrase: String,
http_listen_addr: net::SocketAddr,
http_domain: domain::Name,
https_params: Option<HTTPSParams>,
) -> (sync::Arc<Service>, util::TaskSet<unexpected::Error>) {
let service = sync::Arc::new(Service {
domain_manager: domain_manager.clone(),
target_a,
passphrase,
http_domain: http_domain.clone(),
handlebars: tpl::get(),
});
let task_set = util::TaskSet::new();
task_set.spawn(|canceller| {
tasks::listen_http(
service.clone(),
canceller,
http_listen_addr,
http_domain.clone(),
)
});
if let Some(https_params) = https_params {
task_set.spawn(|canceller| {
tasks::listen_https(
service.clone(),
canceller,
https_params.cert_resolver.clone(),
https_params.listen_addr,
http_domain.clone(),
)
});
task_set.spawn(|canceller| {
tasks::cert_refresher(domain_manager.clone(), canceller, http_domain.clone())
});
}
return (service, task_set);
}
pub async fn stop(service: sync::Arc<Service>, task_set: util::TaskSet<unexpected::Error>) {
task_set
.stop()
.await
.iter()
.for_each(|e| log::error!("error while shutting down http service: {e}"));
sync::Arc::into_inner(service).expect("service didn't get cleaned up");
}
#[derive(Serialize)]
struct BasePresenter<'a, T> {
page_name: &'a str,
data: T,
}
#[derive(Deserialize)]
struct DomainGetArgs {
domain: domain::Name,
}
#[derive(Deserialize)]
struct DomainInitArgs {
domain: domain::Name,
}
#[derive(Deserialize)]
struct DomainSyncArgs {
domain: domain::Name,
passphrase: String,
}
impl<'svc> Service {
fn serve_string(&self, status_code: u16, path: &'_ str, body: Vec<u8>) -> SvcResponse {
let content_type = mime_guess::from_path(path)
.first_or_octet_stream()
.to_string();
match Response::builder()
.status(status_code)
.header("Content-Type", content_type)
.body(body.into())
{
Ok(res) => Ok(res),
Err(err) => Err(format!("failed to build {}: {}", path, err)),
}
}
//// TODO make this use an io::Write, rather than SvcResponse
fn render<T>(&self, status_code: u16, name: &'_ str, value: T) -> SvcResponse
where
T: Serialize,
{
let rendered = match self.handlebars.render(name, &value) {
Ok(res) => res,
Err(handlebars::RenderError {
template_name: None,
..
}) => return self.render_error_page(404, "Static asset not found"),
Err(err) => {
return self.render_error_page(500, format!("template error: {err}").as_str())
}
};
self.serve_string(status_code, name, rendered.into())
}
fn render_error_page(&'svc self, status_code: u16, e: &'_ str) -> SvcResponse {
#[derive(Serialize)]
struct Response<'a> {
error_msg: &'a str,
}
self.render(
status_code,
"/base.html",
BasePresenter {
page_name: "/error.html",
data: &Response { error_msg: e },
},
)
}
fn render_page<T>(&self, name: &'_ str, data: T) -> SvcResponse
where
T: Serialize,
{
self.render(
200,
"/base.html",
BasePresenter {
page_name: name,
data,
},
)
}
fn serve_origin(&self, domain: domain::Name, path: &'_ str) -> SvcResponse {
let mut path_owned;
let path = match path.ends_with('/') {
true => {
path_owned = String::from(path);
path_owned.push_str("index.html");
path_owned.as_str()
}
false => path,
};
let origin = match self.domain_manager.get_origin(&domain) {
Ok(o) => o,
Err(domain::manager::GetOriginError::NotFound) => {
return self.render_error_page(404, "Domain not found")
}
Err(domain::manager::GetOriginError::Unexpected(e)) => {
return self.render_error_page(500, format!("failed to fetch origin: {e}").as_str())
}
};
let mut buf = Vec::<u8>::new();
match origin.read_file_into(path, &mut buf) {
Ok(_) => self.serve_string(200, path, buf),
Err(origin::ReadFileIntoError::FileNotFound) => {
self.render_error_page(404, "File not found")
}
Err(origin::ReadFileIntoError::Unexpected(e)) => {
self.render_error_page(500, format!("failed to fetch file {path}: {e}").as_str())
}
}
}
async fn with_query_req<'a, F, In, Out>(&self, req: &'a Request<Body>, f: F) -> SvcResponse
where
In: Deserialize<'a>,
F: FnOnce(In) -> Out,
Out: future::Future<Output = SvcResponse>,
{
let query = req.uri().query().unwrap_or("");
match serde_urlencoded::from_str::<In>(query) {
Ok(args) => f(args).await,
Err(err) => Err(format!("failed to parse query args: {}", err)),
}
}
fn domain_get(&self, args: DomainGetArgs) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domain: domain::Name,
config: Option<domain::config::Config>,
}
let config = match self.domain_manager.get_config(&args.domain) {
Ok(config) => Some(config),
Err(domain::manager::GetConfigError::NotFound) => None,
Err(domain::manager::GetConfigError::Unexpected(e)) => {
return self
.render_error_page(500, format!("retrieving configuration: {}", e).as_str());
}
};
self.render_page(
"/domain.html",
Response {
domain: args.domain,
config,
},
)
}
fn domain_init(
&self,
args: DomainInitArgs,
domain_config: service::util::FlatConfig,
) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domain: domain::Name,
flat_config: service::util::FlatConfig,
target_a: net::Ipv4Addr,
challenge_token: String,
}
let config: domain::config::Config = match domain_config.try_into() {
Ok(Some(config)) => config,
Ok(None) => return self.render_error_page(400, "domain config is required"),
Err(e) => {
return self.render_error_page(400, format!("invalid domain config: {e}").as_str())
}
};
let config_hash = match config.hash() {
Ok(hash) => hash,
Err(e) => {
return self
.render_error_page(500, format!("failed to hash domain config: {e}").as_str())
}
};
self.render_page(
"/domain_init.html",
Response {
domain: args.domain,
flat_config: config.into(),
target_a: self.target_a,
challenge_token: config_hash,
},
)
}
async fn domain_sync(
&self,
args: DomainSyncArgs,
domain_config: service::util::FlatConfig,
) -> SvcResponse {
if args.passphrase != self.passphrase.as_str() {
return self.render_error_page(401, "Incorrect passphrase");
}
let config: domain::config::Config = match domain_config.try_into() {
Ok(Some(config)) => config,
Ok(None) => return self.render_error_page(400, "domain config is required"),
Err(e) => {
return self.render_error_page(400, format!("invalid domain config: {e}").as_str())
}
};
let sync_result = self
.domain_manager
.sync_with_config(args.domain.clone(), config)
.await;
#[derive(Serialize)]
struct Response {
domain: domain::Name,
error_msg: Option<String>,
}
let error_msg = match sync_result {
Ok(_) => None,
Err(domain::manager::SyncWithConfigError::InvalidURL) => Some("Fetching the git repository failed, please double check that you input the correct URL.".to_string()),
Err(domain::manager::SyncWithConfigError::InvalidBranchName) => Some("The git repository does not have a branch of the given name, please double check that you input the correct name.".to_string()),
Err(domain::manager::SyncWithConfigError::AlreadyInProgress) => Some("The configuration of your domain is still in progress, please refresh in a few minutes.".to_string()),
Err(domain::manager::SyncWithConfigError::TargetANotSet) => Some("The A record is not set correctly on the domain. Please double check that you put the correct value on the record. If the value is correct, then most likely the updated records have not yet propagated. In this case you can refresh in a few minutes to try again.".to_string()),
Err(domain::manager::SyncWithConfigError::ChallengeTokenNotSet) => Some("The TXT record is not set correctly on the domain. Please double check that you put the correct value on the record. If the value is correct, then most likely the updated records have not yet propagated. In this case you can refresh in a few minutes to try again.".to_string()),
Err(domain::manager::SyncWithConfigError::Unexpected(e)) => Some(format!("An unexpected error occurred: {e}")),
};
let response = Response {
domain: args.domain,
error_msg,
};
self.render_page("/domain_sync.html", response)
}
fn domains(&self) -> SvcResponse {
#[derive(Serialize)]
struct Response {
domains: Vec<String>,
}
let domains = match self.domain_manager.all_domains() {
Ok(domains) => domains,
Err(e) => {
return self.render_error_page(500, format!("failed get all domains: {e}").as_str())
}
};
let mut domains: Vec<String> = domains
.into_iter()
.map(|domain| domain.as_str().to_string())
.collect();
domains.sort();
self.render_page("/domains.html", Response { domains })
}
async fn handle_request_inner(&self, req: Request<Body>) -> SvcResponse {
let maybe_host = match (
req.headers()
.get("Host")
.and_then(|v| v.to_str().ok())
.map(strip_port),
req.uri().host().map(strip_port),
) {
(Some(h), _) if h != self.http_domain.as_str() => Some(h),
(_, Some(h)) if h != self.http_domain.as_str() => Some(h),
_ => None,
}
.and_then(|h| domain::Name::from_str(h).ok());
let method = req.method();
let path = req.uri().path();
// Serving acme challenges always takes priority. We serve them from the same store no matter
// the domain, presumably they are cryptographically random enough that it doesn't matter.
if method == Method::GET && path.starts_with("/.well-known/acme-challenge/") {
let token = path.trim_start_matches("/.well-known/acme-challenge/");
if let Ok(key) = self.domain_manager.get_acme_http01_challenge_key(token) {
let body: hyper::Body = key.into();
return match Response::builder().status(200).body(body) {
Ok(res) => Ok(res),
Err(err) => Err(format!(
"failed to write acme http-01 challenge key: {}",
err
)),
};
}
}
// If a managed domain was given then serve that from its origin
if let Some(domain) = maybe_host {
return self.serve_origin(domain, req.uri().path());
}
// Serve main domiply site
if method == Method::GET && path.starts_with("/static/") {
return self.render(200, path, ());
}
match (method, path) {
(&Method::GET, "/") | (&Method::GET, "/index.html") => {
self.render_page("/index.html", ())
}
(&Method::GET, "/domain.html") => {
self.with_query_req(&req, |args: DomainGetArgs| async { self.domain_get(args) })
.await
}
(&Method::GET, "/domain_init.html") => {
self.with_query_req(&req, |args: DomainInitArgs| async {
self.with_query_req(&req, |config: service::util::FlatConfig| async {
self.domain_init(args, config)
})
.await
})
.await
}
(&Method::GET, "/domain_sync.html") => {
self.with_query_req(&req, |args: DomainSyncArgs| async {
self.with_query_req(&req, |config: service::util::FlatConfig| async {
self.domain_sync(args, config).await
})
.await
})
.await
}
(&Method::GET, "/domains.html") => self.domains(),
_ => self.render_error_page(404, "Page not found!"),
}
}
pub async fn handle_request(&self, req: Request<Body>) -> Result<Response<Body>, Infallible> {
match self.handle_request_inner(req).await {
Ok(res) => Ok(res),
Err(err) => panic!("unexpected error {err}"),
}
}
}
fn strip_port(host: &str) -> &str {
match host.rfind(':') {
None => host,
Some(i) => &host[..i],
}
}

132
src/service/http/tasks.rs
View File

@ -1,132 +0,0 @@
use crate::error::unexpected::{self, Mappable};
use crate::{domain, service};
use std::{convert, future, net, sync};
use futures::StreamExt;
use tokio_util::sync::CancellationToken;
pub async fn listen_http(
service: sync::Arc<service::http::Service>,
canceller: CancellationToken,
addr: net::SocketAddr,
domain: domain::Name,
) -> Result<(), unexpected::Error> {
let make_service = hyper::service::make_service_fn(move |_| {
let service = service.clone();
// Create a `Service` for responding to the request.
let hyper_service = hyper::service::service_fn(move |req| {
let service = service.clone();
async move { service.handle_request(req).await }
});
// Return the service to hyper.
async move { Ok::<_, convert::Infallible>(hyper_service) }
});
log::info!("Listening on http://{}:{}", domain.as_str(), addr.port());
let server = hyper::Server::bind(&addr).serve(make_service);
let graceful = server.with_graceful_shutdown(async {
canceller.cancelled().await;
});
graceful.await.or_unexpected()
}
pub async fn listen_https(
service: sync::Arc<service::http::Service>,
canceller: CancellationToken,
cert_resolver: sync::Arc<dyn rustls::server::ResolvesServerCert>,
addr: net::SocketAddr,
domain: domain::Name,
) -> Result<(), unexpected::Error> {
let make_service = hyper::service::make_service_fn(move |_| {
let service = service.clone();
// Create a `Service` for responding to the request.
let hyper_service = hyper::service::service_fn(move |req| {
let service = service.clone();
async move { service.handle_request(req).await }
});
// Return the service to hyper.
async move { Ok::<_, convert::Infallible>(hyper_service) }
});
let server_config: tokio_rustls::TlsAcceptor = sync::Arc::new(
rustls::server::ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth()
.with_cert_resolver(cert_resolver),
)
.into();
let addr_incoming = hyper::server::conn::AddrIncoming::bind(&addr)
.expect("https listen socket creation failed");
let incoming = tls_listener::TlsListener::new(server_config, addr_incoming).filter(|conn| {
if let Err(err) = conn {
log::error!("Error accepting TLS connection: {:?}", err);
future::ready(false)
} else {
future::ready(true)
}
});
let incoming = hyper::server::accept::from_stream(incoming);
log::info!("Listening on https://{}:{}", domain.as_str(), addr.port());
let server = hyper::Server::builder(incoming).serve(make_service);
let graceful = server.with_graceful_shutdown(async {
canceller.cancelled().await;
});
graceful.await.or_unexpected()
}
pub async fn cert_refresher(
domain_manager: sync::Arc<dyn domain::manager::Manager>,
canceller: CancellationToken,
http_domain: domain::Name,
) -> Result<(), unexpected::Error> {
use tokio::time;
let mut interval = time::interval(time::Duration::from_secs(60 * 60));
loop {
tokio::select! {
_ = interval.tick() => (),
_ = canceller.cancelled() => return Ok(()),
}
_ = domain_manager
.sync_cert(http_domain.clone())
.await
.inspect_err(|err| {
log::error!(
"Error while getting cert for {}: {err}",
http_domain.as_str()
)
});
let domains_iter = domain_manager.all_domains();
if let Err(err) = domains_iter {
log::error!("Got error calling all_domains: {err}");
continue;
}
for domain in domains_iter.unwrap().into_iter() {
let _ = domain_manager
.sync_cert(domain.clone())
.await
.inspect_err(|err| {
log::error!("Error while getting cert for {}: {err}", domain.as_str(),)
});
}
}
}

4
src/service/http/tpl.rs → src/service/http_tpl.rs
View File

@ -1,11 +1,11 @@
use handlebars::Handlebars;
#[derive(rust_embed::RustEmbed)]
#[folder = "src/service/http/tpl"]
#[folder = "src/service/http_tpl"]
#[prefix = "/"]
struct Dir;
pub fn get() -> Handlebars<'static> {
pub fn get<'hbs>() -> Handlebars<'hbs> {
let mut reg = Handlebars::new();
reg.register_embed_templates::<Dir>()
.expect("registered embedded templates");

0
src/service/http/tpl/base.html → src/service/http_tpl/base.html
View File

0
src/service/http/tpl/domain.html → src/service/http_tpl/domain.html
View File

0
src/service/http/tpl/domain_init.html → src/service/http_tpl/domain_init.html
View File

0
src/service/http/tpl/domain_sync.html → src/service/http_tpl/domain_sync.html
View File

0
src/service/http/tpl/domains.html → src/service/http_tpl/domains.html
View File

0
src/service/http/tpl/error.html → src/service/http_tpl/error.html
View File

0
src/service/http/tpl/index.html → src/service/http_tpl/index.html
View File

0
src/service/http/tpl/static/bamboo.css → src/service/http_tpl/static/bamboo.css
View File

47
src/util.rs
View File

@ -1,7 +1,4 @@
use std::{error, fs, io, path};
use futures::stream::futures_unordered::FuturesUnordered;
use tokio_util::sync::CancellationToken;
use std::{fs, io, path};
pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
match fs::File::open(path) {
@ -12,45 +9,3 @@ pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
},
}
}
pub struct TaskSet<E>
where
E: error::Error + Send + 'static,
{
canceller: CancellationToken,
wait_group: FuturesUnordered<tokio::task::JoinHandle<Result<(), E>>>,
}
impl<E> TaskSet<E>
where
E: error::Error + Send + 'static,
{
pub fn new() -> TaskSet<E> {
TaskSet {
canceller: CancellationToken::new(),
wait_group: FuturesUnordered::new(),
}
}
pub fn spawn<F, Fut>(&self, mut f: F)
where
Fut: futures::Future<Output = Result<(), E>> + Send + 'static,
F: FnMut(CancellationToken) -> Fut,
{
let canceller = self.canceller.clone();
let handle = tokio::spawn(f(canceller));
self.wait_group.push(handle);
}
pub async fn stop(self) -> Vec<E> {
self.canceller.cancel();
let mut res = Vec::new();
for f in self.wait_group {
if let Err(err) = f.await.expect("task failed") {
res.push(err);
}
}
res
}
}