mirrors/crosvm
1
0
Fork 0
mirror of https://chromium.googlesource.com/crosvm/crosvm synced 2025-02-05 10:10:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

third_party: libslirp-rs: remove unused files.

Browse source 
These files are not used, and libslirp-rs is not a functional crate.
Only context.rs is used in heavily forked form as a part of crosvm's `net_util`.

BUG=b:213151463
TEST=n/a

Change-Id: I6536cde7ff2beed655e83c3d920c6940ed4969a6
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/crosvm/+/3650467
Tested-by: kokoro <noreply+kokoro@google.com>
Reviewed-by: Keiichi Watanabe <keiichiw@chromium.org>
Commit-Queue: Noah Gold <nkgold@google.com>
This commit is contained in:
Noah Gold 2022-05-17 10:58:46 -07:00 committed by Chromeos LUCI
parent 1977979fbb
commit a89e65beb1
14 changed files with 0 additions and 1598 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
third_party/libslirp-rs/Cargo.toml vendored
View file

@ -1,43 +0,0 @@
[package]
name = "libslirp"
version = "4.3.0"
authors = ["Marc-André Lureau <marcandre.lureau@redhat.com>"]
repository = "https://gitlab.freedesktop.org/slirp/libslirp-rs.git"
homepage = "https://gitlab.freedesktop.org/slirp/libslirp-rs"
documentation = "https://docs.rs/libslirp"
description = "High-level bindings & helper process for libslirp."
license-file = "LICENSE"
edition = "2018"
keywords = ["networking", "tcp", "ip", "qemu", "virtualization"]
categories = ["api-bindings", "command-line-utilities", "emulators", "network-programming"]
[features]
default = ["mio", "mio-extras", "ipnetwork", "structopt", "slab"]
helper = ["libc", "zbus", "nix", "libsystemd", "url", "lazy_static", "zvariant", "enumflags2"]
[dependencies]
libslirp-sys = { version = "4.2.0" }
ipnetwork = { version = "0.17", optional = true }
structopt = { version = "0.3.0", optional = true }
mio = { version = "0.6.19", optional = true }
mio-extras = { version = "2.0.5", optional = true }
slab = { version = "0.4.0", optional = true }
libc = { version = "0.2", optional = true }
nix = { version = "0.17", optional = true }
libsystemd = { version = "0.3", optional = true }
url = { version = "2.1", optional = true }
lazy_static = { version = "1.4", optional = true }
zbus = { version = "1.0", optional = true }
zvariant = { version = "2.0", optional = true }
enumflags2 = { version = "0.6.4", optional = true }
[dev-dependencies]
etherparse = "0.8.0"
[[bin]]
name = "libslirp-helper"
required-features = ["default", "helper"]
[[test]]
name = "test-ip"
required-features = ["default"]

16
third_party/libslirp-rs/Makefile vendored
View file

@ -1,16 +0,0 @@
CARGO = cargo
CARGOFLAGS =
SLIRPHELPER = target/debug/libslirp-helper
.PHONY: $(SLIRPHELPER)
$(SLIRPHELPER):
$(CARGO) build --all-features $(CARGOFLAGS)
.PHONY: test
test: $(SLIRPHELPER)
SLIRPHELPER=$(SLIRPHELPER) \
PYTHONPATH=. \
PYTHONIOENCODING=utf-8 \
unshare -Ur \
dbus-run-session --config-file=tests/dbus.conf \
python3 -m unittest -v

319
third_party/libslirp-rs/src/bin/libslirp-helper/main.rs vendored
View file

