fluidattacks_core_git/
warp.rs

1use std::net::{IpAddr, SocketAddr};
2use std::time::Duration;
3use tokio::process::Command;
4use tokio::time::sleep;
5
6const WARP_CLI_TIMEOUT_SECS: u64 = 30;
7const ROUTE_TIMEOUT_SECS: u64 = 5;
8const CONFIG_DELAY_SECS: u64 = 5;
9const DNS_ATTEMPTS: u32 = 40;
10const DNS_INTERVAL_SECS: u64 = 5;
11const DOMAIN_TO_TEST_DNS: &str = "api.ipify.org";
12const IP_SERVICE_URL: &str = "https://api.ipify.org?format=text";
13
14/// Every failure warp reports, carrying the message consumers match on.
15#[derive(Debug)]
16pub struct WarpError {
17    message: String,
18}
19
20impl WarpError {
21    fn new(message: impl Into<String>) -> Self {
22        Self {
23            message: message.into(),
24        }
25    }
26}
27
28impl std::fmt::Display for WarpError {
29    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
30        write!(formatter, "{}", self.message)
31    }
32}
33
34impl std::error::Error for WarpError {}
35
36pub async fn connect() -> Result<(), WarpError> {
37    warp_cli(&["connect"]).await?;
38    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
39    if !wait_dns_ready(DOMAIN_TO_TEST_DNS, DNS_ATTEMPTS, DNS_INTERVAL_SECS).await {
40        return Err(WarpError::new("Failed to resolve DNS"));
41    }
42    Ok(())
43}
44
45pub async fn disconnect() -> Result<(), WarpError> {
46    warp_cli(&["disconnect"]).await?;
47    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
48    Ok(())
49}
50
51pub async fn status() -> Result<String, WarpError> {
52    warp_cli(&["status"]).await
53}
54
55pub async fn set_virtual_network(vnet_name: &str) -> Result<(), WarpError> {
56    let vnet_id = get_virtual_network_id(vnet_name).await?;
57    warp_cli(&["vnet", &vnet_id]).await?;
58    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
59    Ok(())
60}
61
62async fn fetch_public_ip() -> Option<String> {
63    let client = reqwest::Client::builder()
64        .timeout(Duration::from_secs(10))
65        .build()
66        .inspect_err(|error| tracing::error!(error = %error, "error building the public IP client"))
67        .ok()?;
68
69    let response = client
70        .get(IP_SERVICE_URL)
71        .send()
72        .await
73        .inspect_err(|error| tracing::error!(error = %error, "error fetching public IP"))
74        .ok()?;
75
76    if response.status() != reqwest::StatusCode::OK {
77        tracing::error!(status = %response.status(), "failed to fetch public IP");
78        return None;
79    }
80
81    response
82        .text()
83        .await
84        .inspect_err(|error| tracing::error!(error = %error, "error reading public IP"))
85        .ok()
86}
87
88pub async fn test_public_ip(expected_ip: &str) -> bool {
89    fetch_public_ip()
90        .await
91        .is_some_and(|public_ip| public_ip.trim() == expected_ip)
92}
93
94pub async fn wait_public_ip(expected_ip: &str, attempts: u32, interval_secs: u64) -> bool {
95    for attempt in 0..attempts {
96        if test_public_ip(expected_ip).await {
97            return true;
98        }
99        if attempt.saturating_add(1) < attempts {
100            sleep(Duration::from_secs(interval_secs)).await;
101        }
102    }
103    false
104}
105
106pub async fn wait_dns_ready(host: &str, attempts: u32, interval_secs: u64) -> bool {
107    for attempt in 0..attempts {
108        if resolve_ipv4(host).await.is_some() {
109            return true;
110        }
111        if attempt.saturating_add(1) < attempts {
112            sleep(Duration::from_secs(interval_secs)).await;
113        }
114    }
115    false
116}
117
118pub async fn is_using_split_tunnel(host: &str) -> bool {
119    let target = resolve_host(host).await;
120    let route = tokio::time::timeout(
121        Duration::from_secs(ROUTE_TIMEOUT_SECS),
122        Command::new("ip")
123            .args(["route", "get", &target])
124            .kill_on_drop(true)
125            .output(),
126    )
127    .await;
128
129    let output = match route {
130        Ok(Ok(output)) => output,
131        Ok(Err(error)) => {
132            tracing::error!(host, error = %error, "error getting IP route in split tunnel");
133            return false;
134        }
135        Err(_) => {
136            tracing::error!(host, "timeout getting IP route in split tunnel");
137            return false;
138        }
139    };
140
141    if !output.status.success() {
142        let stderr = String::from_utf8_lossy(&output.stderr);
143        tracing::error!(host, %stderr, "error getting IP route in split tunnel");
144        return false;
145    }
146
147    String::from_utf8_lossy(&output.stdout).contains("CloudflareWARP")
148}
149
150pub async fn get_virtual_network_id(vnet_name: &str) -> Result<String, WarpError> {
151    parse_virtual_network_id(&warp_cli(&["vnet"]).await?, vnet_name)
152}
153
154fn parse_virtual_network_id(vnets: &str, vnet_name: &str) -> Result<String, WarpError> {
155    let pattern = format!("ID: (.*)\n  Name: {}\n", regex::escape(vnet_name));
156    let expression =
157        regex::Regex::new(&pattern).map_err(|error| WarpError::new(error.to_string()))?;
158
159    expression
160        .captures(vnets)
161        .and_then(|captures| captures.get(1))
162        .map(|id| id.as_str().trim().to_owned())
163        .ok_or_else(|| WarpError::new(format!("Failed to find virtual network {vnet_name}")))
164}
165
166async fn resolve_host(host: &str) -> String {
167    resolve_ipv4(host)
168        .await
169        .map_or_else(|| host.to_owned(), |address| address.to_string())
170}
171
172async fn resolve_ipv4(host: &str) -> Option<IpAddr> {
173    first_ipv4(tokio::net::lookup_host((host, 0)).await.ok()?)
174}
175
176fn first_ipv4(addresses: impl Iterator<Item = SocketAddr>) -> Option<IpAddr> {
177    addresses.map(|address| address.ip()).find(IpAddr::is_ipv4)
178}
179
180async fn warp_cli(args: &[&str]) -> Result<String, WarpError> {
181    let mut full_args = vec!["--accept-tos"];
182    full_args.extend(args);
183
184    let output = tokio::time::timeout(
185        Duration::from_secs(WARP_CLI_TIMEOUT_SECS),
186        Command::new("warp-cli")
187            .args(&full_args)
188            .kill_on_drop(true)
189            .output(),
190    )
191    .await
192    .map_err(|_| WarpError::new("Failed to run command"))?
193    .map_err(|_| WarpError::new("Failed to run command"))?;
194
195    if !output.status.success() {
196        return Err(WarpError::new(
197            String::from_utf8_lossy(&output.stderr).trim(),
198        ));
199    }
200
201    Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned())
202}
203
204#[cfg(test)]
205mod tests {
206    use super::{first_ipv4, parse_virtual_network_id};
207    use std::net::SocketAddr;
208
209    const VNETS: &str = "\
210Currently selected: 92d3926f-f86c-498c-b27c-85a2037f0d70
211
212Virtual Networks:
213  ID: 11fba1a0-4abe-4e66-8a9e-182506fdc482
214  Name: amsterdam
215  Comment: ZTNA Default Virtual Network
216  Default: false
217
218  ID: 49936bfc-52c8-4247-8a8a-da1ff1cb2eed
219  Name: egress-ips
220  Comment: ZTNA Default Virtual Network
221  Default: false
222";
223
224    #[test]
225    fn a_virtual_network_is_read_by_the_name_that_follows_its_id() {
226        assert_eq!(
227            parse_virtual_network_id(VNETS, "egress-ips").unwrap(),
228            "49936bfc-52c8-4247-8a8a-da1ff1cb2eed"
229        );
230    }
231
232    #[test]
233    fn a_missing_virtual_network_is_reported_by_the_name_asked_for() {
234        let error = parse_virtual_network_id(VNETS, "nope").unwrap_err();
235        assert_eq!(error.to_string(), "Failed to find virtual network nope");
236    }
237
238    #[test]
239    fn a_host_that_answers_on_both_families_routes_through_its_ipv4() {
240        let addresses = [
241            "[2606:4700:90:0:f22e:fbec:5bed:a9b9]:0"
242                .parse::<SocketAddr>()
243                .unwrap(),
244            "172.65.251.78:0".parse::<SocketAddr>().unwrap(),
245        ];
246
247        assert_eq!(
248            first_ipv4(addresses.into_iter()).unwrap().to_string(),
249            "172.65.251.78"
250        );
251    }
252
253    #[test]
254    fn a_host_that_answers_on_ipv6_alone_does_not_resolve() {
255        let addresses = ["[2607:f8b0:4004:c07::71]:0".parse::<SocketAddr>().unwrap()];
256
257        assert!(first_ipv4(addresses.into_iter()).is_none());
258    }
259}
fluidattacks_core_git/warp.rs

fluidattacks_core_git/
warp.rs
