crates/estate-clock/src/fetch.rs
//! Asking every configured source, in parallel, and — with `adjust` — acting on
//! the answer. The one entry point both machines call is [`sync`].
use std::time::{Duration, SystemTime};
use crate::config::{Config, Source};
use crate::decide::{Poll, Sample};
use crate::report::{Outcome, Report, Trigger};
/// How long one HTTPS source gets, connect and read together.
pub const HTTPS_TIMEOUT: Duration = Duration::from_secs(4);
/// Roots an HTTPS source may ALSO be vouched for by, beside the public web
/// roots. Set once per process, by the caller that knows the site.
static EXTRA_ROOTS: std::sync::OnceLock<Vec<Vec<u8>>> = std::sync::OnceLock::new();
/// **Trust these roots (DER) too, for HTTPS sources**, for the rest of this
/// process. The public web roots stay; this ADDS. `false` when roots were
/// already set (the first caller decides).
///
/// Why (2026-09-26, owner decision "LOCAL clock"): the bench's front takes its
/// time from a clock the laptop serves on the mock network, signed by the
/// medium's installer CA — never from api.stripe.com or cloudflare on a mock
/// run. The front passes that CA. It is no new trust: the same CA already
/// authenticates the installer that feeds the front everything it runs with.
pub fn trust_also(roots_der: Vec<Vec<u8>>) -> bool {
EXTRA_ROOTS.set(roots_der).is_ok()
}
/// The client TLS configuration: the public web roots plus `extra`.
///
/// ureq 3 (2026-10-03) takes no `rustls::ClientConfig`, so the roots are handed
/// over as certificates: with nothing added, ureq's own web roots (the Mozilla
/// set, as before); with roots added, the same Mozilla set as certificates
/// (`webpki-root-certs`) plus each added root. An added root that is not a
/// certificate is refused here, by name, as before — ureq would skip it
/// silently. Provider: ring (ureq's `rustls` feature), as before.
pub fn tls_with(extra: &[Vec<u8>]) -> Result<ureq::tls::TlsConfig, String> {
let builder = ureq::tls::TlsConfig::builder().provider(ureq::tls::TlsProvider::Rustls);
if extra.is_empty() {
return Ok(builder.root_certs(ureq::tls::RootCerts::WebPki).build());
}
let mut certs: Vec<ureq::tls::Certificate<'static>> = webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.map(|c| ureq::tls::Certificate::from_der(c.as_ref()).to_owned())
.collect();
for der in extra {
rustls::RootCertStore::empty()
.add(rustls::pki_types::CertificateDer::from(der.clone()))
.map_err(|e| format!("an added clock root is not a certificate: {e}"))?;
certs.push(ureq::tls::Certificate::from_der(der).to_owned());
}
Ok(builder.root_certs(ureq::tls::RootCerts::new_with_certs(&certs)).build())
}
/// **The agent every HTTPS source is asked with**, `tls` its roots.
///
/// ureq 3 (2026-10-03), restated to ureq 2's behaviour: a 4xx/5xx is a response
/// (it still carries the `Date`), no proxy taken from the environment, at most 5
/// redirects, the credential header never forwarded on a redirect, and
/// [`HTTPS_TIMEOUT`] for the connect and for the whole request.
pub fn agent_with(tls: ureq::tls::TlsConfig) -> ureq::Agent {
ureq::Agent::config_builder()
.redirect_auth_headers(ureq::config::RedirectAuthHeaders::Never)
.timeout_connect(Some(HTTPS_TIMEOUT))
.timeout_global(Some(HTTPS_TIMEOUT))
.http_status_as_error(false)
.proxy(None)
.max_redirects(5)
.tls_config(tls)
.build()
.new_agent()
}
/// A ureq transport error as text that never carries the URL: ureq 3's
/// `Display` prints the request URL for `BadUri` and `RequireHttpsOnly`, and a
/// configured source URL may be written `https://user:token@host/`.
fn transport_text(e: &ureq::Error) -> String {
match e {
ureq::Error::BadUri(_) => "invalid url".to_owned(),
ureq::Error::RequireHttpsOnly(_) => "https required".to_owned(),
other => other.to_string(),
}
}
/// Ask one HTTPS source for its `Date`.
///
/// `HEAD`, not `GET`: the header is the whole payload. **A non-2xx status is an
/// answer**: the agent ([`agent_with`]) hands 4xx/5xx back as a response, which
/// still carries the header, and `https://api.stripe.com/` answers an
/// unauthenticated `HEAD` with exactly that. The offset is taken against the
/// midpoint of the request.
pub fn sample_https(agent: &ureq::Agent, url: &str) -> Result<Sample, String> {
let before = SystemTime::now();
let response = agent.head(url).call().map_err(|e| transport_text(&e))?;
let after = SystemTime::now();
let raw = response
.headers()
.get("date")
.and_then(|v| v.to_str().ok())
.map(str::to_owned)
.ok_or_else(|| "answered with no Date header".to_owned())?;
let source_unix_s = crate::date::parse_http_date(&raw)
.ok_or_else(|| format!("sent a Date this cannot read: {raw:?}"))?;
// A `Date` is TRUNCATED to the second: the sender's clock read somewhere in
// [s, s + 1000) ms. The midpoint of that interval is the unbiased estimate;
// taking `s` itself would read every correct clock as ~500 ms fast.
let source_unix_ms = source_unix_s * 1_000 + 500;
let (start, end) = (crate::unix_ms(before), crate::unix_ms(after));
Ok(Sample {
source: url.to_owned(),
source_unix_ms,
offset_ms: source_unix_ms - (start + (end - start) / 2),
})
}
/// Ask every source at once and decide. Blocking; no clock is moved here.
///
/// [`crate::poll::poll_with`] with the HTTPS sampler: the thread-per-source
/// poll, the peer and SNTP samplers and the quorum are that module's, shared with
/// the zero-egress [`crate::poll::poll`].
pub fn poll(sources: &[Source], peer_key: Option<&str>) -> Poll {
let tls = EXTRA_ROOTS
.get()
.and_then(|roots| tls_with(roots).ok())
.unwrap_or_else(|| tls_with(&[]).expect("no added root, nothing to refuse"));
let agent = agent_with(tls);
crate::poll::poll_with(sources, peer_key, &|url| sample_https(&agent, url))
}
/// **The one call.** Read the configuration's verdict, move the clock by the
/// [`crate::adjust`] rule, record and return the report. Never panics, never
/// fails: every outcome is a [`Report`].
#[cfg(feature = "adjust")]
pub fn sync(config: &Result<Config, String>, phase: Trigger) -> Report {
let report = match config {
Err(why) => Report::without_sources(phase, Outcome::Misconfigured, why.clone()),
Ok(Config::NotConfigured) => Report::without_sources(
phase,
Outcome::NotConfigured,
format!(
"no clock source is configured ({}); this machine's clock is whatever it booted with",
crate::config::CLOCK_ENV
),
),
Ok(Config::Off) => Report::without_sources(
phase,
Outcome::Off,
format!("{}=off: this machine was told not to set its own clock", crate::config::CLOCK_ENV),
),
Ok(Config::Sources { sources, peer_key }) => {
let polled = poll(sources, peer_key.as_deref());
let action = crate::adjust::choose(phase, &polled.verdict.decision);
let applied = crate::adjust::apply(action).map_err(|err| err.to_string());
Report::from_poll(phase, &polled, action, applied)
}
};
crate::report::record(report.clone());
report
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_poll_of_one_reachable_peer_decides_on_its_answer_alone() {
let signer = crate::peer::generate_key().unwrap();
let key_line = signer.public_key().to_openssh().unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || crate::peer::serve(listener, &signer, &|| Ok(()), &|_, _| {}));
let polled = poll(&[Source::Peer(addr)], Some(&key_line));
assert!(
matches!(
polled.verdict.decision,
crate::Decision::AlreadyRight { .. }
),
"{polled:?}"
);
}
#[test]
fn a_poll_with_a_broken_peer_key_reports_the_key_not_a_timeout() {
let polled = poll(
&[Source::Peer("127.0.0.1:9".parse().unwrap())],
Some("ssh-ed25519 AAAA"),
);
let Err((_, why)) = &polled.answers[0] else {
panic!("must fail")
};
assert!(why.contains("does not parse"), "{why}");
}
#[test]
fn a_skew_triggered_sync_takes_its_offset_from_the_signed_peer_and_nothing_else() {
// The refusal that triggered this claimed a day of skew (a forged
// credential can claim anything). The handler is `Fn()`, so that number
// cannot even reach here: the only input is the peer's signed answer,
// which agrees with this clock — so nothing moves.
let signer = crate::peer::generate_key().unwrap();
let key_line = signer.public_key().to_openssh().unwrap();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
std::thread::spawn(move || crate::peer::serve(listener, &signer, &|| Ok(()), &|_, _| {}));
let config = Ok(Config::Sources {
sources: vec![Source::Peer(addr)],
peer_key: Some(key_line),
});
let report = sync(&config, Trigger::SkewRefusal);
assert_eq!(report.trigger, Trigger::SkewRefusal);
assert_eq!(report.outcome, Outcome::AlreadyRight, "{report:?}");
assert!(report.offset_ms.abs() < 1_000);
}
#[test]
fn a_timer_sync_against_an_unreachable_front_changes_nothing_and_says_so() {
let signer = crate::peer::generate_key().unwrap();
let key_line = signer.public_key().to_openssh().unwrap();
let closed = std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap();
let config = Ok(Config::Sources {
sources: vec![Source::Peer(closed)],
peer_key: Some(key_line),
});
let before = crate::now_unix_ms();
let report = sync(&config, Trigger::Timer);
assert_eq!(
(report.trigger, report.outcome),
(Trigger::Timer, Outcome::NoQuorum)
);
assert_eq!(report.offset_ms, 0);
assert!(report.sources[0].error.is_some());
assert!(
(crate::now_unix_ms() - before).abs() < 5_000,
"the clock did not move"
);
}
/// Needs the internet; run by hand on a box whose clock is right:
/// `cargo test -p estate-clock --features fetch -- --ignored --nocapture`.
#[test]
#[ignore = "needs the internet"]
fn the_https_sources_the_front_uses_answer_and_agree_with_a_correct_clock() {
let sources: Vec<Source> = [
"https://api.stripe.com/",
"https://www.cloudflare.com/",
"https://www.google.com/",
]
.iter()
.map(|u| Source::Https((*u).to_owned()))
.collect();
let polled = poll(&sources, None);
println!("{polled:?}");
assert!(matches!(
polled.verdict.decision,
crate::Decision::AlreadyRight { .. }
));
}
}
/// **A local HTTPS clock vouched for by an added root** (2026-09-26, the bench's
/// local clock): refused with the public roots alone, read with the root added.
#[cfg(test)]
mod added_roots {
use super::*;
use std::io::{Read as _, Write as _};
fn local_clock() -> (String, Vec<u8>) {
let ca_key = rcgen::KeyPair::generate().unwrap();
let mut ca = rcgen::CertificateParams::default();
ca.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
ca.distinguished_name.push(rcgen::DnType::CommonName, "bench CA");
let ca_cert = ca.self_signed(&ca_key).unwrap();
let key = rcgen::KeyPair::generate().unwrap();
let mut leaf = rcgen::CertificateParams::default();
leaf.subject_alt_names = vec![rcgen::SanType::IpAddress("127.0.0.1".parse().unwrap())];
let leaf_cert = leaf.signed_by(&key, &ca_cert, &ca_key).unwrap();
let server = rustls::ServerConfig::builder_with_provider(std::sync::Arc::new(rustls::crypto::ring::default_provider()))
.with_safe_default_protocol_versions()
.unwrap()
.with_no_client_auth()
.with_single_cert(
vec![leaf_cert.der().clone()],
rustls::pki_types::PrivateKeyDer::Pkcs8(key.serialize_der().into()),
)
.unwrap();
let server = std::sync::Arc::new(server);
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
std::thread::spawn(move || {
for tcp in listener.incoming().flatten() {
let conn = rustls::ServerConnection::new(std::sync::Arc::clone(&server)).unwrap();
let mut tls = rustls::StreamOwned::new(conn, tcp);
let mut buf = [0u8; 1024];
if tls.read(&mut buf).is_ok() {
let _ = tls.write_all(b"HTTP/1.1 204 No Content\r\nDate: Sat, 26 Sep 2026 08:00:00 GMT\r\nConnection: close\r\n\r\n");
let _ = tls.flush();
}
}
});
(format!("https://127.0.0.1:{port}/"), ca_cert.der().to_vec())
}
#[test]
fn a_local_clock_signed_by_an_added_root_is_read_and_without_it_refused() {
let (url, ca) = local_clock();
let public_only = agent_with(tls_with(&[]).unwrap());
assert!(sample_https(&public_only, &url).is_err(), "the public roots alone do not vouch for the bench's clock");
let with_ca = agent_with(tls_with(&[ca]).unwrap());
let s = sample_https(&with_ca, &url).expect("the added root vouches for it");
assert_eq!(s.source_unix_ms, 1_790_409_600_000 + 500);
}
}