@ -1,319 +0,0 @@
use std::convert::TryInto;
use std::error::Error;
use std::fs::File;
use std::io::{self, Cursor, Read, Write};
use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
use std::os::unix::net::UnixDatagram;
use std::path::PathBuf;
use std::process;
use std::rc::Rc;
use enumflags2::BitFlags;
#[cfg(feature = "libsystemd")]
use libsystemd::daemon::{self, NotifyState};
use mio::unix::EventedFd;
use mio::*;
use nix::sched::{setns, CloneFlags};
use structopt::{clap::ArgGroup, StructOpt};
use zbus::dbus_interface;
#[macro_use]
extern crate lazy_static;
mod tun;
#[derive(Debug, StructOpt)]
#[structopt(
name = "libslirp-helper",
about = "slirp helper process",
rename_all = "kebab-case",
group = ArgGroup::with_name("verb").required(true)
)]
struct Opt {
/// Activate debug mode
#[structopt(long)]
debug: bool,
/// Print capabilities
#[structopt(long, group = "verb")]
print_capabilities: bool,
/// Exit with parent process
#[structopt(long)]
exit_with_parent: bool,
/// DBus bus address
#[structopt(long)]
dbus_address: Option<String>,
/// Helper instance ID
#[structopt(long, name = "id")]
dbus_id: Option<String>,
/// Incoming migration data from DBus
#[structopt(long)]
dbus_incoming: bool,
/// Unix datagram socket path
#[structopt(long, parse(from_os_str), group = "verb")]
socket_path: Option<PathBuf>,
/// Unix datagram socket file descriptor
#[structopt(long, group = "verb")]
fd: Option<i32>,
/// Incoming migration data
#[structopt(long)]
incoming_fd: Option<i32>,
/// Set DHCP NBP URL (ex: tftp://10.0.0.1/my-nbp)
#[structopt(long, name = "url")]
dhcp_nbp: Option<String>,
/// Path to network namespace to join.
#[structopt(long)]
netns: Option<PathBuf>,
/// Interface name, such as "tun0".
#[structopt(long, group = "verb")]
interface: Option<String>,
#[structopt(flatten)]
slirp: libslirp::Opt,
}
fn set_exit_with_parent() {
#[cfg(any(target_os = "linux", target_os = "android"))]
unsafe {
libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM, 0, 0, 0);
}
}
const DBUS_TOKEN: Token = Token(10_000_000);
fn slirp_state_read<'a, R: Read>(
slirp: &libslirp::MioHandler<'a>,
reader: &mut R,
) -> Result<(), Box<dyn Error>> {
let mut buf = [0; 4];
reader.read(&mut buf)?;
let in_version = i32::from_be_bytes(buf);
if in_version > libslirp::state_version() {
return Err(format!(
"Incompatible migration data version: {} > {}",
in_version,
libslirp::state_version()
)
.into());
}
slirp.ctxt.state_read(in_version, reader)?;
slirp.register();
Ok(())
}
fn print_capabilities() -> Result<(), Box<dyn Error>> {
io::stdout().write_all(
r#"{
"type": "slirp-helper",
"features": [
"dbus-address",
"dhcp",
"exit-with-parent",
"migrate",
"tftp",
"ipv4",
"ipv6",
"netns",
"notify-socket",
"restrict"
]
}
"#
.as_bytes(),
)?;
Ok(())
}
fn set_netns(fd: RawFd) -> Result<(), nix::Error> {
setns(fd, CloneFlags::CLONE_NEWNET)
}
lazy_static! {
// XXX: when do we get async yet?
static ref POLL: Poll = Poll::new().unwrap();
}
struct Slirp1 {
slirp: Rc<libslirp::MioHandler<'static>>,
}
#[dbus_interface(name = "org.freedesktop.Slirp1.Helper")]
impl Slirp1 {
fn get_info(&self) -> String {
self.slirp.ctxt.connection_info().to_string()
}
}
struct VMState1 {
id: String,
slirp: Rc<libslirp::MioHandler<'static>>,
}
#[dbus_interface(name = "org.qemu.VMState1")]
impl VMState1 {
fn save(&self) -> zbus::fdo::Result<Vec<u8>> {
let mut data = libslirp::state_version().to_be_bytes().to_vec();
let mut state = self
.slirp
.ctxt
.state_get()
.map_err(|e| zbus::fdo::Error::Failed(format!("Failed to save: {}", e)))?;
data.append(&mut state);
Ok(data)
}
fn load(&self, data: &[u8]) -> zbus::fdo::Result<()> {
let mut data = Cursor::new(data);
Ok(slirp_state_read(&self.slirp, &mut data)
.map_err(|e| zbus::fdo::Error::Failed(format!("Failed to load: {}", e)))?)
}
#[dbus_interface(property)]
fn id(&self) -> &str {
&self.id
}
}
fn main() -> Result<(), Box<dyn Error>> {
let m = Opt::clap().get_matches();
let mut opt = Opt::from_clap(&m);
if opt.debug {
dbg!(&opt);
}
if opt.print_capabilities {
return print_capabilities();
}
if m.occurrences_of("dhcp-start") == 0 {
let dhcp_start = opt.slirp.ipv4.net.nth(15).expect("Invalid --net");
opt.slirp.ipv4.dhcp_start = dhcp_start;
}
if let Some(url) = &opt.dhcp_nbp {
let url = url::Url::parse(url)?;
if url.scheme() != "tftp" {
return Err("Invalid NBP URL".into());
}
opt.slirp.tftp.name = Some(url.host_str().unwrap().to_string());
opt.slirp.tftp.bootfile = Some(url.path().to_string());
}
let mut main_netns = None;
if let Some(netns) = &opt.netns {
main_netns = Some(File::open("/proc/self/ns/net")?);
let netns = File::open(netns)?;
set_netns(netns.as_raw_fd())?;
opt.interface.get_or_insert("tun0".to_string());
}
let stream = match &opt {
Opt { fd: Some(fd), .. } => unsafe { UnixDatagram::from_raw_fd(*fd) },
Opt {
socket_path: Some(path),
..
} => UnixDatagram::bind(path)?,
Opt {
interface: Some(tun),
..
} => tun::open(tun)?,
_ => return Err("Missing a socket argument".into()),
};
if let Some(netns) = main_netns {
set_netns(netns.as_raw_fd())?;
}
if opt.exit_with_parent {
set_exit_with_parent();
}
let slirp = Rc::new(libslirp::MioHandler::new(&opt.slirp, &POLL, stream));
let dbus = if let Some(dbus_addr) = opt.dbus_address {
if opt.dbus_id.is_none() {
return Err("You must specify an id with DBus".into());
}
let c = zbus::Connection::new_for_address(&dbus_addr, true)?;
zbus::fdo::DBusProxy::new(&c)?.request_name(
&format!("org.freedesktop.Slirp1_{}", process::id()),
BitFlags::empty(),
)?;
zbus::fdo::DBusProxy::new(&c)?.request_name("org.qemu.VMState1", BitFlags::empty())?;
let dbus_fd = c.as_raw_fd();
POLL.register(
&EventedFd(&dbus_fd),
DBUS_TOKEN,
Ready::readable(),
PollOpt::level(),
)?;
Some(c)
} else {
None
};
let mut s = if let Some(c) = &dbus {
let mut s = zbus::ObjectServer::new(c);
s.at(
&"/org/freedesktop/Slirp1/Helper".try_into()?,
Slirp1 {
slirp: slirp.clone(),
},
)?;
s.at(
&"/org/qemu/VMState1".try_into()?,
VMState1 {
id: opt.dbus_id.unwrap(),
slirp: slirp.clone(),
},
)?;
Some(s)
} else {
None
};
if opt.dbus_incoming && opt.incoming_fd.is_some() {
return Err("Invalid multiple incoming paths.".into());
}
let mut events = Events::with_capacity(1024);
let mut duration = None;
if let Some(fd) = opt.incoming_fd {
let mut f = unsafe { File::from_raw_fd(fd) };
slirp_state_read(&slirp, &mut f)?;
} else if !opt.dbus_incoming {
slirp.register();
}
#[cfg(feature = "libsystemd")]
daemon::notify(true, &[NotifyState::Ready])?;
loop {
if opt.debug {
dbg!(duration);
}
POLL.poll(&mut events, duration)?;
duration = slirp.dispatch(&events)?;
if let Some(dbus) = &dbus {
for event in &events {
match event.token() {
DBUS_TOKEN => {
let m = dbus.receive_message()?;
if let Err(e) = s.as_mut().unwrap().dispatch_message(&m) {
eprintln!("{}", e);
}
}
_ => {
continue;
}
}
}
}
}
}

75
third_party/libslirp-rs/src/bin/libslirp-helper/tun.rs vendored
View file

