1#![cfg(target_os = "macos")]
2
3use std::collections::VecDeque;
16use std::io::{self, Read, Write};
17use std::net::{Ipv4Addr, SocketAddrV4, TcpListener, TcpStream, UdpSocket};
18use std::os::fd::{FromRawFd, IntoRawFd, RawFd};
19use std::os::unix::net::UnixDatagram;
20use std::path::{Path, PathBuf};
21use std::sync::{
22 atomic::{AtomicBool, Ordering},
23 Arc,
24};
25use std::time::Duration;
26
27use a3s_box_core::error::{BoxError, Result};
28use smoltcp::iface::{Config, Interface, SocketSet};
29use smoltcp::socket::{tcp, udp};
30use smoltcp::time::Instant;
31use smoltcp::wire::{EthernetAddress, IpAddress, IpCidr, IpEndpoint, Ipv4Address};
32
33fn smoltcp_now() -> Instant {
37 use std::time::{SystemTime, UNIX_EPOCH};
38 let us = SystemTime::now()
39 .duration_since(UNIX_EPOCH)
40 .unwrap_or_default()
41 .as_micros() as i64;
42 Instant::from_micros(us)
43}
44
45fn to_smoltcp_ipv4(ip: Ipv4Addr) -> Ipv4Address {
47 Ipv4Address::from(ip)
48}
49
50const GATEWAY_MAC: EthernetAddress = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
54const MAX_FRAME: usize = 1514;
56const EPHEMERAL_BASE: u16 = 49152;
58
59struct UnixgramDevice {
72 socket: UnixDatagram,
73 rx_queue: VecDeque<Vec<u8>>,
74}
75
76impl UnixgramDevice {
77 fn drain(&mut self) {
79 let mut buf = vec![0u8; MAX_FRAME];
80 for _ in 0..64 {
81 match self.socket.recv(&mut buf) {
82 Ok(n) => {
83 tracing::trace!(
84 bytes = n,
85 "NetProxy received ethernet frame from guest/libkrun"
86 );
87 self.rx_queue.push_back(buf[..n].to_vec())
88 }
89 Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
90 Err(e) => {
91 tracing::warn!(error = %e, "NetProxy: recv from libkrun failed");
92 break;
93 }
94 }
95 }
96 }
97}
98
99struct OwnedRxToken(Vec<u8>);
101
102impl smoltcp::phy::RxToken for OwnedRxToken {
103 fn consume<R, F>(mut self, f: F) -> R
104 where
105 F: FnOnce(&mut [u8]) -> R,
106 {
107 f(&mut self.0)
108 }
109}
110
111struct TxToken {
116 socket: UnixDatagram,
117}
118
119impl smoltcp::phy::TxToken for TxToken {
120 fn consume<R, F>(self, len: usize, f: F) -> R
121 where
122 F: FnOnce(&mut [u8]) -> R,
123 {
124 let mut buf = vec![0u8; len];
125 let result = f(&mut buf);
126 tracing::trace!(
127 bytes = len,
128 "NetProxy sending ethernet frame to guest/libkrun"
129 );
130 if let Err(e) = self.socket.send(&buf) {
131 tracing::warn!(error = %e, len, "NetProxy: send to libkrun failed");
132 }
133 result
134 }
135}
136
137impl smoltcp::phy::Device for UnixgramDevice {
138 type RxToken<'a>
139 = OwnedRxToken
140 where
141 Self: 'a;
142 type TxToken<'a>
143 = TxToken
144 where
145 Self: 'a;
146
147 fn receive(&mut self, _ts: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
148 let frame = self.rx_queue.pop_front()?;
149 let tx = TxToken {
150 socket: self.socket.try_clone().ok()?,
151 };
152 Some((OwnedRxToken(frame), tx))
153 }
154
155 fn transmit(&mut self, _ts: Instant) -> Option<Self::TxToken<'_>> {
156 Some(TxToken {
157 socket: self.socket.try_clone().ok()?,
158 })
159 }
160
161 fn capabilities(&self) -> smoltcp::phy::DeviceCapabilities {
162 let mut caps = smoltcp::phy::DeviceCapabilities::default();
163 caps.medium = smoltcp::phy::Medium::Ethernet;
164 caps.max_transmission_unit = MAX_FRAME;
165 caps
166 }
167}
168
169struct PortForward {
173 listener: TcpListener,
174 guest_ip: Ipv4Addr,
175 guest_port: u16,
176 pending: Vec<(smoltcp::iface::SocketHandle, TcpStream)>,
178 active: Vec<(smoltcp::iface::SocketHandle, TcpStream)>,
180}
181
182struct ProxyEngine {
185 device: UnixgramDevice,
186 iface: Interface,
187 sockets: SocketSet<'static>,
188 dns_handle: smoltcp::iface::SocketHandle,
189 dns_servers: Vec<Ipv4Addr>,
190 port_forwards: Vec<PortForward>,
191 next_ephemeral: u16,
192 shutdown: Arc<AtomicBool>,
193}
194
195impl ProxyEngine {
196 fn new(
197 socket: UnixDatagram,
198 gateway_ip: Ipv4Addr,
199 prefix_len: u8,
200 dns_servers: Vec<Ipv4Addr>,
201 port_forwards: Vec<PortForward>,
202 shutdown: Arc<AtomicBool>,
203 ) -> Self {
204 let mut device = UnixgramDevice {
205 socket,
206 rx_queue: VecDeque::new(),
207 };
208
209 let config = Config::new(GATEWAY_MAC.into());
211 let mut iface = Interface::new(config, &mut device, smoltcp_now());
212 iface.update_ip_addrs(|addrs| {
213 let cidr = IpCidr::new(IpAddress::Ipv4(to_smoltcp_ipv4(gateway_ip)), prefix_len);
214 addrs.push(cidr).ok();
215 });
216
217 let mut sockets = SocketSet::new(vec![]);
218
219 let dns_rx = udp::PacketBuffer::new(vec![udp::PacketMetadata::EMPTY; 16], vec![0u8; 8192]);
221 let dns_tx = udp::PacketBuffer::new(vec![udp::PacketMetadata::EMPTY; 16], vec![0u8; 8192]);
222 let mut dns_socket = udp::Socket::new(dns_rx, dns_tx);
223 dns_socket.bind(53).ok();
224 let dns_handle = sockets.add(dns_socket);
225
226 Self {
227 device,
228 iface,
229 sockets,
230 dns_handle,
231 dns_servers,
232 port_forwards,
233 next_ephemeral: EPHEMERAL_BASE,
234 shutdown,
235 }
236 }
237
238 fn run(&mut self) {
239 loop {
240 if self.shutdown.load(Ordering::Relaxed) {
241 break;
242 }
243
244 let now = smoltcp_now();
245
246 self.device.drain();
248
249 self.accept_connections(now);
251
252 self.iface.poll(now, &mut self.device, &mut self.sockets);
254
255 self.promote_established();
257
258 self.proxy_data();
260
261 self.forward_dns();
263
264 self.cleanup();
266
267 let delay = self
269 .iface
270 .poll_delay(now, &self.sockets)
271 .unwrap_or(smoltcp::time::Duration::from_millis(1));
272 std::thread::sleep(Duration::from_micros(delay.micros().min(5_000)));
273 }
274 }
275
276 fn accept_connections(&mut self, now: Instant) {
279 let mut new_conns: Vec<(usize, TcpStream, Ipv4Addr, u16)> = Vec::new();
282 for (i, pf) in self.port_forwards.iter_mut().enumerate() {
283 loop {
284 match pf.listener.accept() {
285 Ok((stream, _)) => {
286 stream.set_nonblocking(true).ok();
287 stream.set_nodelay(true).ok();
288 new_conns.push((i, stream, pf.guest_ip, pf.guest_port));
289 }
290 Err(e) if e.kind() == io::ErrorKind::WouldBlock => break,
291 Err(e) => {
292 tracing::warn!(error = %e, "Port-forward listener error");
293 break;
294 }
295 }
296 }
297 }
298
299 for (i, stream, guest_ip, guest_port) in new_conns {
301 let handle = self.open_guest_tcp(guest_ip, guest_port, now);
302 self.port_forwards[i].pending.push((handle, stream));
303 tracing::debug!(
304 guest = %guest_ip,
305 port = guest_port,
306 handle = ?handle,
307 "NetProxy accepted host connection and initiated guest TCP connect"
308 );
309 }
310 }
311
312 fn open_guest_tcp(
314 &mut self,
315 guest_ip: Ipv4Addr,
316 guest_port: u16,
317 _now: Instant,
318 ) -> smoltcp::iface::SocketHandle {
319 let rx = tcp::SocketBuffer::new(vec![0u8; 65536]);
320 let tx = tcp::SocketBuffer::new(vec![0u8; 65536]);
321 let mut socket = tcp::Socket::new(rx, tx);
322
323 let local_port = self.next_ephemeral;
324 self.next_ephemeral = self.next_ephemeral.wrapping_add(1);
325 if self.next_ephemeral < EPHEMERAL_BASE {
326 self.next_ephemeral = EPHEMERAL_BASE;
327 }
328
329 let remote = IpEndpoint::new(IpAddress::Ipv4(to_smoltcp_ipv4(guest_ip)), guest_port);
330 socket
331 .connect(self.iface.context(), remote, local_port)
332 .ok();
333 socket.set_keep_alive(Some(smoltcp::time::Duration::from_secs(30)));
334
335 self.sockets.add(socket)
336 }
337
338 fn promote_established(&mut self) {
341 for pf in &mut self.port_forwards {
342 let mut still_pending = Vec::new();
343 for (handle, stream) in pf.pending.drain(..) {
344 let socket = self.sockets.get::<tcp::Socket>(handle);
345 use smoltcp::socket::tcp::State;
346 match socket.state() {
347 State::Established => {
348 tracing::debug!(handle = ?handle, "NetProxy guest TCP connection established");
349 pf.active.push((handle, stream));
350 }
351 State::Closed | State::TimeWait | State::CloseWait => {
352 tracing::debug!(handle = ?handle, state = ?socket.state(), "NetProxy guest TCP connection closed before establishment");
353 drop(stream);
355 self.sockets.remove(handle);
356 }
357 _ => {
358 still_pending.push((handle, stream));
359 }
360 }
361 }
362 pf.pending = still_pending;
363 }
364 }
365
366 fn proxy_data(&mut self) {
369 for pf in &mut self.port_forwards {
370 for (handle, host_stream) in &mut pf.active {
371 let socket = self.sockets.get_mut::<tcp::Socket>(*handle);
372
373 if socket.can_recv() {
375 socket
376 .recv(|data| {
377 tracing::trace!(handle = ?*handle, bytes = data.len(), "NetProxy forwarding guest -> host bytes");
378 let _ = host_stream.write_all(data);
379 (data.len(), ())
380 })
381 .ok();
382 }
383
384 if socket.can_send() {
386 let mut buf = [0u8; 8192];
387 match host_stream.read(&mut buf) {
388 Ok(0) => {
389 tracing::debug!(handle = ?*handle, "NetProxy host side closed connection");
390 socket.close();
391 }
392 Ok(n) => {
393 tracing::trace!(handle = ?*handle, bytes = n, "NetProxy forwarding host -> guest bytes");
394 socket.send_slice(&buf[..n]).ok();
395 }
396 Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
397 Err(_) => {
398 socket.close();
399 }
400 }
401 }
402 }
403 }
404 }
405
406 fn forward_dns(&mut self) {
409 let dns_server = match self.dns_servers.first() {
410 Some(s) => *s,
411 None => return,
412 };
413
414 let socket = self.sockets.get_mut::<udp::Socket>(self.dns_handle);
415 if !socket.can_recv() {
416 return;
417 }
418 let (query, src_endpoint) = match socket.recv() {
419 Ok(r) => r,
420 Err(_) => return,
421 };
422 let query = query.to_vec();
423 let src = src_endpoint;
424
425 match UdpSocket::bind("0.0.0.0:0") {
427 Ok(udp) => {
428 udp.set_read_timeout(Some(Duration::from_secs(2))).ok();
429 let dest = SocketAddrV4::new(dns_server, 53);
430 if udp.send_to(&query, dest).is_ok() {
431 let mut resp = vec![0u8; 4096];
432 if let Ok((n, _)) = udp.recv_from(&mut resp) {
433 let socket = self.sockets.get_mut::<udp::Socket>(self.dns_handle);
434 socket.send_slice(&resp[..n], src).ok();
435 }
436 }
437 }
438 Err(e) => {
439 tracing::warn!(error = %e, "DNS forward: failed to bind host UDP socket");
440 }
441 }
442 }
443
444 fn cleanup(&mut self) {
447 use smoltcp::socket::tcp::State;
448 for pf in &mut self.port_forwards {
449 let mut to_remove = Vec::new();
451 pf.active.retain(|(handle, _stream)| {
452 let state = self.sockets.get::<tcp::Socket>(*handle).state();
453 match state {
454 State::Closed | State::TimeWait | State::CloseWait => {
455 to_remove.push(*handle);
456 false
457 }
458 _ => true,
459 }
460 });
461 for h in to_remove {
462 self.sockets.remove(h);
463 }
464 }
465 }
466}
467
468pub struct NetProxyManager {
474 socket_path: PathBuf,
475 net_socket_fd: Option<RawFd>,
476 net_proxy_fd: Option<RawFd>,
477}
478
479impl NetProxyManager {
480 pub fn new(box_dir: &Path) -> Self {
483 Self {
484 socket_path: box_dir.join("sockets").join("net.sock"),
485 net_socket_fd: None,
486 net_proxy_fd: None,
487 }
488 }
489
490 pub fn socket_path(&self) -> &Path {
491 &self.socket_path
492 }
493
494 pub fn net_socket_fd(&self) -> Option<RawFd> {
495 self.net_socket_fd
496 }
497
498 pub fn net_proxy_fd(&self) -> Option<RawFd> {
499 self.net_proxy_fd
500 }
501
502 pub fn spawn(
507 &mut self,
508 _ip: Ipv4Addr,
509 _gateway: Ipv4Addr,
510 _prefix_len: u8,
511 _dns_servers: &[Ipv4Addr],
512 _port_map: &[String],
513 ) -> Result<()> {
514 let (proxy_socket, krun_fd) = socketpair_unixgram()?;
515 self.net_socket_fd = Some(krun_fd);
516 self.net_proxy_fd = Some(proxy_socket.into_raw_fd());
517 Ok(())
518 }
519
520 pub fn stop(&mut self) {
521 if let Some(fd) = self.net_socket_fd.take() {
522 unsafe {
523 libc::close(fd);
524 }
525 }
526 if let Some(fd) = self.net_proxy_fd.take() {
527 unsafe {
528 libc::close(fd);
529 }
530 }
531 std::fs::remove_file(&self.socket_path).ok();
532 }
533
534 pub fn is_running(&mut self) -> bool {
535 self.net_socket_fd.is_some() || self.net_proxy_fd.is_some()
536 }
537}
538
539impl Drop for NetProxyManager {
540 fn drop(&mut self) {
541 self.stop();
542 }
543}
544
545pub fn spawn_inherited_netproxy(
548 fd: RawFd,
549 guest_ip: Ipv4Addr,
550 gateway: Ipv4Addr,
551 prefix_len: u8,
552 dns_servers: &[Ipv4Addr],
553 port_map: &[String],
554) -> Result<()> {
555 let socket = unsafe { UnixDatagram::from_raw_fd(fd) };
556 let port_forwards = parse_port_forwards(port_map, guest_ip)
557 .map_err(|e| BoxError::NetworkError(format!("invalid port_map: {}", e)))?;
558 let dns_servers = dns_servers.to_vec();
559 let shutdown = Arc::new(AtomicBool::new(false));
560
561 std::thread::Builder::new()
562 .name("a3s-netproxy".to_string())
563 .spawn(move || {
564 tracing::info!(fd, gateway = %gateway, guest_ip = %guest_ip, "NetProxy thread started");
565 if let Err(e) = socket.set_nonblocking(true) {
566 tracing::error!(error = %e, "NetProxy: set_nonblocking failed");
567 return;
568 }
569
570 let mut engine = ProxyEngine::new(
571 socket,
572 gateway,
573 prefix_len,
574 dns_servers,
575 port_forwards,
576 shutdown,
577 );
578 engine.run();
579 tracing::info!("NetProxy thread exiting");
580 })
581 .map_err(|e| BoxError::NetworkError(format!("failed to spawn netproxy thread: {}", e)))?;
582
583 Ok(())
584}
585
586fn socketpair_unixgram() -> Result<(UnixDatagram, RawFd)> {
587 let mut fds = [-1; 2];
588 let ret = unsafe { libc::socketpair(libc::AF_UNIX, libc::SOCK_DGRAM, 0, fds.as_mut_ptr()) };
589 if ret != 0 {
590 return Err(BoxError::NetworkError(format!(
591 "failed to create unix datagram socketpair: {}",
592 io::Error::last_os_error()
593 )));
594 }
595
596 let proxy_socket = unsafe { UnixDatagram::from_raw_fd(fds[0]) };
597 Ok((proxy_socket, fds[1]))
598}
599
600fn parse_port_forwards(
605 port_map: &[String],
606 guest_ip: Ipv4Addr,
607) -> std::result::Result<Vec<PortForward>, String> {
608 let mut forwards = Vec::new();
609 for entry in port_map {
610 let mapping = a3s_box_core::parse_port_mapping(entry)?;
611 let host_port = mapping.host_port;
612 let guest_port = mapping.guest_port;
613
614 let listener = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, host_port))
615 .map_err(|e| format!("cannot bind 0.0.0.0:{}: {}", host_port, e))?;
616 listener
617 .set_nonblocking(true)
618 .map_err(|e| format!("set_nonblocking on listener: {}", e))?;
619
620 tracing::info!(
621 host_port,
622 guest_port,
623 guest_ip = %guest_ip,
624 "Port-forward listener ready"
625 );
626 forwards.push(PortForward {
627 listener,
628 guest_ip,
629 guest_port,
630 pending: Vec::new(),
631 active: Vec::new(),
632 });
633 }
634 Ok(forwards)
635}
636
637#[cfg(test)]
640mod tests {
641 use super::*;
642
643 fn port_is_bindable(port: u16) -> bool {
644 TcpListener::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port)).is_ok()
645 }
646
647 fn ports_are_bindable(ports: &[u16]) -> bool {
648 ports.iter().copied().all(port_is_bindable)
649 }
650
651 #[test]
652 fn test_smoltcp_now_returns_reasonable_value() {
653 let now = smoltcp_now();
654 assert!(now.micros() > 0);
656 }
657
658 #[test]
659 fn test_to_smoltcp_ipv4_conversion() {
660 let ip = Ipv4Addr::new(10, 88, 0, 1);
661 let smol_ip = to_smoltcp_ipv4(ip);
662 assert_eq!(smol_ip.as_bytes(), &[10, 88, 0, 1]);
663 }
664
665 #[test]
666 fn test_to_smoltcp_ipv4_loopback() {
667 let ip = Ipv4Addr::new(127, 0, 0, 1);
668 let smol_ip = to_smoltcp_ipv4(ip);
669 assert_eq!(smol_ip.as_bytes(), &[127, 0, 0, 1]);
670 }
671
672 #[test]
673 fn test_parse_port_forwards_empty_rules() {
674 let guest = Ipv4Addr::new(10, 89, 0, 2);
675 let fwds = parse_port_forwards(&[], guest).unwrap();
676 assert!(fwds.is_empty());
677 }
678
679 #[test]
680 fn test_parse_port_forwards_rejects_udp_suffix() {
681 let guest = Ipv4Addr::new(10, 89, 0, 2);
682 let rules = vec!["19990:80/udp".to_string()];
683 let error = match parse_port_forwards(&rules, guest) {
684 Ok(_) => panic!("UDP port mapping unexpectedly succeeded"),
685 Err(error) => error,
686 };
687
688 assert!(error.contains("only TCP is supported"));
689 }
690
691 #[test]
692 fn test_parse_port_forwards_multiple_rules() {
693 let guest = Ipv4Addr::new(10, 89, 0, 2);
694 if !ports_are_bindable(&[19991, 19992, 19993]) {
695 eprintln!("skipping test: one or more host ports are not bindable");
696 return;
697 }
698 let rules = vec![
699 "19991:80".to_string(),
700 "19992:443".to_string(),
701 "19993:8080".to_string(),
702 ];
703 let fwds = parse_port_forwards(&rules, guest).unwrap();
704 assert_eq!(fwds.len(), 3);
705 assert_eq!(fwds[0].guest_port, 80);
706 assert_eq!(fwds[1].guest_port, 443);
707 assert_eq!(fwds[2].guest_port, 8080);
708 }
709
710 #[test]
711 fn test_parse_port_forwards_empty_string() {
712 let guest = Ipv4Addr::new(10, 89, 0, 2);
713 let rules = vec!["".to_string()];
715 let result = parse_port_forwards(&rules, guest);
716 assert!(result.is_err());
717 }
718
719 #[test]
720 fn test_netproxy_manager_new() {
721 let dir = tempfile::tempdir().unwrap();
722 let mgr = NetProxyManager::new(dir.path());
723 assert_eq!(
724 mgr.socket_path(),
725 dir.path().join("sockets").join("net.sock")
726 );
727 assert_eq!(mgr.net_socket_fd(), None);
728 }
729
730 #[test]
731 fn test_netproxy_manager_not_running_initially() {
732 let dir = tempfile::tempdir().unwrap();
733 let mut mgr = NetProxyManager::new(dir.path());
734 assert!(!mgr.is_running());
735 }
736
737 #[test]
738 fn test_netproxy_manager_stop_when_not_started() {
739 let dir = tempfile::tempdir().unwrap();
740 let mut mgr = NetProxyManager::new(dir.path());
741 mgr.stop(); assert!(!mgr.is_running());
743 }
744
745 #[test]
746 fn test_netproxy_manager_drop_cleans_up() {
747 let dir = tempfile::tempdir().unwrap();
748 let socket_path = dir.path().join("sockets").join("net.sock");
749 std::fs::create_dir_all(dir.path().join("sockets")).unwrap();
750 std::fs::write(&socket_path, "fake").unwrap();
751 {
752 let _mgr = NetProxyManager::new(dir.path());
753 }
755 assert!(!socket_path.exists());
756 }
757
758 #[test]
759 fn test_parse_port_forwards_valid() {
760 let guest = Ipv4Addr::new(10, 89, 0, 2);
761 if !ports_are_bindable(&[19988, 19443]) {
762 eprintln!("skipping test: one or more host ports are not bindable");
763 return;
764 }
765 let rules = vec!["19988:80".to_string(), "19443:443".to_string()];
767 let fwds = parse_port_forwards(&rules, guest).unwrap();
768 assert_eq!(fwds.len(), 2);
769 assert_eq!(fwds[0].guest_port, 80);
770 assert_eq!(fwds[1].guest_port, 443);
771 }
772
773 #[test]
774 fn test_parse_port_forwards_with_protocol_suffix() {
775 let guest = Ipv4Addr::new(10, 89, 0, 2);
776 if !port_is_bindable(19989) {
777 eprintln!("skipping test: host port 19989 is not bindable");
778 return;
779 }
780 let rules = vec!["19989:80/tcp".to_string()];
781 let fwds = parse_port_forwards(&rules, guest).unwrap();
782 assert_eq!(fwds[0].guest_port, 80);
783 }
784
785 #[test]
786 fn test_parse_port_forwards_invalid_format() {
787 let guest = Ipv4Addr::new(10, 89, 0, 2);
788 assert!(parse_port_forwards(&["notaport".to_string()], guest).is_err());
789 assert!(parse_port_forwards(&["abc:80".to_string()], guest).is_err());
790 assert!(parse_port_forwards(&["80:xyz".to_string()], guest).is_err());
791 }
792
793 }