use std::net::{IpAddr, SocketAddr};
use std::process::ExitStatus;
use std::time::Duration;
use tokio::process::Command;
use tokio::time::sleep;
use crate::errors::{VnetLookupError, WarpCliError, WarpConnectError};
const WARP_CLI_TIMEOUT: Duration = Duration::from_secs(30);
const ROUTE_TIMEOUT_SECS: u64 = 5;
const CONFIG_DELAY_SECS: u64 = 5;
const DNS_ATTEMPTS: u32 = 40;
const DNS_INTERVAL_SECS: u64 = 5;
const DOMAIN_TO_TEST_DNS: &str = "api.ipify.org";
const IP_SERVICE_URL: &str = "https://api.ipify.org?format=text";
const DAEMON_UNREACHABLE: &str = "Error communicating with daemon";
pub async fn connect() -> Result<(), WarpConnectError> {
warp_cli(&["connect"]).await?;
sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
if !wait_dns_ready(DOMAIN_TO_TEST_DNS, DNS_ATTEMPTS, DNS_INTERVAL_SECS).await {
return Err(WarpConnectError::DnsUnresolved);
}
Ok(())
}
pub async fn disconnect() -> Result<(), WarpCliError> {
warp_cli(&["disconnect"]).await?;
sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
Ok(())
}
pub async fn status() -> Result<String, WarpCliError> {
warp_cli(&["status"]).await
}
pub async fn set_virtual_network(vnet_name: &str) -> Result<(), VnetLookupError> {
let vnet_id = get_virtual_network_id(vnet_name).await?;
warp_cli(&["vnet", &vnet_id]).await?;
sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
Ok(())
}
async fn fetch_public_ip() -> Option<String> {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(10))
.build()
.inspect_err(|error| tracing::error!(error = %error, "error building the public IP client"))
.ok()?;
let response = client
.get(IP_SERVICE_URL)
.send()
.await
.inspect_err(|error| tracing::error!(error = %error, "error fetching public IP"))
.ok()?;
if response.status() != reqwest::StatusCode::OK {
tracing::error!(status = %response.status(), "failed to fetch public IP");
return None;
}
response
.text()
.await
.inspect_err(|error| tracing::error!(error = %error, "error reading public IP"))
.ok()
}
pub async fn test_public_ip(expected_ip: &str) -> bool {
fetch_public_ip()
.await
.is_some_and(|public_ip| public_ip.trim() == expected_ip)
}
pub async fn wait_public_ip(expected_ip: &str, attempts: u32, interval_secs: u64) -> bool {
for attempt in 0..attempts {
if test_public_ip(expected_ip).await {
return true;
}
if attempt.saturating_add(1) < attempts {
sleep(Duration::from_secs(interval_secs)).await;
}
}
false
}
pub async fn wait_dns_ready(host: &str, attempts: u32, interval_secs: u64) -> bool {
for attempt in 0..attempts {
if resolve_ipv4(host).await.is_some() {
return true;
}
if attempt.saturating_add(1) < attempts {
sleep(Duration::from_secs(interval_secs)).await;
}
}
false
}
pub async fn is_using_split_tunnel(host: &str) -> bool {
let target = resolve_host(host).await;
let route = tokio::time::timeout(
Duration::from_secs(ROUTE_TIMEOUT_SECS),
Command::new("ip")
.args(["route", "get", &target])
.kill_on_drop(true)
.output(),
)
.await;
let output = match route {
Ok(Ok(output)) => output,
Ok(Err(error)) => {
tracing::error!(host, error = %error, "error getting IP route in split tunnel");
return false;
}
Err(_) => {
tracing::error!(host, "timeout getting IP route in split tunnel");
return false;
}
};
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
tracing::error!(host, %stderr, "error getting IP route in split tunnel");
return false;
}
String::from_utf8_lossy(&output.stdout).contains("CloudflareWARP")
}
pub async fn get_virtual_network_id(vnet_name: &str) -> Result<String, VnetLookupError> {
parse_virtual_network_id(&warp_cli(&["vnet"]).await?, vnet_name)
}
fn parse_virtual_network_id(vnets: &str, vnet_name: &str) -> Result<String, VnetLookupError> {
let named = format!(" Name: {vnet_name}\n");
let lines: Vec<&str> = vnets.split_inclusive('\n').collect();
let id = lines
.windows(2)
.find_map(|pair| match pair {
[id_line, name_line] if *name_line == named => {
id_line.split_once("ID: ").map(|(_, id)| id.trim())
}
_ => None,
})
.ok_or_else(|| VnetLookupError::UnknownNetwork {
name: vnet_name.to_owned(),
})?;
if id.is_empty() {
return Err(VnetLookupError::Unidentified {
name: vnet_name.to_owned(),
});
}
Ok(id.to_owned())
}
async fn resolve_host(host: &str) -> String {
resolve_ipv4(host)
.await
.map_or_else(|| host.to_owned(), |address| address.to_string())
}
async fn resolve_ipv4(host: &str) -> Option<IpAddr> {
first_ipv4(tokio::net::lookup_host((host, 0)).await.ok()?)
}
fn first_ipv4(addresses: impl Iterator<Item = SocketAddr>) -> Option<IpAddr> {
addresses.map(|address| address.ip()).find(IpAddr::is_ipv4)
}
async fn warp_cli(args: &[&str]) -> Result<String, WarpCliError> {
let mut warp_cli = Command::new("warp-cli");
warp_cli.arg("--accept-tos").args(args);
run(&mut warp_cli, WARP_CLI_TIMEOUT).await
}
async fn run(warp_cli: &mut Command, timeout: Duration) -> Result<String, WarpCliError> {
let output = tokio::time::timeout(timeout, warp_cli.kill_on_drop(true).output())
.await
.map_err(|_| WarpCliError::TimedOut { after: timeout })?
.map_err(|source| WarpCliError::NotStarted { source })?;
if !output.status.success() {
return Err(refusal(output.status, &output.stderr));
}
String::from_utf8(output.stdout)
.map(|stdout| stdout.trim().to_owned())
.map_err(|source| WarpCliError::Malformed { source })
}
fn refusal(status: ExitStatus, stderr: &[u8]) -> WarpCliError {
let stderr = String::from_utf8_lossy(stderr).trim().to_owned();
if stderr.contains(DAEMON_UNREACHABLE) {
WarpCliError::DaemonUnreachable { status, stderr }
} else {
WarpCliError::Refused { status, stderr }
}
}
#[cfg(test)]
mod tests {
use super::{first_ipv4, parse_virtual_network_id, run};
use crate::errors::{VnetLookupError, WarpCliError};
use std::net::SocketAddr;
use std::time::Duration;
use tokio::process::Command;
const VNETS: &str = "\
Currently selected: 92d3926f-f86c-498c-b27c-85a2037f0d70
Virtual Networks:
ID: 11fba1a0-4abe-4e66-8a9e-182506fdc482
Name: amsterdam
Comment: ZTNA Default Virtual Network
Default: false
ID: 49936bfc-52c8-4247-8a8a-da1ff1cb2eed
Name: egress-ips
Comment: ZTNA Default Virtual Network
Default: false
";
#[test]
fn a_virtual_network_is_read_by_the_name_that_follows_its_id() {
assert_eq!(
parse_virtual_network_id(VNETS, "egress-ips").unwrap(),
"49936bfc-52c8-4247-8a8a-da1ff1cb2eed"
);
}
#[test]
fn a_missing_virtual_network_is_reported_by_the_name_asked_for() {
let error = parse_virtual_network_id(VNETS, "nope").unwrap_err();
assert!(matches!(&error, VnetLookupError::UnknownNetwork { name } if name == "nope"));
assert_eq!(error.to_string(), "Failed to find virtual network nope");
}
#[test]
fn a_name_is_matched_whole_and_as_written() {
for name in ["egress", "egress.ips"] {
assert!(matches!(
parse_virtual_network_id(VNETS, name),
Err(VnetLookupError::UnknownNetwork { .. })
));
}
}
#[test]
fn the_id_comes_back_without_surrounding_whitespace() {
let vnets = " ID: 49936bfc \r\n Name: egress-ips\n";
assert_eq!(
parse_virtual_network_id(vnets, "egress-ips").unwrap(),
"49936bfc"
);
}
#[test]
fn a_network_listed_without_an_id_is_unidentified() {
let vnets = " ID: \n Name: egress-ips\n";
assert!(matches!(
parse_virtual_network_id(vnets, "egress-ips"),
Err(VnetLookupError::Unidentified { name }) if name == "egress-ips"
));
}
fn shell(script: &str) -> Command {
let mut shell = Command::new("sh");
shell.args(["-c", script]);
shell
}
#[tokio::test]
async fn a_cli_that_cannot_start_keeps_why() {
let outcome = run(
&mut Command::new("warp-cli-that-is-not-installed"),
Duration::from_secs(1),
)
.await;
assert!(matches!(
outcome,
Err(WarpCliError::NotStarted { source }) if source.kind() == std::io::ErrorKind::NotFound
));
}
#[tokio::test]
async fn a_cli_that_does_not_answer_in_time_is_killed() {
let marker = std::env::temp_dir().join(format!("warp-cli-killed-{}", std::process::id()));
let _ = std::fs::remove_file(&marker);
let script = format!("sleep 1; touch '{}'", marker.display());
let outcome = run(&mut shell(&script), Duration::from_millis(100)).await;
tokio::time::sleep(Duration::from_millis(1500)).await;
assert!(matches!(outcome, Err(WarpCliError::TimedOut { .. })));
assert!(!marker.exists());
}
#[tokio::test]
async fn a_refusal_keeps_its_stderr_readable() {
let outcome = run(
&mut shell("printf 'daemon \\377 down' >&2; exit 1"),
Duration::from_secs(5),
)
.await;
assert!(matches!(
outcome,
Err(WarpCliError::Refused { stderr, .. }) if stderr == "daemon \u{FFFD} down"
));
}
#[tokio::test]
async fn a_daemon_that_does_not_answer_is_told_apart_from_a_refusal() {
let outcome = run(
&mut shell("echo 'Error communicating with daemon: connection refused' >&2; exit 1"),
Duration::from_secs(5),
)
.await;
assert!(matches!(
outcome,
Err(WarpCliError::DaemonUnreachable { .. })
));
}
#[tokio::test]
async fn an_answer_that_is_not_utf8_is_malformed() {
let outcome = run(&mut shell("printf 'ID: \\377'"), Duration::from_secs(5)).await;
assert!(matches!(outcome, Err(WarpCliError::Malformed { .. })));
}
#[test]
fn a_host_that_answers_on_both_families_routes_through_its_ipv4() {
let addresses = [
"[2606:4700:90:0:f22e:fbec:5bed:a9b9]:0"
.parse::<SocketAddr>()
.unwrap(),
"172.65.251.78:0".parse::<SocketAddr>().unwrap(),
];
assert_eq!(
first_ipv4(addresses.into_iter()).unwrap().to_string(),
"172.65.251.78"
);
}
#[test]
fn a_host_that_answers_on_ipv6_alone_does_not_resolve() {
let addresses = ["[2607:f8b0:4004:c07::71]:0".parse::<SocketAddr>().unwrap()];
assert!(first_ipv4(addresses.into_iter()).is_none());
}
}