@ -1,75 +0,0 @@
use nix::fcntl::OFlag;
use nix::ioctl_write_ptr;
use nix::sys::stat::Mode;
use std::error::Error;
use std::os::raw::c_short;
use std::os::unix::io::FromRawFd;
use std::os::unix::net::UnixDatagram;
//pub const IFF_TUN: c_short = 0x0001;
pub const IFF_TAP: c_short = 0x0002;
pub const IFF_NO_PI: c_short = 0x1000;
const INTERFACE_NAME_SIZE: usize = 16;
const INTERFACE_REQUEST_UNION_SIZE: usize = 24;
const TUN_MAGIC: u8 = b'T';
const TUN_SETIFF: u8 = 202;
#[repr(C)]
#[derive(Default)]
pub struct InterfaceRequest {
pub interface_name: [u8; INTERFACE_NAME_SIZE],
pub union: InterfaceRequestUnion,
}
impl InterfaceRequest {
pub fn with_interface_name(name: &str) -> Result<Self, Box<dyn Error>> {
let mut interface_request: Self = Default::default();
interface_request.set_interface_name(name)?;
Ok(interface_request)
}
pub fn set_interface_name(&mut self, name: &str) -> Result<(), Box<dyn Error>> {
let name_len = name.len();
let mut name = Vec::from(name);
if name_len < INTERFACE_NAME_SIZE {
name.resize(INTERFACE_NAME_SIZE, 0);
} else {
return Err("interface name too long".into());
}
assert_eq!(name.len(), INTERFACE_NAME_SIZE);
self.interface_name.clone_from_slice(&name);
Ok(())
}
}
#[repr(C)]
pub union InterfaceRequestUnion {
pub data: [u8; INTERFACE_REQUEST_UNION_SIZE],
pub flags: c_short,
}
impl Default for InterfaceRequestUnion {
fn default() -> Self {
InterfaceRequestUnion {
data: Default::default(),
}
}
}
ioctl_write_ptr!(tun_set_iff, TUN_MAGIC, TUN_SETIFF, libc::c_int);
pub fn open(name: &str) -> Result<UnixDatagram, Box<dyn Error>> {
let flags = IFF_TAP | IFF_NO_PI;
let fd = nix::fcntl::open("/dev/net/tun", OFlag::O_RDWR, Mode::empty())?;
let mut ifr = InterfaceRequest::with_interface_name(name)?;
ifr.union.flags = flags;
unsafe { tun_set_iff(fd, &mut ifr as *mut InterfaceRequest as *mut i32) }?;
Ok(unsafe { UnixDatagram::from_raw_fd(fd) })
}

13
third_party/libslirp-rs/src/lib.rs vendored
View file

@ -1,13 +0,0 @@
pub mod context;
#[cfg(all(feature = "structopt", feature = "mio"))]
pub mod mio;
#[cfg(feature = "structopt")]
pub mod opt;
pub mod version;
pub use self::context::{Context, Handler, PollEvents};
#[cfg(all(feature = "structopt", feature = "mio"))]
pub use self::mio::*;
#[cfg(feature = "structopt")]
pub use self::opt::*;
pub use self::version::{state_version, version};

312
third_party/libslirp-rs/src/mio.rs vendored
View file

