use std::{env, process};
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use log::error;
use hyper::service::service_fn;
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::{runtime, select};
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use tokio_rustls::TlsAcceptor;
use crate::commons::file;
use crate::commons::error::Error;
use crate::commons::version::KrillVersion;
use crate::config::Config;
use crate::constants::KRILL_ENV_UPGRADE_ONLY;
use crate::server::properties::PropertiesManager;
use crate::server::manager::KrillManager;
use crate::upgrades::{
finalise_data_migration, post_start_upgrade,
prepare_upgrade_data_migrations, UpgradeError, UpgradeMode,
};
use super::http::{tls, tls_keys};
use super::http::server::HttpServer;
pub async fn start_krill_daemon(
config: Arc<Config>,
mut signal_running: Option<oneshot::Sender<()>>,
) -> Result<(), Error> {
write_pid_file_or_die(&config);
test_data_dirs_or_die(&config);
let properties_manager = PropertiesManager::create(
&config.storage_uri,
config.use_history_cache,
)?;
let upgrade_report = prepare_upgrade_data_migrations(
UpgradeMode::PrepareToFinalise, &config, &properties_manager
).map_err(|e| {
match e {
UpgradeError::CodeOlderThanData(_,_) => {
Error::Custom(e.to_string())
},
_ => {
Error::Custom(format!(
"Upgrade data migration failed with error: {e}\n\n\
NOTE: your data was not changed. Please downgrade \
your krill instance to your previous version."
))
}
}
})?;
if let Some(report) = &upgrade_report {
finalise_data_migration(
report.versions(), &config, &properties_manager
).map_err(|e| {
Error::Custom(format!(
"Finishing prepared migration failed unexpectedly. Please \
check your data {}. If you find folders named \
'arch-cas-{}' or 'arch-pubd-{}' there, then rename them \
to 'cas' and 'pubd' respectively and re-install krill \
version {}. Underlying error was: {}",
config.storage_uri,
report.versions().from(),
report.versions().from(),
report.versions().from(),
e
))
})?;
}
if !properties_manager.is_initialized() {
properties_manager.init(KrillVersion::code_version())?;
}
let krill = KrillManager::build(config.clone()).await?;
if let Some(report) = upgrade_report {
post_start_upgrade(report, &krill).await?;
}
if env::var(KRILL_ENV_UPGRADE_ONLY).is_ok() {
println!("Krill upgrade successful");
std::process::exit(0);
}
let scheduler = krill.build_scheduler();
let scheduler_future = scheduler.run();
let server = HttpServer::new(
krill, config.clone(), &runtime::Handle::current()
)?;
if config.https_mode().is_generate_https_cert() {
tls_keys::create_key_cert_if_needed(config.tls_keys_dir())
.map_err(|e| Error::HttpsSetup(format!("{e}")))?;
}
let http_server_futures = futures_util::future::select_all(
config.socket_addresses().into_iter().map(|socket_addr| {
tokio::spawn(single_http_listener(
server.clone(),
socket_addr,
config.clone(),
signal_running.take(),
))
}),
);
#[cfg(unix)]
if config.unix_socket_enabled() {
config.unix_socket().map(|path| {
tokio::spawn(single_unix_listener(
server.clone(),
path.clone(),
config.clone(),
signal_running.take(),
))
});
}
select!(
_ = http_server_futures => error!("http server stopped unexpectedly"),
_ = scheduler_future => error!("scheduler stopped unexpectedly"),
);
Err(Error::custom("stopping krill process"))
}
async fn single_http_listener(
server: Arc<HttpServer>,
addr: SocketAddr,
config: Arc<Config>,
signal_running: Option<oneshot::Sender<()>>,
) {
let listener = match TcpListener::bind(addr).await {
Ok(listener) => listener,
Err(err) => {
error!("Could not bind to {addr}: {err}");
return;
}
};
let tls = if config.https_mode().is_disable_https() {
None
} else {
match tls::create_server_config(
&tls_keys::key_file_path(config.tls_keys_dir()),
&tls_keys::cert_file_path(config.tls_keys_dir()),
) {
Ok(config) => Some(TlsAcceptor::from(Arc::new(config))),
Err(err) => {
error!("{err}");
return;
}
}
};
if let Some(tx) = signal_running {
let _ = tx.send(());
}
loop {
let stream = match listener.accept().await {
Ok((stream, _addr)) => {
tls::MaybeTlsTcpStream::new(stream, tls.as_ref())
}
Err(err) => {
error!("Fatal error in HTTP server {addr}: {err}");
return;
}
};
let server = server.clone();
tokio::task::spawn(async move {
let _ = hyper_util::server::conn::auto::Builder::new(
TokioExecutor::new(),
)
.serve_connection(
TokioIo::new(stream),
service_fn(move |req| {
let server = server.clone();
async move { server.process_request(req).await }
}),
)
.await;
});
}
}
#[cfg(unix)]
async fn single_unix_listener(
server: Arc<HttpServer>,
path: std::path::PathBuf,
_config: Arc<Config>,
signal_running: Option<oneshot::Sender<()>>,
) {
use tokio::net::UnixListener;
if path.exists() && let Err(err) = std::fs::remove_file(&path) {
error!("Fatal error in UNIX socket: {err}");
return;
}
let listener = match UnixListener::bind(&path) {
Ok(listener) => listener,
Err(err) => {
error!("Could not bind to {}: {}", &path.to_string_lossy(), err);
return;
}
};
if let Some(tx) = signal_running {
let _ = tx.send(());
}
loop {
use tokio::net::unix;
let (stream, _addr) = match listener.accept().await {
Ok(stream) => stream,
Err(err) => {
error!("Fatal UNIX socket error: {err}");
continue;
}
};
let uid: unix::uid_t = match stream.peer_cred() {
Ok(cred) => cred.uid(),
Err(err) => {
error!("Could not obtain peer credentials: {err}");
continue;
}
};
let user: nix::unistd::User = match nix::unistd::User::from_uid(
nix::unistd::Uid::from_raw(uid)
) {
Ok(Some(user)) => user,
Err(err) => {
error!("Could not obtain user details for UNIX socket: {err}");
continue;
},
_ => {
error!("Could not obtain user details for UNIX socket");
continue;
}
};
let server = server.clone();
tokio::task::spawn(async move {
let _ = hyper_util::server::conn::auto::Builder::new(
TokioExecutor::new(),
)
.serve_connection(
TokioIo::new(stream),
service_fn(move |mut req| {
let extensions = req.extensions_mut();
extensions.insert(user.clone());
let server = server.clone();
async move { server.process_request(req).await }
}),
)
.await;
});
}
}
fn write_pid_file_or_die(config: &Config) {
if let Err(e) = file::save(
process::id().to_string().as_bytes(), config.pid_file()
) {
print_write_error_hint_and_die(format!(
"Could not write PID file: {e}"
));
}
}
fn test_data_dirs_or_die(config: &Config) {
test_data_dir_or_die("tls_keys_dir", config.tls_keys_dir());
test_data_dir_or_die("repo_dir", config.repo_dir());
if let Some(rfc8181_log_dir) = &config.rfc8181_log_dir {
test_data_dir_or_die("rfc8181_log_dir", rfc8181_log_dir);
}
if let Some(rfc6492_log_dir) = &config.rfc6492_log_dir {
test_data_dir_or_die("rfc6492_log_dir", rfc6492_log_dir);
}
}
fn test_data_dir_or_die(config_item: &str, dir: &Path) {
let test_file = dir.join("test");
if let Err(e) = file::save(b"test", &test_file) {
print_write_error_hint_and_die(format!(
"Cannot write to dir '{}' for configuration setting '{}', \
Error: {}",
dir.to_string_lossy(),
config_item,
e
));
}
else if let Err(e) = file::delete_file(&test_file) {
print_write_error_hint_and_die(format!(
"Cannot delete test file '{}' in dir for configuration setting \
'{}', Error: {}",
test_file.to_string_lossy(),
config_item,
e
));
}
}
fn print_write_error_hint_and_die(error_msg: String) {
eprintln!("{error_msg}");
eprintln!();
eprintln!("Hint: if you use systemd you may need to override the allowed");
eprintln!("ReadWritePaths, the easiest way may be by doing ");
eprintln!("'systemctl edit krill' and add a section like:");
eprintln!();
eprintln!("[Service]");
eprintln!("ReadWritePaths=/local/path1 /local/path2 ...");
}