diff --git a/CHANGELOG.md b/CHANGELOG.md index f9753872..3dfb63e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ ## Unreleased: mitmproxy_rs next +- Add IPv6 support to Linux local capture mode. ## 17 July 2026: mitmproxy_rs 0.12.10 diff --git a/Cargo.lock b/Cargo.lock index fc77f9f2..efeb64b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2236,7 +2236,9 @@ dependencies = [ "mitmproxy", "mitmproxy-linux-ebpf", "mitmproxy-linux-ebpf-common", + "netlink-packet-route", "prost 0.14.4", + "rtnetlink", "tokio", "tun", ] @@ -2307,6 +2309,54 @@ dependencies = [ "pxfm", ] +[[package]] +name = "netlink-packet-core" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3463cbb78394cb0141e2c926b93fc2197e473394b761986eca3b9da2c63ae0f4" +dependencies = [ + "paste", +] + +[[package]] +name = "netlink-packet-route" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be8919612f6028ab4eacbbfe1234a9a43e3722c6e0915e7ff519066991905092" +dependencies = [ + "bitflags 2.13.1", + "libc", + "log", + "netlink-packet-core", +] + +[[package]] +name = "netlink-proto" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b65d130ee111430e47eed7896ea43ca693c387f097dd97376bffafbf25812128" +dependencies = [ + "bytes", + "futures", + "log", + "netlink-packet-core", + "netlink-sys", + "thiserror 2.0.18", +] + +[[package]] +name = "netlink-sys" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd6c30ed10fa69cc491d491b85cc971f6bdeb8e7367b7cde2ee6cc878d583fae" +dependencies = [ + "bytes", + "futures-util", + "libc", + "log", + "tokio", +] + [[package]] name = "nix" version = "0.25.1" @@ -2637,6 +2687,12 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -3198,6 +3254,24 @@ dependencies = [ "serde", ] +[[package]] +name = "rtnetlink" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc19f84f710fa2f337617f9bc0400260a94224bde7bae28fd8879f3771ca5784" +dependencies = [ + "futures-channel", + "futures-util", + "log", + "netlink-packet-core", + "netlink-packet-route", + "netlink-proto", + "netlink-sys", + "nix 0.30.1", + "thiserror 1.0.69", + "tokio", +] + [[package]] name = "rustc_version" version = "0.4.1" diff --git a/mitmproxy-linux/Cargo.toml b/mitmproxy-linux/Cargo.toml index d3db739a..553e2ac6 100644 --- a/mitmproxy-linux/Cargo.toml +++ b/mitmproxy-linux/Cargo.toml @@ -29,6 +29,8 @@ prost = "0.14.3" internet-packet = { version = "0.2.0", features = ["checksums"] } libc = "0.2.186" const-sha1 = "0.3.0" +rtnetlink = "0.21.0" +netlink-packet-route = "0.30.0" [target.'cfg(target_os = "linux")'.build-dependencies] anyhow = { version = "1.0.102", features = ["backtrace"] } diff --git a/mitmproxy-linux/src/main2.rs b/mitmproxy-linux/src/main2.rs index dcf38a1a..b3ea60ef 100644 --- a/mitmproxy-linux/src/main2.rs +++ b/mitmproxy-linux/src/main2.rs @@ -1,27 +1,30 @@ -use std::{fs, iter}; -use std::fs::Permissions; use anyhow::Context; -use anyhow::anyhow; use anyhow::Result; -use aya::{Ebpf, EbpfLoader}; +use anyhow::anyhow; +use aya::Btf; use aya::maps::Array; +use aya::programs::{CgroupSock, links::CgroupAttachMode}; +use aya::{Ebpf, EbpfLoader}; +use log::{debug, error, info, warn}; +use mitmproxy::ipc::FromProxy; +use mitmproxy::ipc::{PacketWithMeta, from_proxy}; +use mitmproxy::packet_sources::IPC_BUF_SIZE; +use mitmproxy::packet_sources::tun::create_tun_device; +use mitmproxy_linux_ebpf_common::{Action, INTERCEPT_CONF_LEN}; +use netlink_packet_route::rule::{RuleAction, RuleAttribute, RuleMessage}; +use prost::Message; +use prost::bytes::{Bytes, BytesMut}; +use std::fs::Permissions; +use std::net::Ipv6Addr; use std::os::unix::fs::PermissionsExt; use std::path::PathBuf; -use aya::Btf; -use aya::programs::{links::CgroupAttachMode, CgroupSock}; -use log::{debug, warn, info, error}; -use prost::bytes::{Bytes, BytesMut}; +use std::{fs, iter}; +use tokio::io::AsyncReadExt; use tokio::net::UnixDatagram; use tokio::select; -use mitmproxy::packet_sources::tun::create_tun_device; +use tokio::signal::unix::{SignalKind, signal}; +use tokio::task::JoinHandle; use tun::AbstractDevice; -use prost::Message; -use tokio::io::AsyncReadExt; -use tokio::signal::unix::{signal, SignalKind}; -use mitmproxy::ipc::{PacketWithMeta, from_proxy}; -use mitmproxy::ipc::FromProxy; -use mitmproxy::packet_sources::IPC_BUF_SIZE; -use mitmproxy_linux_ebpf_common::{Action, INTERCEPT_CONF_LEN}; // We can't implement aya::Pod in mitmproxy-linux-ebpf-common, so we do it on a newtype. // (see https://github.com/aya-rs/aya/pull/59) @@ -35,7 +38,10 @@ const BPF_PROG: &[u8] = aya::include_bytes_aligned!(concat!(env!("OUT_DIR"), "/m const BPF_HASH: [u8; 20] = const_sha1::sha1(BPF_PROG).as_bytes(); fn load_bpf(device_index: u32) -> Result { - debug!("Loading BPF program ({:x})...", Bytes::from_static(&BPF_HASH)); + debug!( + "Loading BPF program ({:x})...", + Bytes::from_static(&BPF_HASH) + ); let mut ebpf = EbpfLoader::new() .btf(Btf::from_sys_fs().ok().as_ref()) .set_global("INTERFACE_ID", &device_index, true) @@ -47,19 +53,22 @@ fn load_bpf(device_index: u32) -> Result { } debug!("Attaching BPF_CGROUP_INET_SOCK_CREATE program..."); - let prog: &mut CgroupSock = ebpf.program_mut("cgroup_sock_create").context("failed to get cgroup_sock_create")?.try_into()?; + let prog: &mut CgroupSock = ebpf + .program_mut("cgroup_sock_create") + .context("failed to get cgroup_sock_create")? + .try_into()?; // root cgroup to get all events. let cgroup = fs::File::open("/sys/fs/cgroup/").context("failed to open \"/sys/fs/cgroup/\"")?; - prog.load().context("failed to load cgroup_sock_create program")?; - prog.attach(&cgroup, CgroupAttachMode::Single).context("failed to attach cgroup_sock_create program")?; + prog.load() + .context("failed to load cgroup_sock_create program")?; + prog.attach(&cgroup, CgroupAttachMode::Single) + .context("failed to attach cgroup_sock_create program")?; Ok(ebpf) } #[tokio::main] async fn main() -> anyhow::Result<()> { - env_logger::Builder::from_env( - env_logger::Env::default().default_filter_or("info") - ) + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")) //.format_target(false) .format_timestamp(None) .init(); @@ -76,18 +85,19 @@ async fn main() -> anyhow::Result<()> { debug!("Creating tun device..."); let (mut device, name) = create_tun_device(None)?; - let device_index = device.tun_index().context("failed to get tun device index")? as u32; + let device_index = device + .tun_index() + .context("failed to get tun device index")? as u32; debug!("Tun device created: {name} (id={device_index})"); - let mut ebpf = load_bpf(device_index).context("eBPF initialization failed")?; - - debug!("Getting INTERCEPT_CONF map..."); - let mut intercept_conf = { - let map = ebpf.map_mut("INTERCEPT_CONF") - .context("couldn't get INTERCEPT_CONF map")?; - Array::<_, ActionWrapper>::try_from(map) - .context("Cannot cast INTERCEPT_CONF to Array")? - }; + // Set up IPv6 policy-based routing so that IPv6 traffic from eBPF-bound + // sockets is routed into the tun. IPv4 needs no such route. + let ipv6 = ipv6::Ipv6Routes::setup(name, device_index) + .await + .inspect_err(|e| { + warn!("Failed to set up IPv6 policy routing: {e:?}"); + }) + .ok(); debug!("Connecting to {}...", mitmproxy_addr.display()); let ipc = UnixDatagram::bind(&redirector_addr) @@ -98,78 +108,104 @@ async fn main() -> anyhow::Result<()> { fs::set_permissions(&redirector_addr, Permissions::from_mode(0o777))?; println!("{}", redirector_addr.to_string_lossy()); - // Exit cleanly on SIGINT/SIGTERM - tokio::spawn(async { - let mut sigint = signal(SignalKind::interrupt()).context("failed to register SIGINT listener").unwrap(); - let mut sigterm = signal(SignalKind::terminate()).context("failed to register SIGTERM listener").unwrap(); - select! { - _ = sigint.recv() => (), - _ = sigterm.recv() => (), - } - std::process::exit(0); - }); + let mut main_loop = tokio::spawn(async move { + let mut ipc_buf = Vec::with_capacity(IPC_BUF_SIZE); + let mut dev_buf = BytesMut::with_capacity(IPC_BUF_SIZE); - let mut ipc_buf = Vec::with_capacity(IPC_BUF_SIZE); - let mut dev_buf = BytesMut::with_capacity(IPC_BUF_SIZE); - - loop { - ipc_buf.clear(); - select! { - r = ipc.recv_buf(&mut ipc_buf) => { - match r { - Ok(len) if len > 0 => { - let Ok(FromProxy { message: Some(message)}) = FromProxy::decode(ipc_buf.as_slice()) else { - return Err(anyhow!("Received invalid IPC message: {:?}", &ipc_buf[..len])); - }; - // debug!("Received IPC message: {message:?}"); - - match message { - from_proxy::Message::Packet(packet) => { - // debug!("Forwarding Packet to device: {}", packet.data.len()); - device.send(&packet.data).await.context("failed to send packet")?; - } - from_proxy::Message::InterceptConf(conf) => { - debug!("Updating ebpf intercept conf: {conf:?}"); - if conf.actions.len() > INTERCEPT_CONF_LEN as usize { - error!("Truncating intercept conf to {INTERCEPT_CONF_LEN} elements."); + let mut ebpf = load_bpf(device_index).context("eBPF initialization failed")?; + debug!("Getting INTERCEPT_CONF map..."); + let mut intercept_conf = { + let map = ebpf + .map_mut("INTERCEPT_CONF") + .context("couldn't get INTERCEPT_CONF map")?; + Array::<_, ActionWrapper>::try_from(map) + .context("Cannot cast INTERCEPT_CONF to Array")? + }; + + loop { + ipc_buf.clear(); + select! { + r = ipc.recv_buf(&mut ipc_buf) => { + match r { + Ok(len) if len > 0 => { + let Ok(FromProxy { message: Some(message)}) = FromProxy::decode(ipc_buf.as_slice()) else { + return Err(anyhow!("Received invalid IPC message: {:?}", &ipc_buf[..len])); + }; + // debug!("Received IPC message: {message:?}"); + + match message { + from_proxy::Message::Packet(packet) => { + // debug!("Forwarding Packet to device: {}", packet.data.len()); + device.send(&packet.data).await.context("failed to send packet")?; } - let actions = conf.actions - .iter() - .map(|s| Action::from(s.as_str())) - .chain(iter::once(Action::None)) - .take(INTERCEPT_CONF_LEN as usize); - for (i, action) in actions.enumerate() { - intercept_conf.set(i as u32, ActionWrapper(action), 0) - .context("failed to update INTERCEPT_CONF")?; + from_proxy::Message::InterceptConf(conf) => { + debug!("Updating ebpf intercept conf: {conf:?}"); + if conf.actions.len() > INTERCEPT_CONF_LEN as usize { + error!("Truncating intercept conf to {INTERCEPT_CONF_LEN} elements."); + } + let actions = conf.actions + .iter() + .map(|s| Action::from(s.as_str())) + .chain(iter::once(Action::None)) + .take(INTERCEPT_CONF_LEN as usize); + for (i, action) in actions.enumerate() { + intercept_conf.set(i as u32, ActionWrapper(action), 0) + .context("failed to update INTERCEPT_CONF")?; + } } } } + _ => { + info!("IPC read failed. Exiting."); + return Ok::<(), anyhow::Error>(()); + } } - _ => { - info!("IPC read failed. Exiting."); - std::process::exit(0); - } - } - }, - // ... or process incoming packets - r = device.read_buf(&mut dev_buf) => { - r.context("TUN read() failed")?; - - let packet = PacketWithMeta { - data: dev_buf.split().freeze(), - tunnel_info: None, - }; - - packet.encode(&mut ipc_buf)?; - // debug!("Sending packet to proxy: {} {:?}", encoded.len(), &encoded); - ipc.send(ipc_buf.as_slice()).await?; - - // Reclaim space in dev_buf. - drop(packet); - assert!(dev_buf.try_reclaim(IPC_BUF_SIZE)); - }, + }, + // ... or process incoming packets + r = device.read_buf(&mut dev_buf) => { + r.context("TUN read() failed")?; + + let packet = PacketWithMeta { + data: dev_buf.split().freeze(), + tunnel_info: None, + }; + + packet.encode(&mut ipc_buf)?; + // debug!("Sending packet to proxy: {} {:?}", encoded.len(), &encoded); + ipc.send(ipc_buf.as_slice()).await?; + + // Reclaim space in dev_buf. + drop(packet); + assert!(dev_buf.try_reclaim(IPC_BUF_SIZE)); + }, + } + } + }); + + // Wait for a shutdown signal, or for the main loop to terminate on its own. + let mut sigint = + signal(SignalKind::interrupt()).context("failed to register SIGINT listener")?; + let mut sigterm = + signal(SignalKind::terminate()).context("failed to register SIGTERM listener")?; + let result = select! { + _ = sigint.recv() => { + info!("Received SIGINT, exiting."); + Ok(()) } + _ = sigterm.recv() => { + info!("Received SIGTERM, exiting."); + Ok(()) + } + r = &mut main_loop => r.unwrap_or_else(|e| Err(anyhow!("main loop task failed: {e}"))), + }; + + if let Some(ipv6) = ipv6 + && let Err(e) = ipv6.cleanup().await + { + debug!("Failed to clean up IPv6 policy routing rule: {e}"); } + + result } /// Bump the memlock rlimit. This is needed for older kernels that don't use the @@ -185,6 +221,87 @@ fn bump_memlock_rlimit() { } } +/// For IPv4, just assigning the packet to the right interface is enough for it to show up +/// on the TUN device. For IPv6, the kernel requires us to set up a plausible route first. +mod ipv6 { + use super::*; + + // Randomly chosen table ID and rule priority to avoid colliding with other tools + // on the host. https://xkcd.com/221/ + const TUN_IPV6_PBR_TABLE: u32 = 1783940182; + const TUN_IPV6_PBR_PRIORITY: u32 = 14285; + + pub struct Ipv6Routes { + tun_name: String, + handle: rtnetlink::Handle, + task: JoinHandle<()>, + } + + impl Ipv6Routes { + /// `ip -6 rule add oif lookup priority ` + /// `ip -6 route add default dev table
` + pub async fn setup(tun_name: String, tun_index: u32) -> Result { + let (conn, handle, _) = + rtnetlink::new_connection().context("failed to create rtnetlink connection")?; + let task = tokio::spawn(conn); + + // 1. Add the policy routing rule. + let mut rule = handle.rule().add(); + *rule.message_mut() = pbr_rule_message(tun_name.clone()); + rule.replace() + .execute() + .await + .context("failed to add IPv6 policy routing rule")?; + + // 2. Add a default route via the tun into our dedicated table. + let route = rtnetlink::RouteMessageBuilder::::new() + .table_id(TUN_IPV6_PBR_TABLE) + .output_interface(tun_index) + .build(); + handle + .route() + .add(route) + .replace() + .execute() + .await + .context("failed to add IPv6 default route in PBR table")?; + + Ok(Self { + handle, + task, + tun_name, + }) + } + + pub async fn cleanup(self) -> Result<()> { + self.handle + .rule() + .del(pbr_rule_message(self.tun_name)) + .execute() + .await + .context("failed to delete IPv6 policy routing rule")?; + // The route in the dedicated table is attached to the tun and automatically + // removed by the kernel when the device is destroyed. + self.task.abort(); + Ok(()) + } + } + + /// The policy routing rule shared by setup and cleanup. + fn pbr_rule_message(tun_name: String) -> RuleMessage { + let mut message = RuleMessage::default(); + message.header.family = netlink_packet_route::AddressFamily::Inet6; + message.header.action = RuleAction::ToTable; + message.attributes.push(RuleAttribute::Oifname(tun_name)); + message + .attributes + .push(RuleAttribute::Table(TUN_IPV6_PBR_TABLE)); + message + .attributes + .push(RuleAttribute::Priority(TUN_IPV6_PBR_PRIORITY)); + message + } +} #[cfg(test)] mod tests { @@ -195,5 +312,4 @@ mod tests { async fn bpf_load() { load_bpf(0).unwrap(); } - -} \ No newline at end of file +}