@ -1,312 +0,0 @@
use crate::context::{Context, Handler, PollEvents};
use crate::opt::Opt;
use mio::unix::{EventedFd, UnixReady};
use mio::*;
use mio_extras::timer::Timer as MioTimer;
use slab::Slab;
use std::cell::RefCell;
use std::fmt;
use std::fs::File;
use std::io;
use std::io::prelude::*;
use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
use std::os::unix::net::UnixDatagram;
use std::rc::Rc;
use std::time::{Duration, Instant};
struct MyTimer {
func: Rc<RefCell<Box<dyn FnMut()>>>,
timer: MioTimer<()>,
}
impl fmt::Debug for MyTimer {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "MyTimer {{}}")
}
}
#[derive(Debug)]
struct MyFd {
fd: RawFd,
events: PollEvents,
revents: Option<PollEvents>,
}
impl MyFd {
fn new(fd: RawFd, events: PollEvents) -> Self {
Self {
events,
fd,
revents: None,
}
}
}
#[derive(Debug)]
enum MyToken {
Fd(MyFd),
Timer(MyTimer),
}
pub struct Inner<'a> {
start: Instant,
stream: UnixDatagram,
poll: &'a Poll,
tokens: Slab<MyToken>,
}
pub struct MioHandler<'a> {
inner: Rc<RefCell<Inner<'a>>>,
pub ctxt: Context<Rc<RefCell<Inner<'a>>>>,
}
impl<'a> Handler for Inner<'a> {
type Timer = usize;
fn clock_get_ns(&mut self) -> i64 {
const NANOS_PER_SEC: u64 = 1_000_000_000;
let d = self.start.elapsed();
(d.as_secs() * NANOS_PER_SEC + d.subsec_nanos() as u64) as i64
}
fn timer_new(&mut self, func: Box<dyn FnMut()>) -> Box<Self::Timer> {
let timer = MioTimer::default();
let tok = self.tokens.insert(MyToken::Timer(MyTimer {
func: Rc::new(RefCell::new(func)),
timer,
}));
let timer = match &self.tokens[tok] {
MyToken::Timer(MyTimer { timer: t, .. }) => t,
_ => panic!(),
};
self.poll
.register(timer, Token(tok), Ready::readable(), PollOpt::edge())
.unwrap();
Box::new(tok)
}
fn timer_mod(&mut self, timer: &mut Box<Self::Timer>, expire_time: i64) {
let when = Duration::from_millis(expire_time as u64);
let timer = match &mut self.tokens[**timer] {
MyToken::Timer(MyTimer { timer: t, .. }) => t,
_ => panic!(),
};
timer.set_timeout(when, ());
}
fn timer_free(&mut self, timer: Box<Self::Timer>) {
let t = match &self.tokens[*timer] {
MyToken::Timer(MyTimer { timer: t, .. }) => t,
_ => panic!(),
};
self.poll.deregister(t).unwrap();
self.tokens.remove(*timer);
drop(timer); // for clarity
}
fn send_packet(&mut self, buf: &[u8]) -> io::Result<usize> {
self.stream.send(buf)
}
fn guest_error(&mut self, msg: &str) {
eprintln!("guest error: {}", msg);
}
fn register_poll_fd(&mut self, _fd: RawFd) {}
fn unregister_poll_fd(&mut self, _fd: RawFd) {}
fn notify(&mut self) {}
}
fn to_mio_ready(events: PollEvents) -> mio::Ready {
let mut ready = UnixReady::from(Ready::empty());
if events.has_in() {
ready.insert(Ready::readable());
}
if events.has_out() {
ready.insert(Ready::writable());
}
if events.has_hup() {
ready.insert(UnixReady::hup());
}
if events.has_err() {
ready.insert(UnixReady::error());
}
if events.has_pri() {
ready.insert(UnixReady::priority());
}
Ready::from(ready)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn to_mio_ready_test() {
assert_eq!(to_mio_ready(PollEvents::empty()), Ready::empty());
assert_eq!(to_mio_ready(PollEvents::poll_in()), Ready::readable());
assert_eq!(to_mio_ready(PollEvents::poll_out()), Ready::writable());
assert_eq!(
to_mio_ready(PollEvents::poll_err()),
Ready::from(UnixReady::error())
);
assert_eq!(
to_mio_ready(PollEvents::poll_pri()),
Ready::from(UnixReady::priority())
);
assert_eq!(
to_mio_ready(PollEvents::poll_hup()),
Ready::from(UnixReady::hup())
);
let ev = PollEvents::poll_in() | PollEvents::poll_pri();
let ev = to_mio_ready(ev);
assert!(ev.is_readable());
// bug, see https://github.com/carllerche/mio/pull/897
assert!(!ev.is_writable());
}
}
fn from_mio_ready(ready: mio::Ready) -> PollEvents {
let mut events = PollEvents::empty();
let ready = UnixReady::from(ready);
if ready.is_readable() {
events |= PollEvents::poll_in();
}
if ready.is_writable() {
events |= PollEvents::poll_out();
}
if ready.is_hup() {
events |= PollEvents::poll_hup();
}
if ready.is_error() {
events |= PollEvents::poll_err();
}
if ready.is_priority() {
events |= PollEvents::poll_pri();
}
events
}
const SOCKET: Token = Token(1_000_000);
impl<'a> MioHandler<'a> {
pub fn new(opt: &Opt, poll: &'a Poll, stream: UnixDatagram) -> Self {
let inner = Rc::new(RefCell::new(Inner {
start: Instant::now(),
poll,
stream,
tokens: Slab::with_capacity(1024),
}));
Self {
inner: inner.clone(),
ctxt: Context::new_with_opt(opt, inner.clone()),
}
}
pub fn register(&self) {
let inner = self.inner.borrow();
let fd = inner.stream.as_raw_fd();
inner
.poll
.register(&EventedFd(&fd), SOCKET, Ready::readable(), PollOpt::level())
.unwrap();
}
pub fn dispatch(&self, events: &Events) -> io::Result<Option<Duration>> {
let inner = self.inner.clone();
for (_, token) in inner.borrow().tokens.iter() {
if let MyToken::Fd(fd) = token {
let ev = EventedFd(&fd.fd);
inner.borrow().poll.deregister(&ev)?;
}
}
for event in events {
match event.token() {
SOCKET => {
const NET_BUFSIZE: usize = 4096 + 65536; // defined by Emu
let mut buffer = [0; NET_BUFSIZE];
let fd = self.inner.borrow_mut().stream.as_raw_fd();
let mut f = unsafe { File::from_raw_fd(fd) };
let len = f.read(&mut buffer[..]).unwrap();
f.into_raw_fd();
self.ctxt.input(&buffer[..len]);
}
i if i.0 < inner.borrow().tokens.capacity() => {
let events = from_mio_ready(event.readiness());
let mut inner = inner.borrow_mut();
let token = &mut inner.tokens[i.0];
match token {
MyToken::Fd(fd) => {
// libslirp doesn't like getting more events...
fd.revents = Some(events & fd.events);
}
MyToken::Timer(MyTimer { func, .. }) => {
let func = func.clone();
drop(inner);
let func = &mut **func.borrow_mut();
func();
}
}
}
_ => continue,
}
}
self.ctxt.pollfds_poll(false, |idx| {
let token = &mut inner.borrow_mut().tokens[idx as usize];
if let MyToken::Fd(fd) = token {
fd.revents.take().unwrap_or(PollEvents::empty())
} else {
panic!();
}
});
inner
.borrow_mut()
.tokens
.retain(|_, v| if let MyToken::Fd(_) = v { false } else { true });
let mut timeout = u32::MAX;
self.ctxt.pollfds_fill(&mut timeout, |fd, events| {
let ready = to_mio_ready(events);
let tok = inner
.borrow_mut()
.tokens
.insert(MyToken::Fd(MyFd::new(fd, events)));
let ev = EventedFd(&fd);
inner
.borrow()
.poll
.register(&ev, Token(tok), ready, PollOpt::level())
.unwrap();
tok as i32
});
let duration = if timeout == u32::MAX {
None
} else {
Some(Duration::from_millis(timeout as u64))
};
Ok(duration)
}
}

88
third_party/libslirp-rs/src/opt.rs vendored
View file

