use super::*;
pub(crate) async fn paid_exit_route_probe_measurement(
dns_health: &crate::secure_dns_runtime::SecureDnsHealthProbe,
app: &AppConfig,
now_unix: u64,
bind_interface: &str,
) -> Result<PaidRouteProbeMeasurement> {
dns_health.check_paid_exit().await?;
let args = paid_exit_health_probe_args();
let (measurement, _, bandwidth_error) =
paid_exit_probe_measurement(&args, app, now_unix, Some(bind_interface)).await?;
if let Some(error) = bandwidth_error {
eprintln!("paid-exit: automatic bandwidth sample incomplete: {error}");
}
if !automatic_probe_observed_public_ip(&measurement) {
return Err(anyhow!(
"paid exit free probe did not observe a public exit IP"
));
}
Ok(measurement)
}
fn paid_exit_health_probe_args() -> PaidExitProbeArgs {
PaidExitProbeArgs {
config: None,
session: String::new(),
ip_url: None,
stun_servers: Vec::new(),
no_stun: true,
geoip_url_template: None,
no_geoip: true,
download_url: None,
upload_url: None,
bandwidth_bytes: 0,
no_bandwidth: true,
samples: 1,
timeout_secs: 5,
no_reload_daemon: true,
json: false,
}
}
pub(super) fn automatic_probe_observed_public_ip(measurement: &PaidRouteProbeMeasurement) -> bool {
measurement.realized_exit_ip.is_some() && measurement.success_count() > 0
}
pub(crate) async fn update_automatic_paid_exit(
automatic: &mut PaidExitAutomaticBuyer,
runtime: &crate::fips_private_mesh::FipsPrivateTunnelRuntime,
app: &mut AppConfig,
config_path: &Path,
buyer_delta: &PaidRouteUsage,
now_unix: u64,
) -> Result<bool> {
automatic.cancel_if_disabled(app);
if !PaidExitAutomaticBuyer::enabled(app) {
return Ok(false);
}
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.observe_presence(&runtime.peer_statuses(), now_unix);
candidate.observe_usage(buyer_delta, now_unix);
}
if automatic.probe.is_none()
&& automatic.candidate.as_ref().is_some_and(|candidate| {
candidate.probe_started_at.is_none()
&& candidate.last_authenticated_at.is_some_and(|observed| {
now_unix.saturating_sub(observed) <= PAID_EXIT_AUTO_HEALTH_TTL_SECS
})
})
{
let probe_app = app.clone();
let dns_health = runtime.paid_exit_dns_health_probe();
let bind_interface = runtime.iface().to_string();
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.probe_started_at = Some(now_unix);
candidate.last_tx_at = None;
candidate.last_rx_at = None;
}
automatic.probe = Some(PaidExitAutomaticProbe {
generation: automatic.generation,
task: tokio::spawn(async move {
let dns_health = dns_health?;
paid_exit_route_probe_measurement(
&dns_health,
&probe_app,
now_unix,
&bind_interface,
)
.await
}),
});
}
if automatic
.probe
.as_ref()
.is_some_and(|probe| probe.task.is_finished())
{
let probe = automatic.probe.take().expect("finished probe exists");
let result = probe
.task
.await
.map_err(|error| anyhow!("automatic paid exit probe task failed: {error}"))?;
if probe.generation == automatic.generation {
match result {
Ok(measurement) => {
let session_id = automatic
.candidate
.as_ref()
.map(|candidate| candidate.session_id.clone())
.ok_or_else(|| anyhow!("automatic paid exit probe lost its candidate"))?;
record_paid_exit_probe(config_path, &session_id, measurement, now_unix)?;
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.probe_succeeded = true;
if candidate.health_evidence_fresh(now_unix) {
candidate.last_healthy_at = Some(now_unix);
}
}
}
Err(error) => {
eprintln!("paid-exit: automatic free probe failed: {error}");
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.failed = true;
}
}
}
}
}
let fund = automatic.candidate.as_ref().is_some_and(|candidate| {
!candidate.failed
&& !candidate.funding_attempted
&& candidate.health_evidence_fresh(now_unix)
});
if fund {
let session_id = automatic
.candidate
.as_ref()
.map(|candidate| candidate.session_id.clone())
.expect("funding candidate exists");
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.funding_attempted = true;
}
match fund_automatic_paid_exit(app, config_path, &session_id, now_unix).await {
Ok(envelope) => {
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.funded = true;
candidate.last_healthy_at = Some(now_unix);
}
if let Err(error) = queue_paid_exit_payment(app, config_path, &envelope) {
eprintln!("paid-exit: automatic channel-open queue failed: {error}");
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.failed = true;
}
}
}
Err(error) => {
eprintln!("paid-exit: automatic funding failed: {error}");
if let Some(candidate) = automatic.candidate.as_mut() {
candidate.failed = true;
}
}
}
}
if automatic
.candidate
.as_ref()
.is_some_and(|candidate| candidate.should_failover(now_unix))
{
suspend_automatic_paid_exit(automatic, runtime, config_path, now_unix)?;
if !PaidExitAutomaticBuyer::enabled(app) {
return Ok(false);
}
app.set_internet_source(nostr_vpn_core::config::InternetSource::PaidAutomatic);
app.save(config_path)?;
automatic.cancel_candidate(true, now_unix);
return Ok(true);
}
Ok(false)
}
pub(crate) fn record_paid_exit_probe(
config_path: &Path,
session_id: &str,
measurement: PaidRouteProbeMeasurement,
now_unix: u64,
) -> Result<()> {
let store_path = paid_route_store_file_path(config_path);
update_paid_route_store(&store_path, |store| {
store.update_session_probe(UpdatePaidRouteSessionProbeRequest {
session_id: session_id.to_string(),
realized_exit_ip: measurement.realized_exit_ip,
observed_country_code: measurement.observed_country_code,
observed_asn: measurement.observed_asn,
quality: Some(measurement.quality),
now_unix,
})?;
Ok(())
})
}
#[cfg(test)]
mod health_probe_tests {
use super::*;
#[test]
fn health_probe_uses_fast_https_without_bandwidth_traffic() {
let args = paid_exit_health_probe_args();
assert!(args.no_stun);
assert!(args.no_bandwidth);
assert_eq!(args.bandwidth_bytes, 0);
assert_eq!(args.samples, 1);
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[tokio::test]
async fn health_probe_http_client_cannot_fall_back_from_bound_interface() {
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.expect("bind loopback test server");
let address = listener.local_addr().expect("read test server address");
let server = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.expect("accept test request");
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut request = [0_u8; 1024];
let _ = stream.read(&mut request).await.expect("read test request");
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
.await
.expect("write test response");
});
#[cfg(target_os = "linux")]
let loopback_interface = "lo";
#[cfg(target_os = "macos")]
let loopback_interface = "lo0";
let client = paid_exit_probe_http_client(Duration::from_secs(2), Some(loopback_interface))
.expect("build interface-bound client");
let response = client
.get(format!("http://{address}"))
.send()
.await
.expect("request through loopback interface");
assert_eq!(response.text().await.expect("read response"), "ok");
server.await.expect("test server task");
let missing_listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.expect("bind missing-interface test server");
let missing_address = missing_listener
.local_addr()
.expect("read missing-interface server address");
let missing_interface =
paid_exit_probe_http_client(Duration::from_secs(2), Some("nvpn-missing"))
.expect("build missing-interface client");
assert!(
missing_interface
.get(format!("http://{missing_address}"))
.send()
.await
.is_err(),
"an unavailable paid tunnel must fail instead of falling back"
);
}
}