fluidattacks_core_git/
warp.rs

1use std::net::{IpAddr, SocketAddr};
2use std::process::ExitStatus;
3use std::time::Duration;
4use tokio::process::Command;
5use tokio::time::sleep;
6
7use crate::errors::{VnetLookupError, WarpCliError, WarpConnectError};
8
9const WARP_CLI_TIMEOUT: Duration = Duration::from_secs(30);
10const ROUTE_TIMEOUT_SECS: u64 = 5;
11const CONFIG_DELAY_SECS: u64 = 5;
12const DNS_ATTEMPTS: u32 = 40;
13const DNS_INTERVAL_SECS: u64 = 5;
14const DOMAIN_TO_TEST_DNS: &str = "api.ipify.org";
15const IP_SERVICE_URL: &str = "https://api.ipify.org?format=text";
16const DAEMON_UNREACHABLE: &str = "Error communicating with daemon";
17
18pub async fn connect() -> Result<(), WarpConnectError> {
19    warp_cli(&["connect"]).await?;
20    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
21    if !wait_dns_ready(DOMAIN_TO_TEST_DNS, DNS_ATTEMPTS, DNS_INTERVAL_SECS).await {
22        return Err(WarpConnectError::DnsUnresolved);
23    }
24    Ok(())
25}
26
27pub async fn disconnect() -> Result<(), WarpCliError> {
28    warp_cli(&["disconnect"]).await?;
29    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
30    Ok(())
31}
32
33pub async fn status() -> Result<String, WarpCliError> {
34    warp_cli(&["status"]).await
35}
36
37pub async fn set_virtual_network(vnet_name: &str) -> Result<(), VnetLookupError> {
38    let vnet_id = get_virtual_network_id(vnet_name).await?;
39    warp_cli(&["vnet", &vnet_id]).await?;
40    sleep(Duration::from_secs(CONFIG_DELAY_SECS)).await;
41    Ok(())
42}
43
44async fn fetch_public_ip() -> Option<String> {
45    let client = reqwest::Client::builder()
46        .timeout(Duration::from_secs(10))
47        .build()
48        .inspect_err(|error| tracing::error!(error = %error, "error building the public IP client"))
49        .ok()?;
50
51    let response = client
52        .get(IP_SERVICE_URL)
53        .send()
54        .await
55        .inspect_err(|error| tracing::error!(error = %error, "error fetching public IP"))
56        .ok()?;
57
58    if response.status() != reqwest::StatusCode::OK {
59        tracing::error!(status = %response.status(), "failed to fetch public IP");
60        return None;
61    }
62
63    response
64        .text()
65        .await
66        .inspect_err(|error| tracing::error!(error = %error, "error reading public IP"))
67        .ok()
68}
69
70pub async fn test_public_ip(expected_ip: &str) -> bool {
71    fetch_public_ip()
72        .await
73        .is_some_and(|public_ip| public_ip.trim() == expected_ip)
74}
75
76pub async fn wait_public_ip(expected_ip: &str, attempts: u32, interval_secs: u64) -> bool {
77    for attempt in 0..attempts {
78        if test_public_ip(expected_ip).await {
79            return true;
80        }
81        if attempt.saturating_add(1) < attempts {
82            sleep(Duration::from_secs(interval_secs)).await;
83        }
84    }
85    false
86}
87
88pub async fn wait_dns_ready(host: &str, attempts: u32, interval_secs: u64) -> bool {
89    for attempt in 0..attempts {
90        if resolve_ipv4(host).await.is_some() {
91            return true;
92        }
93        if attempt.saturating_add(1) < attempts {
94            sleep(Duration::from_secs(interval_secs)).await;
95        }
96    }
97    false
98}
99
100pub async fn is_using_split_tunnel(host: &str) -> bool {
101    let target = resolve_host(host).await;
102    let route = tokio::time::timeout(
103        Duration::from_secs(ROUTE_TIMEOUT_SECS),
104        Command::new("ip")
105            .args(["route", "get", &target])
106            .kill_on_drop(true)
107            .output(),
108    )
109    .await;
110
111    let output = match route {
112        Ok(Ok(output)) => output,
113        Ok(Err(error)) => {
114            tracing::error!(host, error = %error, "error getting IP route in split tunnel");
115            return false;
116        }
117        Err(_) => {
118            tracing::error!(host, "timeout getting IP route in split tunnel");
119            return false;
120        }
121    };
122
123    if !output.status.success() {
124        let stderr = String::from_utf8_lossy(&output.stderr);
125        tracing::error!(host, %stderr, "error getting IP route in split tunnel");
126        return false;
127    }
128
129    String::from_utf8_lossy(&output.stdout).contains("CloudflareWARP")
130}
131
132/// The id of the virtual network called exactly `vnet_name`, as `warp-cli vnet` lists it.
133pub async fn get_virtual_network_id(vnet_name: &str) -> Result<String, VnetLookupError> {
134    parse_virtual_network_id(&warp_cli(&["vnet"]).await?, vnet_name)
135}
136
137// Each network is an `ID: <id>` line followed by its `  Name: <name>` line.
138fn parse_virtual_network_id(vnets: &str, vnet_name: &str) -> Result<String, VnetLookupError> {
139    let named = format!("  Name: {vnet_name}\n");
140    let lines: Vec<&str> = vnets.split_inclusive('\n').collect();
141    let id = lines
142        .windows(2)
143        .find_map(|pair| match pair {
144            [id_line, name_line] if *name_line == named => {
145                id_line.split_once("ID: ").map(|(_, id)| id.trim())
146            }
147            _ => None,
148        })
149        .ok_or_else(|| VnetLookupError::UnknownNetwork {
150            name: vnet_name.to_owned(),
151        })?;
152
153    if id.is_empty() {
154        return Err(VnetLookupError::Unidentified {
155            name: vnet_name.to_owned(),
156        });
157    }
158    Ok(id.to_owned())
159}
160
161async fn resolve_host(host: &str) -> String {
162    resolve_ipv4(host)
163        .await
164        .map_or_else(|| host.to_owned(), |address| address.to_string())
165}
166
167async fn resolve_ipv4(host: &str) -> Option<IpAddr> {
168    first_ipv4(tokio::net::lookup_host((host, 0)).await.ok()?)
169}
170
171fn first_ipv4(addresses: impl Iterator<Item = SocketAddr>) -> Option<IpAddr> {
172    addresses.map(|address| address.ip()).find(IpAddr::is_ipv4)
173}
174
175async fn warp_cli(args: &[&str]) -> Result<String, WarpCliError> {
176    let mut warp_cli = Command::new("warp-cli");
177    warp_cli.arg("--accept-tos").args(args);
178    run(&mut warp_cli, WARP_CLI_TIMEOUT).await
179}
180
181async fn run(warp_cli: &mut Command, timeout: Duration) -> Result<String, WarpCliError> {
182    let output = tokio::time::timeout(timeout, warp_cli.kill_on_drop(true).output())
183        .await
184        .map_err(|_| WarpCliError::TimedOut { after: timeout })?
185        .map_err(|source| WarpCliError::NotStarted { source })?;
186
187    if !output.status.success() {
188        return Err(refusal(output.status, &output.stderr));
189    }
190    String::from_utf8(output.stdout)
191        .map(|stdout| stdout.trim().to_owned())
192        .map_err(|source| WarpCliError::Malformed { source })
193}
194
195fn refusal(status: ExitStatus, stderr: &[u8]) -> WarpCliError {
196    let stderr = String::from_utf8_lossy(stderr).trim().to_owned();
197    if stderr.contains(DAEMON_UNREACHABLE) {
198        WarpCliError::DaemonUnreachable { status, stderr }
199    } else {
200        WarpCliError::Refused { status, stderr }
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use super::{first_ipv4, parse_virtual_network_id, run};
207    use crate::errors::{VnetLookupError, WarpCliError};
208    use std::net::SocketAddr;
209    use std::time::Duration;
210    use tokio::process::Command;
211
212    const VNETS: &str = "\
213Currently selected: 92d3926f-f86c-498c-b27c-85a2037f0d70
214
215Virtual Networks:
216  ID: 11fba1a0-4abe-4e66-8a9e-182506fdc482
217  Name: amsterdam
218  Comment: ZTNA Default Virtual Network
219  Default: false
220
221  ID: 49936bfc-52c8-4247-8a8a-da1ff1cb2eed
222  Name: egress-ips
223  Comment: ZTNA Default Virtual Network
224  Default: false
225";
226
227    #[test]
228    fn a_virtual_network_is_read_by_the_name_that_follows_its_id() {
229        assert_eq!(
230            parse_virtual_network_id(VNETS, "egress-ips").unwrap(),
231            "49936bfc-52c8-4247-8a8a-da1ff1cb2eed"
232        );
233    }
234
235    #[test]
236    fn a_missing_virtual_network_is_reported_by_the_name_asked_for() {
237        let error = parse_virtual_network_id(VNETS, "nope").unwrap_err();
238
239        assert!(matches!(&error, VnetLookupError::UnknownNetwork { name } if name == "nope"));
240        assert_eq!(error.to_string(), "Failed to find virtual network nope");
241    }
242
243    #[test]
244    fn a_name_is_matched_whole_and_as_written() {
245        for name in ["egress", "egress.ips"] {
246            assert!(matches!(
247                parse_virtual_network_id(VNETS, name),
248                Err(VnetLookupError::UnknownNetwork { .. })
249            ));
250        }
251    }
252
253    #[test]
254    fn the_id_comes_back_without_surrounding_whitespace() {
255        let vnets = "  ID: 49936bfc  \r\n  Name: egress-ips\n";
256
257        assert_eq!(
258            parse_virtual_network_id(vnets, "egress-ips").unwrap(),
259            "49936bfc"
260        );
261    }
262
263    #[test]
264    fn a_network_listed_without_an_id_is_unidentified() {
265        let vnets = "  ID:   \n  Name: egress-ips\n";
266
267        assert!(matches!(
268            parse_virtual_network_id(vnets, "egress-ips"),
269            Err(VnetLookupError::Unidentified { name }) if name == "egress-ips"
270        ));
271    }
272
273    fn shell(script: &str) -> Command {
274        let mut shell = Command::new("sh");
275        shell.args(["-c", script]);
276        shell
277    }
278
279    #[tokio::test]
280    async fn a_cli_that_cannot_start_keeps_why() {
281        let outcome = run(
282            &mut Command::new("warp-cli-that-is-not-installed"),
283            Duration::from_secs(1),
284        )
285        .await;
286
287        assert!(matches!(
288            outcome,
289            Err(WarpCliError::NotStarted { source }) if source.kind() == std::io::ErrorKind::NotFound
290        ));
291    }
292
293    #[tokio::test]
294    async fn a_cli_that_does_not_answer_in_time_is_killed() {
295        let marker = std::env::temp_dir().join(format!("warp-cli-killed-{}", std::process::id()));
296        let _ = std::fs::remove_file(&marker);
297        let script = format!("sleep 1; touch '{}'", marker.display());
298
299        let outcome = run(&mut shell(&script), Duration::from_millis(100)).await;
300        tokio::time::sleep(Duration::from_millis(1500)).await;
301
302        assert!(matches!(outcome, Err(WarpCliError::TimedOut { .. })));
303        assert!(!marker.exists());
304    }
305
306    #[tokio::test]
307    async fn a_refusal_keeps_its_stderr_readable() {
308        let outcome = run(
309            &mut shell("printf 'daemon \\377 down' >&2; exit 1"),
310            Duration::from_secs(5),
311        )
312        .await;
313
314        assert!(matches!(
315            outcome,
316            Err(WarpCliError::Refused { stderr, .. }) if stderr == "daemon \u{FFFD} down"
317        ));
318    }
319
320    #[tokio::test]
321    async fn a_daemon_that_does_not_answer_is_told_apart_from_a_refusal() {
322        let outcome = run(
323            &mut shell("echo 'Error communicating with daemon: connection refused' >&2; exit 1"),
324            Duration::from_secs(5),
325        )
326        .await;
327
328        assert!(matches!(
329            outcome,
330            Err(WarpCliError::DaemonUnreachable { .. })
331        ));
332    }
333
334    #[tokio::test]
335    async fn an_answer_that_is_not_utf8_is_malformed() {
336        let outcome = run(&mut shell("printf 'ID: \\377'"), Duration::from_secs(5)).await;
337
338        assert!(matches!(outcome, Err(WarpCliError::Malformed { .. })));
339    }
340
341    #[test]
342    fn a_host_that_answers_on_both_families_routes_through_its_ipv4() {
343        let addresses = [
344            "[2606:4700:90:0:f22e:fbec:5bed:a9b9]:0"
345                .parse::<SocketAddr>()
346                .unwrap(),
347            "172.65.251.78:0".parse::<SocketAddr>().unwrap(),
348        ];
349
350        assert_eq!(
351            first_ipv4(addresses.into_iter()).unwrap().to_string(),
352            "172.65.251.78"
353        );
354    }
355
356    #[test]
357    fn a_host_that_answers_on_ipv6_alone_does_not_resolve() {
358        let addresses = ["[2607:f8b0:4004:c07::71]:0".parse::<SocketAddr>().unwrap()];
359
360        assert!(first_ipv4(addresses.into_iter()).is_none());
361    }
362}
fluidattacks_core_git/warp.rs

fluidattacks_core_git/
warp.rs