@ -1,88 +0,0 @@
use ipnetwork::{Ipv4Network, Ipv6Network};
use std::net::{Ipv4Addr, Ipv6Addr};
use std::path::PathBuf;
use structopt::StructOpt;
#[derive(Debug, StructOpt)]
pub struct OptIpv4 {
/// Whether to disable IPv4
#[structopt(name = "disable-ipv4", long = "disable-ipv4")]
pub disable: bool,
/// IPv4 network CIDR
#[structopt(name = "net", long = "net", default_value = "10.0.2.0/24")]
pub net: Ipv4Network,
/// Guest-visible address of the host
#[structopt(long, default_value = "10.0.2.2")]
pub host: Ipv4Addr,
/// The first of the 16 IPs the built-in DHCP server can assign
#[structopt(
name = "dhcp-start",
long = "dhcp-start",
short,
default_value = "10.0.2.15"
)]
pub dhcp_start: Ipv4Addr,
/// Guest-visible address of the virtual nameserver
#[structopt(long = "dhcp-dns", default_value = "10.0.2.3")]
pub dns: Ipv4Addr,
}
#[derive(Debug, StructOpt)]
pub struct OptIpv6 {
/// Whether to disable IPv6
#[structopt(name = "disable-ipv6", long = "disable-ipv6")]
pub disable: bool,
/// IPv6 network CIDR
#[structopt(name = "net6", long = "net6", default_value = "fec0::/64")]
pub net6: Ipv6Network,
/// Guest-visible IPv6 address of the host
#[structopt(name = "host-ipv6", long = "host-ipv6", default_value = "fec0::2")]
pub host: Ipv6Addr,
/// Guest-visible address of the virtual nameserver
#[structopt(name = "dns-ipv6", long, default_value = "fec0::3")]
pub dns: Ipv6Addr,
}
#[derive(Debug, StructOpt)]
pub struct OptTftp {
/// RFC2132 "TFTP server name" string
#[structopt(name = "name", long = "tftp-name")]
pub name: Option<String>,
/// root directory of the built-in TFTP server
#[structopt(name = "root-path", parse(from_os_str), long = "tftp")]
pub root: Option<PathBuf>,
/// BOOTP filename, for use with tftp
#[structopt(long = "dhcp-bootfile")]
pub bootfile: Option<String>,
}
#[derive(Debug, StructOpt)]
#[structopt(name = "slirp-opt")]
pub struct Opt {
/// Isolate guest from host
#[structopt(long, short)]
pub restrict: bool,
/// Set interface MTU
#[structopt(long, default_value = "1500")]
pub mtu: usize,
/// Prohibit connection to 127.0.0.1
#[structopt(long)]
pub disable_host_loopback: bool,
/// Client hostname reported by the builtin DHCP server
#[structopt(long)]
pub hostname: Option<String>,
/// List of DNS suffixes to search, passed as DHCP option to the guest
#[structopt(long = "dns-suffixes")]
pub dns_suffixes: Vec<String>,
/// Guest-visible domain name of the virtual nameserver from DHCP server
#[structopt(long)]
pub domainname: Option<String>,
#[structopt(flatten)]
pub ipv4: OptIpv4,
#[structopt(flatten)]
pub ipv6: OptIpv6,
#[structopt(flatten)]
pub tftp: OptTftp,
}

12
third_party/libslirp-rs/src/version.rs vendored
View file

@ -1,12 +0,0 @@
use libslirp_sys::*;
use std::ffi::CStr;
use std::str;
pub fn version() -> &'static str {
str::from_utf8(unsafe { CStr::from_ptr(slirp_version_string()) }.to_bytes()).unwrap_or("")
}
pub fn state_version() -> i32 {
unsafe { slirp_state_version() }
}

0
third_party/libslirp-rs/tests/__init__.py vendored
View file

298
third_party/libslirp-rs/tests/base.py vendored
View file

