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 _ in 0..attempts {
96        if test_public_ip(expected_ip).await {
97            return true;
98        }
99        sleep(Duration::from_secs(interval_secs)).await;
100    }
101    false
102}
103
104pub async fn wait_dns_ready(host: &str, attempts: u32, interval_secs: u64) -> bool {
105    for _ in 0..attempts {
106        if resolve_ipv4(host).await.is_some() {
107            return true;
108        }
109        sleep(Duration::from_secs(interval_secs)).await;
110    }
111    false
112}
113
114pub async fn is_using_split_tunnel(host: &str) -> bool {
115    let target = resolve_host(host).await;
116    let route = tokio::time::timeout(
117        Duration::from_secs(ROUTE_TIMEOUT_SECS),
118        Command::new("ip")
119            .args(["route", "get", &target])
120            .kill_on_drop(true)
121            .output(),
122    )
123    .await;
124
125    let output = match route {
126        Ok(Ok(output)) => output,
127        Ok(Err(error)) => {
128            tracing::error!(host, error = %error, "error getting IP route in split tunnel");
129            return false;
130        }
131        Err(_) => {
132            tracing::error!(host, "timeout getting IP route in split tunnel");
133            return false;
134        }
135    };
136
137    if !output.status.success() {
138        let stderr = String::from_utf8_lossy(&output.stderr);
139        tracing::error!(host, %stderr, "error getting IP route in split tunnel");
140        return false;
141    }
142
143    String::from_utf8_lossy(&output.stdout).contains("CloudflareWARP")
144}
145
146pub async fn get_virtual_network_id(vnet_name: &str) -> Result<String, WarpError> {
147    parse_virtual_network_id(&warp_cli(&["vnet"]).await?, vnet_name)
148}
149
150fn parse_virtual_network_id(vnets: &str, vnet_name: &str) -> Result<String, WarpError> {
151    let pattern = format!("ID: (.*)\n  Name: {}\n", regex::escape(vnet_name));
152    let expression =
153        regex::Regex::new(&pattern).map_err(|error| WarpError::new(error.to_string()))?;
154
155    expression
156        .captures(vnets)
157        .and_then(|captures| captures.get(1))
158        .map(|id| id.as_str().trim().to_owned())
159        .ok_or_else(|| WarpError::new(format!("Failed to find virtual network {vnet_name}")))
160}
161
162async fn resolve_host(host: &str) -> String {
163    resolve_ipv4(host)
164        .await
165        .map_or_else(|| host.to_owned(), |address| address.to_string())
166}
167
168async fn resolve_ipv4(host: &str) -> Option<IpAddr> {
169    first_ipv4(tokio::net::lookup_host((host, 0)).await.ok()?)
170}
171
172fn first_ipv4(addresses: impl Iterator<Item = SocketAddr>) -> Option<IpAddr> {
173    addresses.map(|address| address.ip()).find(IpAddr::is_ipv4)
174}
175
176async fn warp_cli(args: &[&str]) -> Result<String, WarpError> {
177    let mut full_args = vec!["--accept-tos"];
178    full_args.extend(args);
179
180    let output = tokio::time::timeout(
181        Duration::from_secs(WARP_CLI_TIMEOUT_SECS),
182        Command::new("warp-cli")
183            .args(&full_args)
184            .kill_on_drop(true)
185            .output(),
186    )
187    .await
188    .map_err(|_| WarpError::new("Failed to run command"))?
189    .map_err(|_| WarpError::new("Failed to run command"))?;
190
191    if !output.status.success() {
192        return Err(WarpError::new(
193            String::from_utf8_lossy(&output.stderr).trim(),
194        ));
195    }
196
197    Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned())
198}
199
200#[cfg(test)]
201mod tests {
202    use super::{first_ipv4, parse_virtual_network_id};
203    use std::net::SocketAddr;
204
205    const VNETS: &str = "\
206Currently selected: 92d3926f-f86c-498c-b27c-85a2037f0d70
207
208Virtual Networks:
209  ID: 11fba1a0-4abe-4e66-8a9e-182506fdc482
210  Name: amsterdam
211  Comment: ZTNA Default Virtual Network
212  Default: false
213
214  ID: 49936bfc-52c8-4247-8a8a-da1ff1cb2eed
215  Name: egress-ips
216  Comment: ZTNA Default Virtual Network
217  Default: false
218";
219
220    #[test]
221    fn a_virtual_network_is_read_by_the_name_that_follows_its_id() {
222        assert_eq!(
223            parse_virtual_network_id(VNETS, "egress-ips").unwrap(),
224            "49936bfc-52c8-4247-8a8a-da1ff1cb2eed"
225        );
226    }
227
228    #[test]
229    fn a_missing_virtual_network_is_reported_by_the_name_asked_for() {
230        let error = parse_virtual_network_id(VNETS, "nope").unwrap_err();
231        assert_eq!(error.to_string(), "Failed to find virtual network nope");
232    }
233
234    #[test]
235    fn a_host_that_answers_on_both_families_routes_through_its_ipv4() {
236        let addresses = [
237            "[2606:4700:90:0:f22e:fbec:5bed:a9b9]:0"
238                .parse::<SocketAddr>()
239                .unwrap(),
240            "172.65.251.78:0".parse::<SocketAddr>().unwrap(),
241        ];
242
243        assert_eq!(
244            first_ipv4(addresses.into_iter()).unwrap().to_string(),
245            "172.65.251.78"
246        );
247    }
248
249    #[test]
250    fn a_host_that_answers_on_ipv6_alone_does_not_resolve() {
251        let addresses = ["[2607:f8b0:4004:c07::71]:0".parse::<SocketAddr>().unwrap()];
252
253        assert!(first_ipv4(addresses.into_iter()).is_none());
254    }
255}
fluidattacks_core_git/warp.rs

fluidattacks_core_git/
warp.rs
