OXYGENHUB
# da72c3b192c3 oxygenflow / crates / flow-init / src / egress.rs
2.7 KiB · Rust · last changed Allow-listed egress over vsock: network "egress" { allow "host"... } 20 days ago source blame history raw
1
//! The guest's half of allow-listed egress (oxygenflow#55): a proxy on
2
//! 127.0.0.1:3128 that relays each connection, unread, over vsock to the
3
//! runner (host CID 2), which decides what may pass. The VM has no network
4
//! device; this is its only way out.
5
6
use std::fs::File;
7
use std::io::{Read, Write};
8
use std::net::{TcpListener, TcpStream};
9
use std::os::fd::{AsRawFd, FromRawFd};
10
11
/// Where the job's tools reach the proxy, and the vsock port behind it.
12
const LISTEN: &str = "127.0.0.1:3128";
13
const VSOCK_PORT: u32 = 3128;
14
15
/// Start relaying in the background (before the job starts, so its first
16
/// connection finds the proxy listening).
17
pub fn start() -> Result<(), String> {
18
    let listener = TcpListener::bind(LISTEN).map_err(|e| format!("egress proxy on {LISTEN}: {e}"))?;
19
    std::thread::spawn(move || {
20
        for conn in listener.incoming().flatten() {
21
            std::thread::spawn(move || relay(conn));
22
        }
23
    });
24
    Ok(())
25
}
26
27
fn vsock_to_host(port: u32) -> std::io::Result<File> {
28
    let fd = unsafe { libc::socket(libc::AF_VSOCK, libc::SOCK_STREAM | libc::SOCK_CLOEXEC, 0) };
29
    if fd < 0 {
30
        return Err(std::io::Error::last_os_error());
31
    }
32
    // Owned from here, so it is closed on every path.
33
    let file = unsafe { File::from_raw_fd(fd) };
34
    let mut addr: libc::sockaddr_vm = unsafe { std::mem::zeroed() };
35
    addr.svm_family = libc::AF_VSOCK as libc::sa_family_t;
36
    addr.svm_cid = libc::VMADDR_CID_HOST;
37
    addr.svm_port = port;
38
    let rc = unsafe {
39
        libc::connect(
40
            fd,
41
            &addr as *const libc::sockaddr_vm as *const libc::sockaddr,
42
            std::mem::size_of::<libc::sockaddr_vm>() as libc::socklen_t,
43
        )
44
    };
45
    if rc != 0 {
46
        return Err(std::io::Error::last_os_error());
47
    }
48
    Ok(file)
49
}
50
51
fn relay(tcp: TcpStream) {
52
    let Ok(host) = vsock_to_host(VSOCK_PORT) else {
53
        let mut tcp = tcp;
54
        let _ = tcp.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\nthe runner's egress proxy did not answer\n");
55
        return;
56
    };
57
    let (Ok(mut tcp_r), Ok(mut host_w)) = (tcp.try_clone(), host.try_clone()) else { return };
58
    let up = std::thread::spawn(move || {
59
        copy(&mut tcp_r, &mut host_w);
60
        unsafe { libc::shutdown(host_w.as_raw_fd(), libc::SHUT_WR) };
61
    });
62
    let (mut host_r, mut tcp_w) = (host, tcp);
63
    copy(&mut host_r, &mut tcp_w);
64
    let _ = tcp_w.shutdown(std::net::Shutdown::Write);
65
    let _ = up.join();
66
}
67
68
fn copy(from: &mut impl Read, to: &mut impl Write) {
69
    let mut buf = [0u8; 16 * 1024];
70
    loop {
71
        match from.read(&mut buf) {
72
            Ok(0) | Err(_) => return,
73
            Ok(n) => {
74
                if to.write_all(&buf[..n]).is_err() {
75
                    return;
76
                }
77
            }
78
        }
79
    }
80
}