VXLAN multiplexing and congestion control

CHANGELOG.md

@@ -13,6 +13,9 @@ All Nullnet releases with the relative changes are documented in this file.
 - Per-service egress/ingress traffic filters: arbitrary AND/OR/group combinations of Country, Organization, Src IP (ingress), and Dst IP (egress) conditions, evaluated via `rpn-predicate-interpreter` — replaces the country-only egress/ingress policy ([#171](https://github.com/NullNet-ai/nullnet/pull/171) — fixes [#143](https://github.com/NullNet-ai/nullnet/issues/143))
 - Persist ingress and egress sessions to SQLite and show the full history on the Sessions page, filterable by status, service, direction, and policy verdict, with its own retention window ([#170](https://github.com/NullNet-ai/nullnet/pull/170) — fixes [#156](https://github.com/NullNet-ai/nullnet/issues/156))
 ### Changed
+- Show bounded live and recent ingress pressure, congestion-control state, and active ingress groups on the Nodes page ([#206](https://github.com/NullNet-ai/nullnet/pull/206)).
+- Adapt encrypted ingress multiplexing, with or without a configured hard cap, to sustained RPC admission congestion by pausing dedicated setups and conservatively probing verified recovery per host ([#206](https://github.com/NullNet-ai/nullnet/pull/206)).
+- Multiplex encrypted ingress overflow with client-IP policy isolation, persistent host-pair IPsec transports, and direct same-host forwarding ([#206](https://github.com/NullNet-ai/nullnet/pull/206)).
 - Keep routine successful proxy requests out of Events, isolate diagnostic event RPC admission from routing, and group event filters by scope ([#204](https://github.com/NullNet-ai/nullnet/pull/204)).
 - Reduce fresh-edge cleanup work, batch conntrack scans, and share TC classifier infrastructure while preserving dedicated encrypted endpoints ([#201](https://github.com/NullNet-ai/nullnet/pull/201)).
 - Keep active backend chains visible to conntrack so liveness tracking does not release endpoints that still carry traffic ([#201](https://github.com/NullNet-ai/nullnet/pull/201)).

SETUP.md

@@ -256,6 +256,16 @@ Run the setup scripts as root. The supplied systemd services also run as root; p
 - `timeout` controls proxy-reachability: when present the service is a proxy-reachable entry point
   with that per-client idle timeout in seconds (`0` disables the timeout); omit it to keep the
   service off the proxy (backend-only)
+- `max_networks` is an optional hard cap on dedicated ingress networks for a proxy-reachable
+  service. Independently of that setting, encrypted VXLAN ingress switches new clients to
+  multiplexing when the proxy's RPC backlog and slot-wait time grow for consecutive windows. A
+  host-scoped controller first pauses dedicated setups, then additively admits one more probe per
+  healthy window; renewed congestion immediately returns the rate to zero. Stable queues do not
+  activate it, and only sustained successful probes under traffic remove the limit (idle time does
+  not count as recovery). Multiplexed clients get one policy
+  generation per client IP over a shared VXLAN. Cross-host groups share a persistent host-pair IPsec
+  transport; packet metadata and endpoint maps preserve per-client isolation. Same-host multiplexing
+  uses the same isolation policy with direct eBPF redirection and no VXLAN hairpin.
 - `proxy_dependencies` is a list of independent dep chains walked when the service is reached via a
   `Proxy` RPC from nullnet-proxy; each inner array is one linear branch and all branches are brought
   up in parallel

ebpf/src/main.rs

@@ -3,6 +3,7 @@
 
 mod device_events;
 mod endpoint_tc;
+mod policy;
 
 use aya_ebpf::{
     bindings::{TC_ACT_OK, TC_ACT_SHOT},
@@ -48,6 +49,9 @@ use network_types::{
 
 const VXLAN_PORT: u16 = 47890;
 const FORWARD_PORT: u16 = 9999;
+const POLICY_PORT: u16 = 4791;
+const POLICY_MARK: u32 = 0x4e80_0000;
+const POLICY_MARK_MASK: u32 = 0xffc0_0000;
 
 const PROTO_TCP: u8 = 6;
 const PROTO_UDP: u8 = 17;
@@ -159,6 +163,16 @@ fn try_firewall(ctx: &TcContext, is_egress: bool) -> Result<i32, ()> {
                 _ => return Ok(TC_ACT_SHOT),
             };
 
+            let mark = unsafe { (*ctx.skb.skb).mark };
+            if !is_egress
+                && proto == PROTO_UDP
+                && dst_port == POLICY_PORT
+                && mark & POLICY_MARK_MASK == POLICY_MARK
+                && (is_peer(src) || is_peer(dst))
+            {
+                return Ok(TC_ACT_OK);
+            }
+
             let key = ct_key(src, dst, src_port, dst_port, proto);
             if ct_hit(&key) || base_allow(is_egress, proto, src, dst, src_port, dst_port) {
                 ct_insert(&key);

ebpf/src/policy.rs

@@ -0,0 +1,178 @@
+//! Permission-gated IPv4 forwarding over shared policy transports.
+use aya_ebpf::{
+    EbpfContext,
+    bindings::{TC_ACT_SHOT, bpf_tunnel_key},
+    helpers::{
+        bpf_ktime_get_ns, bpf_redirect, bpf_skb_get_tunnel_key, bpf_skb_get_tunnel_opt,
+        bpf_skb_set_tunnel_key, bpf_skb_set_tunnel_opt,
+    },
+    macros::{classifier, map},
+    maps::{Array, HashMap, LruHashMap},
+    programs::TcContext,
+};
+use core::mem::{size_of, zeroed};
+
+#[path = "../../shared/policy.rs"]
+#[allow(dead_code)]
+mod types;
+use types::{APP_IP, FlowKey, OutKey, Rule, XFRM_ROUTE_BIT};
+
+#[map]
+static POLICY_OUT: HashMap<OutKey, Rule> = HashMap::with_max_entries(65536, 0);
+#[map]
+static POLICY_IN: HashMap<u32, Rule> = HashMap::with_max_entries(65536, 0);
+#[map]
+static POLICY_DEADLINE: Array<u64> = Array::with_max_entries(1, 0);
+#[map]
+static POLICY_FLOWS: LruHashMap<FlowKey, u64> = LruHashMap::with_max_entries(262144, 0);
+
+fn alive() -> bool {
+    POLICY_DEADLINE
+        .get(0)
+        .is_some_and(|deadline| *deadline > unsafe { bpf_ktime_get_ns() })
+}
+
+fn header(ctx: &TcContext) -> Result<(usize, u8), i64> {
+    if u16::from_be(ctx.load::<u16>(12)?) != 0x0800 {
+        return Err(-1);
+    }
+    let version = ctx.load::<u8>(14)?;
+    let len = usize::from(version & 15) * 4;
+    if version >> 4 != 4 || len < 20 || u16::from_be(ctx.load::<u16>(20)?) & 0x3fff != 0 {
+        return Err(-1);
+    }
+    let proto = ctx.load::<u8>(23)?;
+    if proto != 6 && proto != 17 && proto != 1 {
+        return Err(-1);
+    }
+    Ok((len, proto))
+}
+
+fn translate(
+    ctx: &TcContext,
+    offset: usize,
+    target: u32,
+    len: usize,
+    proto: u8,
+) -> Result<(), i64> {
+    let old = ctx.load::<u32>(offset)?;
+    let new = target.to_be();
+    ctx.l3_csum_replace(24, old as u64, new as u64, 4)?;
+    let checksum = 14 + len + if proto == 6 { 16 } else { 6 };
+    if proto == 6 || (proto == 17 && ctx.load::<u16>(checksum)? != 0) {
+        ctx.l4_csum_replace(checksum, old as u64, new as u64, 4 | 16 | 32)?;
+    }
+    ctx.store(offset, &new, 0)
+}
+
+fn flow(
+    ctx: &TcContext,
+    rule: &Rule,
+    interface: u32,
+    len: usize,
+    proto: u8,
+    incoming: bool,
+) -> Result<(), i64> {
+    if proto != 6 {
+        return Ok(());
+    }
+    let source = ctx.load::<u16>(14 + len)?;
+    let dest = ctx.load::<u16>(16 + len)?;
+    let flags = ctx.load::<u8>(27 + len)?;
+    let key = FlowKey {
+        interface,
+        peer: rule.peer,
+        local_port: if incoming { dest } else { source },
+        remote_port: if incoming { source } else { dest },
+    };
+    if flags & 0x02 != 0 {
+        POLICY_FLOWS.insert(key, rule.generation, 0)?;
+    } else if unsafe { POLICY_FLOWS.get(key) }.copied() != Some(rule.generation) {
+        return Err(-1);
+    }
+    Ok(())
+}
+
+#[classifier]
+pub fn nullnet_policy_out(ctx: TcContext) -> i32 {
+    outgoing(&ctx).unwrap_or(TC_ACT_SHOT)
+}
+
+fn outgoing(ctx: &TcContext) -> Result<i32, i64> {
+    if !alive() {
+        return Err(-1);
+    }
+    let (len, proto) = header(ctx)?;
+    if u32::from_be(ctx.load::<u32>(26)?) != APP_IP {
+        return Err(-1);
+    }
+    let interface = unsafe { (*ctx.skb.skb).ifindex };
+    let key = OutKey {
+        interface,
+        peer: u32::from_be(ctx.load::<u32>(30)?),
+    };
+    let rule = unsafe { POLICY_OUT.get(key) }.copied().ok_or(-1)?;
+    if rule.enabled == 0 {
+        return Err(-1);
+    }
+    flow(ctx, &rule, interface, len, proto, false)?;
+    translate(ctx, 26, rule.local, len, proto)?;
+
+    if rule.underlay_local == rule.underlay_remote {
+        translate(ctx, 30, APP_IP, len, proto)?;
+        ctx.store(0, &rule.mac, 0)?;
+        return Ok(unsafe { bpf_redirect(rule.output, 0) } as i32);
+    }
+
+    let mut tunnel: bpf_tunnel_key = unsafe { zeroed() };
+    tunnel.tunnel_id = (rule.generation & 0xff_ffff) as u32;
+    tunnel.tunnel_ttl = 64;
+    tunnel.__bindgen_anon_1.remote_ipv4 = rule.underlay_remote;
+    tunnel.__bindgen_anon_3.local_ipv4 = rule.underlay_local;
+    let mut option = (rule.generation >> 24) as u32 | (1 << 22);
+    let skb = ctx.as_ptr().cast();
+    if unsafe { bpf_skb_set_tunnel_key(skb, &mut tunnel, size_of::<bpf_tunnel_key>() as u32, 2) }
+        != 0
+        || unsafe { bpf_skb_set_tunnel_opt(skb, (&mut option as *mut u32).cast(), 4) } != 0
+    {
+        return Err(-1);
+    }
+    ctx.set_mark(rule.mark | XFRM_ROUTE_BIT);
+    Ok(unsafe { bpf_redirect(rule.transport, 0) } as i32)
+}
+
+#[classifier]
+pub fn nullnet_policy_in(ctx: TcContext) -> i32 {
+    incoming(&ctx).unwrap_or(TC_ACT_SHOT)
+}
+
+fn incoming(ctx: &TcContext) -> Result<i32, i64> {
+    if !alive() {
+        return Err(-1);
+    }
+    let (len, proto) = header(ctx)?;
+    let local = u32::from_be(ctx.load::<u32>(30)?);
+    let rule = unsafe { POLICY_IN.get(local) }.copied().ok_or(-1)?;
+    let mut tunnel: bpf_tunnel_key = unsafe { zeroed() };
+    let mut option = 0u32;
+    let skb = ctx.as_ptr().cast();
+    if unsafe { (*ctx.skb.skb).mark } != rule.mark
+        || u32::from_be(ctx.load::<u32>(26)?) != rule.peer
+        || unsafe {
+            bpf_skb_get_tunnel_key(skb, &mut tunnel, size_of::<bpf_tunnel_key>() as u32, 0)
+        } != 0
+        || unsafe { bpf_skb_get_tunnel_opt(skb, (&mut option as *mut u32).cast(), 4) } != 4
+    {
+        return Err(-1);
+    }
+    if unsafe { tunnel.__bindgen_anon_1.remote_ipv4 } != rule.underlay_remote
+        || (u64::from(option & 0xffff) << 24 | u64::from(tunnel.tunnel_id)) != rule.generation
+    {
+        return Err(-1);
+    }
+    flow(ctx, &rule, rule.output, len, proto, true)?;
+    translate(ctx, 30, APP_IP, len, proto)?;
+    ctx.store(0, &rule.mac, 0)?;
+    ctx.set_mark(0);
+    Ok(unsafe { bpf_redirect(rule.output, 0) } as i32)
+}

members/nullnet-client/src/commands/endpoint_namespace.rs

@@ -15,6 +15,10 @@ pub(crate) struct EndpointNamespace {
 }
 
 impl EndpointNamespace {
+    pub(crate) async fn root() -> io::Result<Arc<Self>> {
+        Self::open(File::open("/proc/self/ns/net")?, None).await
+    }
+
     pub(crate) fn validate(&self) -> io::Result<()> {
         if let Some(process) = &self.process {
             let mut poll = libc::pollfd {

members/nullnet-client/src/commands/mod.rs

@@ -16,9 +16,12 @@ mod native_xfrm;
 mod netlink;
 pub(crate) mod nfqueue;
 mod ovs;
+pub(crate) mod policy;
 pub(crate) mod vxlan;
 mod vxlan_cleanup;
+pub(crate) mod vxlan_environment;
 mod vxlan_recovery;
+mod xfrm_shards;
 
 pub(crate) async fn setup_br0(rtnetlink_handle: &RtNetLinkHandle) {
     // create the bridge
@@ -114,6 +117,21 @@ pub(crate) fn find_ethernet_interface(ip: Ipv4Addr) -> Option<String> {
         })
 }
 
+pub(crate) async fn initialize_policy_shards(
+    rtnetlink_handle: &RtNetLinkHandle,
+) -> Result<(), Error> {
+    let ip = find_ethernet_ip(rtnetlink_handle)
+        .await
+        .ok_or("could not find the local ethernet IP")
+        .handle_err(location!())?;
+    let interface = find_ethernet_interface(ip)
+        .ok_or("could not find the local ethernet interface")
+        .handle_err(location!())?;
+    xfrm_shards::initialize(&rtnetlink_handle.handle, &interface)
+        .await
+        .handle_err(location!())
+}
+
 #[derive(Clone)]
 pub(crate) struct RtNetLinkHandle {
     handle: Handle,
@@ -151,6 +169,12 @@ pub(crate) async fn cleanup_network(
     egress::init();
     install_overlay_forwarding().handle_err(location!())?;
     let mss_error = install_mss_clamp();
+    xfrm_shards::recover(&rtnetlink_handle.handle)
+        .await
+        .handle_err(location!())?;
+    policy::recover(&rtnetlink_handle.handle)
+        .await
+        .handle_err(location!())?;
     vxlan_recovery::recover(&rtnetlink_handle.handle)
         .await
         .handle_err(location!())?;
@@ -171,31 +195,27 @@ pub(crate) async fn cleanup_network(
 
 // Preserve routed overlay traffic without changing the host's FORWARD policy.
 fn install_overlay_forwarding() -> std::io::Result<()> {
-    for direction in ["--src-group", "--dst-group"] {
-        let rule = [
-            "FORWARD",
-            "-m",
-            "devgroup",
-            direction,
-            "0x4e000000/0xffc00000",
-            "-j",
-            "ACCEPT",
-        ];
-        let check = std::process::Command::new("iptables")
-            .args(["-w", "-C"])
-            .args(rule)
-            .output()?;
-        if check.status.success() {
-            continue;
-        }
-        let installed = std::process::Command::new("iptables")
-            .args(["-w", "-I"])
-            .args(rule)
-            .output()?;
-        if !installed.status.success() {
-            return Err(std::io::Error::other(
-                String::from_utf8_lossy(&installed.stderr).into_owned(),
-            ));
+    for group in ["0x4e000000/0xffc00000", "0x4e700000/0xffffffff"] {
+        for direction in ["--src-group", "--dst-group"] {
+            let rule = [
+                "FORWARD", "-m", "devgroup", direction, group, "-j", "ACCEPT",
+            ];
+            let check = std::process::Command::new("iptables")
+                .args(["-w", "-C"])
+                .args(rule)
+                .output()?;
+            if check.status.success() {
+                continue;
+            }
+            let installed = std::process::Command::new("iptables")
+                .args(["-w", "-I"])
+                .args(rule)
+                .output()?;
+            if !installed.status.success() {
+                return Err(std::io::Error::other(
+                    String::from_utf8_lossy(&installed.stderr).into_owned(),
+                ));
+            }
         }
     }
     Ok(())
@@ -206,6 +226,8 @@ fn install_overlay_forwarding() -> std::io::Result<()> {
 /// the VXLAN maximum (see nullnet-server's `net_id_pool`).
 const XFRM_SPI_MIN: u32 = 1_000 + 101;
 const XFRM_SPI_MAX: u32 = 1_000 + 2_097_151;
+const POLICY_SPI_MIN: u32 = 0x4e40_0000;
+const POLICY_SPI_MAX: u32 = 0x4e7f_ffff;
 
 /// Delete IPsec state/policy pairs left behind by a previous run.
 ///
@@ -302,9 +324,9 @@ fn parse_src_dst(line: &str) -> Option<(String, String)> {
 fn parse_our_spi(line: &str) -> Option<String> {
     let raw = line.split_whitespace().skip_while(|t| *t != "spi").nth(1)?;
     let value = u32::from_str_radix(raw.strip_prefix("0x")?, 16).ok()?;
-    (XFRM_SPI_MIN..=XFRM_SPI_MAX)
-        .contains(&value)
-        .then(|| raw.to_string())
+    ((XFRM_SPI_MIN..=XFRM_SPI_MAX).contains(&value)
+        || (POLICY_SPI_MIN..=POLICY_SPI_MAX).contains(&value))
+    .then(|| raw.to_string())
 }
 
 fn privileged_output(args: &[&str]) -> Option<String> {