@ -1,298 +0,0 @@
import ctypes
import functools
import io
import json
import os
import pathlib
import shlex
import signal
import socket
import subprocess
import tempfile
import unittest
from scapy.all import StreamSocket, sndrcv, Ether, conf, Route, ARP
SLIRPHELPER = os.environ.get("SLIRPHELPER")
LIBC = ctypes.CDLL("libc.so.6")
CLONE_NEWNET = 0x40000000
ORIGINAL_NET_NS = open("/proc/self/ns/net", "rb")
THISDIR = pathlib.Path(__file__).parent.absolute()
DBUS_SESSION_BUS_ADDRESS = os.environ.get("DBUS_SESSION_BUS_ADDRESS")
@functools.lru_cache()
def helper_capabilities():
p = subprocess.run(
[SLIRPHELPER, "--print-capabilities"], stdout=subprocess.PIPE, text=True
)
return json.loads(p.stdout)
def has_cap(cap):
return cap in helper_capabilities()["features"]
class Process:
def __init__(self, argv, close_fds=True, env=None):
self.p = subprocess.Popen(
argv,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
close_fds=close_fds,
env=env,
)
self.rc = None
def stdout_all(self):
return self.p.stdout.read()
def stdout_line(self):
return self.p.stdout.readline()
def stderr_all(self):
return self.p.stderr.read()
def stderr_line(self):
return self.p.stderr.readline()
def close(self, kill=True):
"""Returns process return code."""
if self.p:
if kill:
# Ensure the process registers two signals by sending a combo of
# SIGINT and SIGTERM. Sending the same signal two times is racy
# because the process can't reliably detect how many times the
# signal was sent.
self.p.send_signal(signal.SIGINT)
self.p.send_signal(signal.SIGTERM)
self.rc = self.p.wait()
self.p.stderr.close()
self.p.stdout.close()
self.p = None
return self.rc
def graceful_stop(self, wait=True):
self.p.send_signal(signal.SIGINT)
if wait:
self.p.wait()
class TestCase(unittest.TestCase):
has_notify_socket = None
execno = 0
def setUp(self):
if self.has_notify_socket is None:
self.has_notify_socket = has_cap("notify-socket")
self.cleanups = None
prev_net_fd = open("/proc/self/ns/net", "rb")
r = LIBC.unshare(CLONE_NEWNET)
if r != 0:
self.fail('Are you running within "unshare -Ur" ? Need unshare() syscall.')
self.guest_net_fd = open("/proc/self/ns/net", "rb")
self._add_teardown(self.guest_net_fd)
# mode tap, means ethernet headers
os.system(
"ip link set lo up;"
"ip tuntap add mode tap name tun0;"
"ip link set tun0 mtu 65521;"
"ip link set tun0 up;"
"ip addr add 10.0.2.100/24 dev tun0;"
"ip addr add 2001:2::100/32 dev tun0 nodad;"
"ip route add 0.0.0.0/0 via 10.0.2.2 dev tun0;"
"ip route add ::/0 via 2001:2::2 dev tun0;"
)
w = subprocess.Popen(["/bin/sleep", "1073741824"])
self.guest_ns_pid = w.pid
self._add_teardown(w)
LIBC.setns(prev_net_fd.fileno(), CLONE_NEWNET)
prev_net_fd.close()
self._tmpdir = tempfile.TemporaryDirectory()
self._add_teardown(self._tmpdir)
def tearDown(self):
while self.cleanups:
item = self.cleanups.pop()
if isinstance(item, subprocess.Popen):
item.send_signal(signal.SIGINT)
item.wait()
elif isinstance(item, Process):
item.close()
if getattr(item, "stdout", None):
item.stdout.close()
if getattr(item, "stderr", None):
item.stderr.close()
elif isinstance(item, io.BufferedReader):
item.close()
elif isinstance(item, tempfile.TemporaryDirectory):
item.cleanup()
else:
print("Unknown cleanup type")
print(type(item))
def run_helper(self, argv1=[], wait_ready=True, netns=True):
if isinstance(argv1, str):
argv1 = shlex.split(argv1)
a = [SLIRPHELPER] + argv1
if netns:
a = a + ["--netns", self.net_ns_path(), "--interface", "tun0"]
sn = None
env = None
if self.has_notify_socket and wait_ready:
self.execno += 1
sn = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
path = self.get_tmp_filename("sn-%d" % self.execno)
sn.bind(path)
env = dict(os.environ, NOTIFY_SOCKET="%s" % path)
p = Process(a, close_fds=False, env=env)
if sn:
sn.settimeout(1) # FIXME: remove timeout, end if process exit
try:
self.assertIn("READY=1", sn.recv(4096).decode())
except:
print(p.stderr_all())
sn.close()
self._add_teardown(p)
return p
def skipIfNotCapable(self, cap):
if not has_cap(cap):
self.skipTest("since '%s' capability is missing" % cap)
def start_echo(self, udp=False):
cmd = [THISDIR / "echo.py"]
if udp:
cmd += ["-u"]
p = Process(cmd)
self._add_teardown(p)
return int(p.stdout_line())
def assertTcpEcho(self, ip, port):
data = os.getrandom(16)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((ip, port))
s.sendall(data)
self.assertEqual(s.recv(len(data)), data)
def assertUdpEcho(self, ip, port):
data = os.getrandom(16)
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.sendto(data, (ip, port))
self.assertEqual(s.recv(len(data)), data)
def get_tmp_filename(self, name):
return os.path.join(self._tmpdir.name, name)
def _add_teardown(self, item):
if not self.cleanups:
self.cleanups = []
self.cleanups.append(item)
def net_ns_path(self):
return "/proc/%s/ns/net" % self.guest_ns_pid
def guest_netns(self):
xself = self
class controlled_execution:
def __enter__(self):
self.prev_net_fd = open("/proc/self/ns/net", "rb")
LIBC.setns(xself.guest_net_fd.fileno(), CLONE_NEWNET)
def __exit__(self, type, value, traceback):
LIBC.setns(self.prev_net_fd.fileno(), CLONE_NEWNET)
self.prev_net_fd.close()
return controlled_execution()
class testScapySocket:
def __init__(self, fd):
ss = StreamSocket(fd)
ss.basecls = Ether
self.ss = ss
conf.route = Route() # reinitializes the route based on the NS
self.e = Ether(src="52:55:0a:00:02:42")
def send(self, x):
self.ss.send(self.e / x)
def recv(self, x):
# this is not symmetrical with send, which appends Ether
# header, but ss.basecls will strip it of: not sure if that's
# the best way of doing things in fact, but that seem to work..
return self.ss.recv(x)
def fileno(self):
return self.ss.fileno()
def sr1(self, x, checkIPaddr=True, *args, **kwargs):
conf.checkIPaddr = checkIPaddr
kwargs.setdefault("verbose", False)
ans, _ = sndrcv(self.ss, self.e / x, *args, **kwargs)
return ans[0][1]
def sr(self, x, checkIPaddr=True, *args, **kwargs):
conf.checkIPaddr = checkIPaddr
kwargs.setdefault("verbose", False)
return sndrcv(self.ss, self.e / x, *args, **kwargs)
def withScapy():
def decorate(fn):
@functools.wraps(fn)
def maybe(*args, **kw):
sp = socket.socketpair(type=socket.SOCK_DGRAM)
os.set_inheritable(sp[0].fileno(), True)
self = args[0]
arg = kw.pop("parg", "")
p = self.run_helper(arg + " --fd %d" % sp[0].fileno(), netns=False)
s = testScapySocket(sp[1])
# gratious advertizing ARP
s.send(ARP(psrc="10.0.2.100", pdst="10.0.2.100", hwsrc=s.e.src))
kw["s"] = s
ret = fn(*args, **kw)
sp[0].close()
sp[1].close()
return ret
return maybe
return decorate
def isolateHostNetwork():
def decorate(fn):
@functools.wraps(fn)
def maybe(*args, **kw):
prev_net_fd = open("/proc/self/ns/net", "rb")
r = LIBC.unshare(CLONE_NEWNET)
if r != 0:
self.fail(
'Are you running within "unshare -Ur" ? Need unshare() syscall.'
)
# mode tun, since we don't actually plan on anyone reading the other side.
os.system(
"ip link set lo up;"
"ip tuntap add mode tun name eth0;"
"ip link set eth0 mtu 65521;"
"ip link set eth0 up;"
"ip addr add 192.168.1.100/24 dev eth0;"
"ip addr add 3ffe::100/16 dev eth0 nodad;"
"ip route add 0.0.0.0/0 via 192.168.1.1 dev eth0;"
"ip route add ::/0 via 3ffe::1 dev eth0;"
)
ret = fn(*args, **kw)
LIBC.setns(prev_net_fd.fileno(), CLONE_NEWNET)
prev_net_fd.close()
return ret
return maybe
return decorate

22
third_party/libslirp-rs/tests/dbus.conf vendored
View file

@ -1,22 +0,0 @@
<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
<busconfig>
<type>slirpnetstack-test</type>
<!-- Workaround abstract namespace and NEWNS issue -->
<listen>unix:dir=/tmp</listen>
<auth>EXTERNAL</auth>
<policy context='default'>
<!-- Allow everything to be sent -->
<allow send_destination='*' eavesdrop='true'/>
<!-- Allow everything to be received -->
<allow eavesdrop='true'/>
<!-- Allow anyone to own anything -->
<allow own='*'/>
</policy>
<include if_selinux_enabled='yes' selinux_root_relative='yes'>contexts/dbus_contexts</include>
</busconfig>

35
third_party/libslirp-rs/tests/echo.py vendored
View file

