use super::dirs::NO_HOST;
use super::service::{Holder, holder};
use super::{Doctor, Finding, Verdict};
use hennery_host::connection::Standing;
use hennery_host::identity::Paired;
use hennery_host::transport::{CollectorEndpoint, CollectorTransport, CollectorTrust};
use reqwest::Url;
use std::time::Duration;
pub const STEP_TIMEOUT: Duration = Duration::from_secs(5);
pub const SKEW_WARN: u64 = 30;
pub const SKEW_FAIL: u64 = 300;
const MAX_QUOTE: usize = 200;
const NOT_PAIRED: &str = "the host is not paired";
const NO_ANSWER: &str = "the collector did not answer (check 7)";
fn quoted(text: &str) -> String {
let mut out: String = text.chars().take(MAX_QUOTE).collect();
if text.chars().count() > MAX_QUOTE {
out.push('…');
}
out
}
pub fn shown_url(url: &str) -> String {
match Url::parse(url) {
Ok(mut parsed) if !parsed.username().is_empty() || parsed.password().is_some() => {
let _ = parsed.set_username("");
let _ = parsed.set_password(None);
parsed.to_string()
}
Ok(_) => url.to_string(),
Err(_) => "an unreadable URL".to_string(),
}
}
pub const HELLO_TIMEOUT: Duration = Duration::from_secs(35);
fn block_on<T: Send + 'static>(future: impl std::future::Future<Output = T> + Send + 'static) -> T {
std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("a runtime")
.block_on(future)
})
.join()
.expect("the request thread")
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Health {
Answered {
status: u16,
date: Option<i64>,
},
Tls(String),
Failed(String),
}
pub fn healthz(base: &Url) -> Health {
healthz_with(base, CollectorTrust::System)
}
fn healthz_with(base: &Url, trust: CollectorTrust) -> Health {
let transport =
CollectorEndpoint::public(base.as_str()).and_then(|endpoint| CollectorTransport::new(endpoint, trust));
match transport {
Ok(transport) => healthz_transport(&transport),
Err(_) => Health::Failed("collector endpoint or trust configuration failed".into()),
}
}
fn healthz_transport(transport: &CollectorTransport) -> Health {
let transport = transport.clone();
block_on(async move {
let operation = async {
let response = transport.health().await?;
Ok::<_, anyhow::Error>(Health::Answered {
status: response.status().as_u16(),
date: response
.headers()
.get(reqwest::header::DATE)
.and_then(|v| v.to_str().ok())
.and_then(http_date),
})
};
match tokio::time::timeout(STEP_TIMEOUT, operation).await {
Ok(Ok(health)) => health,
Ok(Err(error)) if hennery_host::transport::classify(error.as_ref()).is_tls() => Health::Tls(format!(
"{}",
hennery_host::transport::failure(hennery_host::transport::classify(error.as_ref()))
)),
Ok(Err(_)) => Health::Failed("DNS/TCP/HTTP connection failed".into()),
Err(_) => Health::Failed("collector health deadline exceeded".into()),
}
})
}
pub fn http_date(text: &str) -> Option<i64> {
let parts: Vec<&str> = text.split_whitespace().collect();
let [_, day, month, year, time, "GMT"] = parts.as_slice() else {
return None;
};
let month = [
"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
]
.iter()
.position(|m| m == month)? as i64
+ 1;
let (day, year): (i64, i64) = (day.parse().ok()?, year.parse().ok()?);
let hms: Vec<i64> = time.split(':').map(|p| p.parse().ok()).collect::<Option<_>>()?;
let [h, m, s] = hms.as_slice() else {
return None;
};
let y = if month <= 2 { year - 1 } else { year };
let era = y.div_euclid(400);
let yoe = y - era * 400;
let mp = (month + 9) % 12;
let doy = (153 * mp + 2) / 5 + day - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
let days = era * 146_097 + doe - 719_468;
Some(days * 86_400 + h * 3600 + m * 60 + s)
}
fn collector_of(doctor: &Doctor, paired: &Paired) -> Result<(String, String), String> {
let up = doctor.dirs.collector.as_ref().filter(|_| {
doctor
.dirs
.host
.as_ref()
.is_some_and(|h| h.parent() == Some(doctor.dirs.root.as_path()))
});
if let Some(collector) = up {
let file = crate::config::FileConfig::load(collector).map_err(|e| format!("{e:#}"))?;
let addresses = file
.listen(&[])
.and_then(|given| crate::listen_addresses(&given))
.map_err(|e| format!("{e:#}"))?;
let base = addresses
.iter()
.map(|a| crate::loopback_url(a))
.find(|base| hennery_host::pairing::collector_ws_url(base).is_ok())
.ok_or("no listen address of the collector's is reached over loopback")?;
let ws = hennery_host::pairing::collector_ws_url(&base).map_err(|e| format!("{e:#}"))?;
CollectorEndpoint::stored(&paired.collector_url)
.and_then(|stored| stored.require_compatible(&CollectorEndpoint::stored(&ws)?))
.map_err(|e| e.to_string())?;
return Ok((base, ws));
}
let ws = paired.collector_url.clone();
let mut url = Url::parse(&ws).map_err(|_| "host.toml's collector is not a URL".to_string())?;
let scheme = match url.scheme() {
"wss" => "https",
"ws" => "http",
other => return Err(format!("host.toml's collector has the scheme {other}")),
};
url.set_scheme(scheme).map_err(|()| "a scheme".to_string())?;
url.set_path("/");
Ok((url.to_string(), ws))
}
pub fn collector(doctor: &Doctor) -> [Finding; 2] {
let not_run = |why| [Finding::NotRun { number: 7, why }, Finding::NotRun { number: 8, why }];
let Some(host) = &doctor.dirs.host else {
return not_run(NO_HOST);
};
let mut verdict = Verdict::default();
match hennery_host::enrollment::inspect(host) {
Ok(hennery_host::enrollment::EnrollmentState::Uncertain(attempt)) => {
verdict.fail(attempt.diagnostic(), "check Hosts, revoke the possible enrollment and explicitly archive this attempt; no identity probe was sent");
return [
Finding::Checked(verdict.check(7, "collector")),
Finding::NotRun {
number: 8,
why: NO_ANSWER,
},
];
}
Err(err) => {
verdict.fail(
err.to_string(),
"preserve enrollment artifacts and complete explicit recovery; no identity probe was sent",
);
return [
Finding::Checked(verdict.check(7, "collector")),
Finding::NotRun {
number: 8,
why: NO_ANSWER,
},
];
}
_ => {}
}
let paired = match Paired::read(host) {
Ok(Some(paired)) => paired,
Ok(None) => return not_run(NOT_PAIRED),
Err(err) => {
let fix =
if host.join("host.toml.pending").exists() || host.join(hennery_host::enrollment::JOURNAL).exists() {
"run `hennery host run` once (it finishes the pairing), then doctor again"
} else {
"fix host.toml, or pair this host again (`hennery host join`)"
};
if host.join("host.toml.pending").exists() || host.join(hennery_host::enrollment::JOURNAL).exists() {
verdict.warn(format!("{err}"), fix);
} else {
verdict.fail(format!("{err}"), fix);
}
return [
Finding::Checked(verdict.check(7, "collector")),
Finding::NotRun {
number: 8,
why: NO_ANSWER,
},
];
}
};
let date = reach(doctor, host, &paired, &mut verdict);
let seven = Finding::Checked(verdict.check(7, "collector"));
let eight = match date {
None => Finding::NotRun {
number: 8,
why: NO_ANSWER,
},
Some(date) => Finding::Checked(skew(date, now())),
};
[seven, eight]
}
fn now() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_secs() as i64)
}
pub fn health_shown(shown: &str) -> String {
format!("{}/healthz", shown.trim_end_matches('/'))
}
fn reach(doctor: &Doctor, host: &std::path::Path, paired: &Paired, verdict: &mut Verdict) -> Option<Option<i64>> {
let (base, ws) = match collector_of(doctor, paired) {
Ok(both) => both,
Err(why) => {
verdict.fail(why, "fix the collector's config.toml, or pair this host again");
return None;
}
};
let rejoin = "pair this host again: `hennery host join <the collector's URL>`";
let shown = shown_url(&base);
let health = health_shown(&shown);
let url = match hennery_host::pairing::parse_public_url(&base) {
Ok(url) => url,
Err(err) => {
verdict.fail(format!("{err}").replace(&base, &shown), rejoin);
return None;
}
};
let transport = match CollectorEndpoint::public(url.as_str())
.and_then(|endpoint| CollectorTransport::new(endpoint, paired.trust.clone()))
{
Ok(transport) => transport,
Err(_) => {
verdict.fail(
"collector trust configuration failed",
"install the intended CA or repair native roots; TLS validation is required",
);
return None;
}
};
verdict.ok(format!("trust: {}", paired.trust.source()));
if transport.root_warnings > 0 {
verdict.warn(
format!(
"native root loading had {} failures with usable roots",
transport.root_warnings
),
"inspect native root sources",
);
}
let name = url.host_str().unwrap_or_default().to_string();
let endpoint = transport.endpoint.clone();
let addresses = match block_on(async move {
tokio::time::timeout(STEP_TIMEOUT, endpoint.addresses())
.await
.unwrap_or_else(|_| Err(hennery_host::transport::failure(hennery_host::transport::Stage::Dns)))
}) {
Ok(addresses) if !addresses.is_empty() => addresses,
Ok(_) => {
verdict.fail(
format!("{name} does not resolve"),
"check the name in DNS, or the collector's URL in host.toml",
);
return None;
}
Err(_) => {
verdict.fail(
format!("{name} did not resolve within {STEP_TIMEOUT:?}"),
"check this machine's DNS",
);
return None;
}
};
if let Err(err) = connect(&addresses) {
verdict.fail(
format!("nothing answers on {} ({})", addresses[0], err.kind()),
"start the collector, or open the way to it (firewall, VPN)",
);
return None;
}
let date = match healthz_transport(&transport) {
Health::Tls(why) => {
verdict.fail(
format!("{shown}: TLS failed: {why}"),
"give the collector a certificate for its name (and check this machine's clock: an expired-looking certificate can be the clock)",
);
return None;
}
Health::Failed(why) => {
verdict.fail(format!("{health}: {why}"), "check the collector's log");
return None;
}
Health::Answered { status, .. } if status != 200 => {
verdict.fail(
format!("{health} answered {status}"),
"check that this URL is the collector's, and its log",
);
return None;
}
Health::Answered { date, .. } => date,
};
verdict.ok(format!("{shown} answers"));
let _ = ws;
hello(doctor, host, paired, &transport, verdict);
Some(date)
}
fn connect(addresses: &[std::net::SocketAddr]) -> std::io::Result<()> {
let addresses = addresses.to_vec();
block_on(async move {
tokio::time::timeout(STEP_TIMEOUT, async move {
let mut first = None;
for address in addresses {
match tokio::net::TcpStream::connect(address).await {
Ok(_) => return Ok(()),
Err(err) => {
first.get_or_insert(err);
}
}
}
Err(first.unwrap_or_else(|| std::io::ErrorKind::NotFound.into()))
})
.await
.unwrap_or_else(|_| Err(std::io::ErrorKind::TimedOut.into()))
})
}
fn hello(
doctor: &Doctor,
host: &std::path::Path,
paired: &Paired,
transport: &CollectorTransport,
verdict: &mut Verdict,
) {
let check = transport.clone();
if let Err(error) = block_on(async move { check.preflight().await }) {
verdict.fail(
format!("WSS upgrade/nonce check failed: {error}"),
"forward the WebSocket upgrade and hennery-hello-nonce; remove ingress login redirects",
);
return;
}
if holder(doctor, &host.join(crate::lock::HOST_LOCK)) != Holder::Nobody {
verdict.ok("WSS upgrade/nonce checked; authentication not checked: hello not sent because a host may be connected on this directory (see check 14)");
return;
}
let (transport, id, key) = (transport.clone(), paired.host_id.clone(), paired.key.clone());
let standing = block_on(async move {
tokio::time::timeout(
HELLO_TIMEOUT,
hennery_host::connection::probe_transport(&transport, &id, &key),
)
.await
});
let rejoin = "pair it again: remove host.key and host.toml, then `hennery host join <url>`";
match standing {
Err(_) => verdict.fail(
format!("its hello did not finish within {HELLO_TIMEOUT:?}"),
"check the collector's log",
),
Ok(Ok(Standing::Accepted)) => verdict.ok(
"its hello is accepted (the collector records this host as seen now, with this binary's version)",
),
Ok(Ok(Standing::Connected)) => verdict.warn(
"this host's key is in use by another live connection, and no host here holds host.lock",
"run doctor again in a minute; if it stays and no other copy of this directory is yours, revoke the host in the Hosts view and pair again (`hennery host join`)",
),
Ok(Ok(Standing::Revoked)) => verdict.fail(
"the collector revoked this host",
"re-pair it with `hennery host join <url>` (it gets a new key and id)",
),
Ok(Ok(Standing::Unknown)) => verdict.fail("the collector does not know this host or its key", rejoin),
Ok(Err(err)) => verdict.fail(
format!("its hello failed: {}", quoted(&err.to_string())),
"check the collector's log",
),
}
}
pub fn skew(date: Option<i64>, now: i64) -> super::Check {
let mut verdict = Verdict::default();
match date {
None => verdict.warn(
"the collector's answer had no Date",
"check this machine's clock by hand (`date`)",
),
Some(date) => {
let off = now.abs_diff(date);
let fix = "set this machine's clock from the network (NTP)";
if off > SKEW_FAIL {
verdict.fail(format!("this clock is {off} s off the collector's"), fix);
} else if off > SKEW_WARN {
verdict.warn(format!("this clock is {off} s off the collector's"), fix);
} else {
verdict.ok(format!("within {off} s of the collector's"));
}
}
}
verdict.check(8, "clock")
}
fn served(doctor: &Doctor, dir: &std::path::Path) -> bool {
use crate::service::unit::Role;
let cx = doctor.cx;
let [role] = cx.installed()[..] else {
return false;
};
let Some(data) = crate::service::read_command_line(cx, role).and_then(|a| crate::service::data_dir_of(&a)) else {
return false;
};
let collector = match role {
Role::Up => data.join("collector"),
Role::Collector => data,
Role::Host => return false,
};
collector.canonicalize().ok() == dir.canonicalize().ok()
&& crate::service::managed(cx, role)
.ok()
.flatten()
.is_some_and(|m| m.running)
}
pub fn listeners(doctor: &Doctor) -> Finding {
let Some(collector) = &doctor.dirs.collector else {
return Finding::NotRun {
number: 16,
why: "no collector data directory",
};
};
let mut verdict = Verdict::default();
let file = match crate::config::FileConfig::load(collector) {
Ok(file) => file,
Err(err) => {
verdict.fail(format!("{err}"), "fix the collector's config.toml");
return Finding::Checked(verdict.check(16, "listeners"));
}
};
let addresses = match file.listen(&[]).and_then(|given| crate::listen_addresses(&given)) {
Ok(addresses) => addresses,
Err(err) => {
verdict.fail(format!("{err}"), "fix listen in the collector's config.toml");
return Finding::Checked(verdict.check(16, "listeners"));
}
};
let running = served(doctor, collector);
for address in &addresses {
let answered = crate::healthcheck::probe_address(address)
.and_then(|a| crate::healthcheck::check(a, crate::healthcheck::TIMEOUT));
match answered {
Ok(()) => verdict.ok(format!("{address} answers")),
Err(_) if running => verdict.fail(
format!("{address}: nothing answers, though the service runs"),
"check listen in config.toml against the collector's log (a taken port fails its start)",
),
Err(_) => verdict.warn(
format!("{address}: no collector answers (is it running?)"),
"start it (`hennery up`, `hennery collector`, or the service)",
),
}
}
if doctor.cx.env.contains_key("HENNERY_LISTEN") {
verdict.ok("HENNERY_LISTEN is set in this shell: a service does not see it");
}
match file.public_url(None) {
None => verdict.ok("no public_url in config.toml (the one stored at setup is not read)"),
Some(public) => match hennery_host::pairing::parse_public_url(&public) {
Err(err) => verdict.warn(
format!("{err}").replace(&public, &shown_url(&public)),
"fix public_url in config.toml",
),
Ok(url) => match if doctor.collector_trust == CollectorTrust::System {
healthz(&url)
} else {
healthz_with(&url, doctor.collector_trust.clone())
} {
Health::Answered { status: 200, .. } => {
verdict.ok(format!("public_url {} answers /healthz", shown_url(&public)))
}
Health::Answered { status, .. } => verdict.warn(
format!("public_url {} answers /healthz with {status}", shown_url(&public)),
"point public_url, or the proxy in front, at this collector",
),
Health::Tls(why) | Health::Failed(why) => verdict.warn(
format!("public_url {} reaches no listener or proxy: {why}", shown_url(&public)),
"point public_url, or the proxy in front, at this collector",
),
},
},
}
let configured = file.proxy_policy().expect("validated config").is_configured();
let attribution = format!(
"proxy-address attribution is {}; missing header uses the TCP peer; health success cannot prove header stripping or separate limiter budgets",
if configured { "configured" } else { "unconfigured" }
);
if configured
|| file
.public_url(None)
.is_some_and(|url| Url::parse(&url).is_ok_and(|url| url.scheme() == "https"))
{
verdict.warn(attribution, "verify the exact peer and overwritten Hennery-Client-IP header with two clients; restrict ingress until the public deployment gate passes");
} else {
verdict.ok(attribution);
}
verdict.ok(format!("external collector trust: {}", doctor.collector_trust.source()));
Finding::Checked(verdict.check(16, "listeners"))
}