fluidattacks-core-git 0.68.0

use std::net::{IpAddr, SocketAddr};
use std::time::Duration;
use tokio::process::Command;
use tokio::time::sleep;

const WARP_CLI_TIMEOUT_SECS: u64 = 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";

/// Every failure warp reports, carrying the message consumers match on.
#[derive(Debug)]
pub struct WarpError {
    message: String,
}

impl WarpError {
    fn new(message: impl Into<String>) -> Self {
        Self {
            message: message.into(),
        }
    }
}

impl std::fmt::Display for WarpError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{}", self.message)
    }
}

impl std::error::Error for WarpError {}

pub async fn connect() -> Result<(), WarpError> {
    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(WarpError::new("Failed to resolve DNS"));
    }
    Ok(())
}

pub async fn disconnect() -> Result<(), WarpError> {
    warp_cli(&["disconnect"]).await?;
    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
    Ok(())
}

pub async fn status() -> Result<String, WarpError> {
    warp_cli(&["status"]).await
}

pub async fn set_virtual_network(vnet_name: &str) -> Result<(), WarpError> {
    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, WarpError> {
    parse_virtual_network_id(&warp_cli(&["vnet"]).await?, vnet_name)
}

fn parse_virtual_network_id(vnets: &str, vnet_name: &str) -> Result<String, WarpError> {
    let pattern = format!("ID: (.*)\n  Name: {}\n", regex::escape(vnet_name));
    let expression =
        regex::Regex::new(&pattern).map_err(|error| WarpError::new(error.to_string()))?;

    expression
        .captures(vnets)
        .and_then(|captures| captures.get(1))
        .map(|id| id.as_str().trim().to_owned())
        .ok_or_else(|| WarpError::new(format!("Failed to find virtual network {vnet_name}")))
}

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, WarpError> {
    let mut full_args = vec!["--accept-tos"];
    full_args.extend(args);

    let output = tokio::time::timeout(
        Duration::from_secs(WARP_CLI_TIMEOUT_SECS),
        Command::new("warp-cli")
            .args(&full_args)
            .kill_on_drop(true)
            .output(),
    )
    .await
    .map_err(|_| WarpError::new("Failed to run command"))?
    .map_err(|_| WarpError::new("Failed to run command"))?;

    if !output.status.success() {
        return Err(WarpError::new(
            String::from_utf8_lossy(&output.stderr).trim(),
        ));
    }

    Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned())
}

#[cfg(test)]
mod tests {
    use super::{first_ipv4, parse_virtual_network_id};
    use std::net::SocketAddr;

    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_eq!(error.to_string(), "Failed to find virtual network nope");
    }

    #[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());
    }
}