@ -1,35 +0,0 @@
#!/usr/bin/env python3
import socket
import sys
import getopt
def main(argv):
stype = socket.SOCK_STREAM
opts, args = getopt.getopt(argv, "u")
for opt, arg in opts:
if opt == "-u":
stype = socket.SOCK_DGRAM
s = socket.socket(socket.AF_INET, stype)
s.bind(("", 0))
print(s.getsockname()[1], flush=True)
if stype == socket.SOCK_STREAM:
s.listen(1)
s, _ = s.accept()
while 1:
data, addr = s.recvfrom(1024)
if not data:
break
if addr:
s.sendto(data, addr)
else:
s.sendall(data)
s.close()
if __name__ == "__main__":
main(sys.argv[1:])

119
third_party/libslirp-rs/tests/test-ip.rs vendored
View file

@ -1,119 +0,0 @@
use etherparse::{PacketBuilder, TcpOptionElement};
use libslirp;
use std::io;
use std::os::unix::io::RawFd;
use std::time::Instant;
use structopt::StructOpt;
impl libslirp::Handler for App {
type Timer = usize;
fn clock_get_ns(&mut self) -> i64 {
const NANOS_PER_SEC: u64 = 1_000_000_000;
let d = self.start.elapsed();
(d.as_secs() * NANOS_PER_SEC + d.subsec_nanos() as u64) as i64
}
fn timer_new(&mut self, _func: Box<dyn FnMut()>) -> Box<Self::Timer> {
Box::new(0)
}
fn timer_mod(&mut self, _timer: &mut Box<Self::Timer>, _expire_time: i64) {}
fn timer_free(&mut self, timer: Box<Self::Timer>) {
drop(timer);
}
fn send_packet(&mut self, buf: &[u8]) -> io::Result<usize> {
//self.stream.send(buf).unwrap() as isize
Ok(buf.len())
}
fn guest_error(&mut self, msg: &str) {
eprintln!("guest error: {}", msg);
}
fn register_poll_fd(&mut self, fd: RawFd) {
println!("register_poll_fd: fd={:?}", fd);
}
fn unregister_poll_fd(&mut self, fd: RawFd) {
println!("unregister_poll_fd: fd={:?}", fd);
}
fn notify(&mut self) {
println!("notify");
}
}
struct App {
start: Instant,
}
#[test]
fn ip() {
let opt = libslirp::Opt::from_args();
let app = App {
start: Instant::now(),
};
let ctxt = libslirp::Context::new_with_opt(&opt, app);
{
let builder = PacketBuilder::ethernet2(
[1, 2, 3, 4, 5, 6], //source mac
[7, 8, 9, 10, 11, 12], //destination mac
)
.ipv4(
[192, 168, 1, 1], //source ip
[192, 168, 1, 2], //desitination ip
20, //time to life
)
.udp(
21, //source port
1234, //desitnation port
);
//payload of the udp packet
let payload = [1, 2, 3, 4, 5, 6, 7, 8];
let mut buffer = Vec::<u8>::with_capacity(builder.size(payload.len()));
builder.write(&mut buffer, &payload).unwrap();
ctxt.input(&buffer);
}
{
let builder = PacketBuilder::ethernet2(
[1, 2, 3, 4, 5, 6], //source mac
[7, 8, 9, 10, 11, 12], //destionation mac
)
.ipv4(
[192, 168, 1, 1], //source ip
[192, 168, 1, 2], //desitionation ip
20, //time to life
)
.tcp(
21, //source port
1234, //desitnation port
1, //sequence number
26180, //window size
)
//set additional tcp header fields
.ns() //set the ns flag
//supported flags: ns(), fin(), syn(), rst(), psh(), ece(), cwr()
.ack(123) //ack flag + the ack number
.urg(23) //urg flag + urgent pointer
//tcp header options
.options(&[
TcpOptionElement::Nop,
TcpOptionElement::MaximumSegmentSize(1234),
])
.unwrap();
//payload of the tcp packet
let payload = [1, 2, 3, 4, 5, 6, 7, 8];
//get some memory to store the result
let mut buffer = Vec::<u8>::with_capacity(builder.size(payload.len()));
builder.write(&mut buffer, &payload).unwrap();
ctxt.input(&buffer);
}
}

246
third_party/libslirp-rs/tests/test.py vendored
View file

@ -1,246 +0,0 @@
import json
import unittest
from . import base
from scapy.all import *
from ipaddress import IPv4Address
from pydbus import SessionBus
class CLITest(base.TestCase):
def test_help(self):
""" Test if -h prints stuff looking like help screen. """
p = self.run_helper("-h", netns=False, wait_ready=False)
e = p.stderr_all()
self.assertFalse(e)
o = p.stdout_all().lower()
self.assertIn("usage:", o)
def test_print_capabilities(self):
""" Test if --print-capabilities output valid json. """
p = self.run_helper("--print-capabilities", netns=False, wait_ready=False)
e = p.stderr_all()
self.assertFalse(e)
o = p.stdout_all()
j = json.loads(o)
self.assertEqual(j["type"], "slirp-helper")
if "features" in j:
self.assertIsInstance(j["features"], list)
f = set(j["features"])
unknown = f.difference(
{
"dbus-address",
"dhcp",
"exit-with-parent",
"ipv4",
"ipv6",
"migrate",
"netns",
"notify-socket",
"restrict",
"tftp",
}
)
for cap in unknown:
if not cap.startswith("x-"):
self.fail("Unknown capability: %s" % cap)
def test_restrict(self):
""" Basic test if 'restrict' options exists. """
self.skipIfNotCapable("restrict")
self.run_helper("--restrict")
def test_ipv4(self):
""" Basic test if 'ipv4' options exists. """
self.skipIfNotCapable("ipv4")
self.run_helper("--disable-ipv4 --net 12.12.0.1/8")
def test_ipv4(self):
""" Basic test if 'ipv6' options exists. """
self.skipIfNotCapable("ipv6")
self.run_helper("--disable-ipv6 --net6 fec0::/64")
def test_exit_with_parent(self):
""" Basic test if 'exit-with-parent' option exists. """
self.skipIfNotCapable("exit-with-parent")
self.run_helper("--exit-with-parent")
def test_tftp(self):
""" Basic test if 'tftp' options exists. """
self.skipIfNotCapable("tftp")
self.run_helper("--tftp .")
def test_net(self):
""" Basic test if --net parses successfully. """
p = self.run_helper("--net 12.12.0.1/23")
p.graceful_stop()
p = self.run_helper("--net wefo/23", wait_ready=False)
e = p.stderr_all()
self.assertTrue(e)
p.graceful_stop()
def test_dbus(self):
""" Test if --dbus-address works. """
self.skipIfNotCapable("dbus-address")
if not base.DBUS_SESSION_BUS_ADDRESS:
self.skipTest("DBUS_SESSION_BUS_ADDRESS unset")
p = self.run_helper(
"--dbus-id TestId --dbus-address %s" % base.DBUS_SESSION_BUS_ADDRESS
)
bus = SessionBus()
iface = bus.get(".Slirp1_%u" % p.p.pid, "/org/freedesktop/Slirp1/Helper")
info = iface.GetInfo()
self.assertIn("Protocol[State]", info)
class ConnTest(base.TestCase):
@base.withScapy()
def test_ping(self, s):
""" Test Scapy ping """
pkt = s.sr1(IP(dst="10.0.2.2") / ICMP())
self.assertEqual(pkt.sprintf("%ICMP.type%"), "echo-reply")
@base.isolateHostNetwork()
def test_restrict(self):
""" Test --restrict behaviour """
port = self.start_echo()
self.run_helper("--restrict")
with self.guest_netns():
with self.assertRaises((ConnectionError, ConnectionRefusedError)):
self.assertTcpEcho("192.168.1.100", port)
@base.isolateHostNetwork()
def test_tcp_echo(self):
""" Test TCP echo """
port = self.start_echo()
self.run_helper()
with self.guest_netns():
self.assertTcpEcho("192.168.1.100", port)
@base.isolateHostNetwork()
def test_udp_echo(self):
""" Test UDP echo """
port = self.start_echo(udp=True)
self.run_helper()
with self.guest_netns():
self.assertUdpEcho("192.168.1.100", port)
@unittest.skipUnless(base.has_cap("dhcp"), "Missing 'dhcp' feature")
class DHCPTest(base.TestCase):
@base.withScapy()
def test_dhcp_v4(self, s):
""" Test DHCPv4 discover """
bootp = BOOTP(xid=RandInt())
dhcp = DHCP(options=[("message-type", "discover"), "end"])
p = (
IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ bootp
/ dhcp
)
pkt = s.sr1(p, checkIPaddr=False)
self.assertEqual(pkt.sprintf("%BOOTP.op%"), "BOOTREPLY")
addr = IPv4Address(pkt[BOOTP].yiaddr)
self.assertGreaterEqual(addr, IPv4Address("10.0.2.15"))
self.assertLess(addr, IPv4Address("10.0.2.100"))
for o in pkt[DHCP].options:
if o[0] in ("router", "server_id"):
self.assertEqual(o[1], "10.0.2.2")
opts = [o[0] for o in pkt[DHCP].options if isinstance(o, tuple)]
self.assertIn("router", opts)
self.assertIn("name_server", opts)
self.assertIn("lease_time", opts)
self.assertIn("server_id", opts)
@base.withScapy()
def dhcp_and_net(self, s):
bootp = BOOTP(xid=RandInt())
dhcp = DHCP(options=[("message-type", "discover"), "end"])
p = (
IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ bootp
/ dhcp
)
pkt = s.sr1(p, checkIPaddr=False)
self.assertEqual(pkt.sprintf("%BOOTP.op%"), "BOOTREPLY")
addr = IPv4Address(pkt[BOOTP].yiaddr)
self.assertGreaterEqual(addr, IPv4Address("12.34.56.15"))
self.assertLess(addr, IPv4Address("12.34.56.100"))
def test_dhcp_and_net(self):
""" Test DHCPv4 and -net """
self.dhcp_and_net(parg="--net 12.34.56.1/24")
@base.withScapy()
def dhcp_dns(self, s):
bootp = BOOTP(xid=RandInt())
dhcp = DHCP(options=[("message-type", "discover"), "end"])
p = (
IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ bootp
/ dhcp
)
pkt = s.sr1(p, checkIPaddr=False)
# BOOTREPLY
for o in pkt[DHCP].options:
if o[0] == "name_server":
self.assertEqual(o[1], "8.8.8.8")
return
self.fail()
def test_dhcp_dns(self):
""" Test DHCPv4 DNS option """
self.dhcp_dns(parg="--dhcp-dns 8.8.8.8")
@base.withScapy()
def dhcp_nbp(self, s):
bootp = BOOTP(xid=RandInt())
dhcp = DHCP(options=[("message-type", "discover"), "end"])
p = (
IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ bootp
/ dhcp
)
pkt = s.sr1(p, checkIPaddr=False)
# BOOTREPLY
bootFileName = pkt[BOOTP].file.partition(b"\0")[0].decode()
tftpServerName = None
for o in pkt[DHCP].options:
if o[0] == "boot-file-name":
bootFileName = o[1].decode() # Higher precedence?
elif o[0] in (
66,
"tftp-server-name",
"tftp_server_name",
): # FIXME: scapy doesn't know that field?
tftpServerName = o[1].decode()
self.assertEqual(tftpServerName, "10.0.0.1")
self.assertEqual(bootFileName, "/my-nbp")
def test_dhcp_nbp(self):
""" Test DHCPv4 NBP option """
self.dhcp_nbp(parg="--dhcp-nbp tftp://10.0.0.1/my-nbp")
@base.withScapy()
def dhcp_bootfile(self, s):
bootp = BOOTP(xid=RandInt())
dhcp = DHCP(options=[("message-type", "discover"), "end"])
p = (
IP(src="0.0.0.0", dst="255.255.255.255")
/ UDP(sport=68, dport=67)
/ bootp
/ dhcp
)
pkt = s.sr1(p, checkIPaddr=False)
# BOOTREPLY
self.assertEqual(
pkt[BOOTP].file.partition(b"\0")[0].decode(), "http://boot.netboot.xyz/"
)
def test_dhcp_bootfile(self):
""" Test DHCPv4 bootfile option """
self.dhcp_bootfile(parg="--dhcp-bootfile http://boot.netboot.xyz/")