diff --git a/members/nullnet-client/src/commands/dnat.rs b/members/nullnet-client/src/commands/dnat.rs index 46b704a..1e7e484 100644 --- a/members/nullnet-client/src/commands/dnat.rs +++ b/members/nullnet-client/src/commands/dnat.rs @@ -28,23 +28,38 @@ pub(crate) fn init() { /// Install a DNAT for `port → overlay_ip:port`. When `container_ip` is a /// real address, the rule is scoped to that source via `-s` so co-located -/// replicas hit independent chains. `Ipv4Addr::UNSPECIFIED` (0.0.0.0) means -/// "no source filter" — used by legacy callers that don't know the source. +/// replicas hit independent chains. When `dest_ip` is a real address, the +/// rule is additionally scoped to that destination via `-d` — so two +/// backend-trigger dependencies sharing a port from the same initiator (each +/// with its own placeholder address, see `placeholder.rs`) get independent +/// rules instead of one clobbering the other. `Ipv4Addr::UNSPECIFIED` +/// (0.0.0.0) means "no filter" for either — used by legacy callers that +/// don't know the source/destination. /// Returns `false` if any of the per-proto `iptables` rules failed to apply. -pub(crate) fn install(port: u16, overlay_ip: Ipv4Addr, container_ip: Ipv4Addr) -> bool { +pub(crate) fn install( + port: u16, + overlay_ip: Ipv4Addr, + container_ip: Ipv4Addr, + dest_ip: Ipv4Addr, +) -> bool { let mut ok = true; for proto in PROTOS { - ok &= run_iptables("-A", proto, port, overlay_ip, container_ip); + ok &= run_iptables("-A", proto, port, overlay_ip, container_ip, dest_ip); } flush_conntrack(port, container_ip); ok } /// Returns `false` if any of the per-proto `iptables` rules failed to delete. -pub(crate) fn remove(port: u16, overlay_ip: Ipv4Addr, container_ip: Ipv4Addr) -> bool { +pub(crate) fn remove( + port: u16, + overlay_ip: Ipv4Addr, + container_ip: Ipv4Addr, + dest_ip: Ipv4Addr, +) -> bool { let mut ok = true; for proto in PROTOS { - ok &= run_iptables("-D", proto, port, overlay_ip, container_ip); + ok &= run_iptables("-D", proto, port, overlay_ip, container_ip, dest_ip); } flush_conntrack(port, container_ip); ok @@ -57,14 +72,19 @@ fn run_iptables( port: u16, overlay_ip: Ipv4Addr, container_ip: Ipv4Addr, + dest_ip: Ipv4Addr, ) -> bool { let port_s = port.to_string(); let target = format!("{overlay_ip}:{port}"); let container_ip_s = container_ip.to_string(); + let dest_ip_s = dest_ip.to_string(); let mut args: Vec<&str> = vec!["iptables", "-t", "nat", action, CHAIN, "-p", proto]; if !container_ip.is_unspecified() { args.extend_from_slice(&["-s", &container_ip_s]); } + if !dest_ip.is_unspecified() { + args.extend_from_slice(&["-d", &dest_ip_s]); + } args.extend_from_slice(&[ "--dport", &port_s, @@ -79,19 +99,28 @@ fn run_iptables( } else { container_ip_s.clone() }; + let dst = if dest_ip.is_unspecified() { + "any".to_string() + } else { + dest_ip_s.clone() + }; match status { Ok(s) if s.success() => { - println!("[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -> {target}"); + println!( + "[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -d {dst} -> {target}" + ); true } Ok(s) => { eprintln!( - "[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -> {target} exited {s}" + "[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -d {dst} -> {target} exited {s}" ); false } Err(e) => { - eprintln!("[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -> {target}: {e}"); + eprintln!( + "[dnat] iptables {action} {CHAIN} {proto}/{port} -s {src} -d {dst} -> {target}: {e}" + ); false } } diff --git a/members/nullnet-client/src/commands/egress.rs b/members/nullnet-client/src/commands/egress.rs index df88b1e..92b4567 100644 --- a/members/nullnet-client/src/commands/egress.rs +++ b/members/nullnet-client/src/commands/egress.rs @@ -95,6 +95,19 @@ pub(crate) fn init() { &["ipset", "add", "-exist", INTERNAL_SET, range], ); } + // The backend-trigger placeholder block (see `placeholder::placeholder_cidr`) + // is synthetic — never real internet traffic — but isn't one of the + // static private/special ranges above. Without this, a trigger dial to + // a port `nullnet_watched_ports` hasn't caught up on yet (a startup race + // against the gRPC round trip that populates it, see `nfqueue::apply_ports_diff`) + // falls through to this rule instead and gets misclassified as an egress + // candidate — held for the egress-trigger timeout, then dropped, well + // past most callers' own request timeout. + let placeholder_cidr = crate::placeholder::placeholder_cidr(); + sudo_ok( + "ipset add internal placeholder", + &["ipset", "add", "-exist", INTERNAL_SET, &placeholder_cidr], + ); // NFQUEUE trigger rule: NEW flows to non-internal destinations → queue 1. // The listener filters by container (registered services only). --queue-bypass @@ -213,6 +226,27 @@ pub(crate) fn install_steer( ], ); } + // Same bypass for the backend-trigger placeholder block — see the + // matching comment in `init()`. Lands at base+9, still clear of the + // catch-all at base+15. + let placeholder_cidr = crate::placeholder::placeholder_cidr(); + let placeholder_prio = (base + INTERNAL_RANGES.len() as u32).to_string(); + ok &= sudo_ok( + "ip rule add placeholder bypass", + &[ + "ip", + "rule", + "add", + "from", + &cip, + "to", + &placeholder_cidr, + "lookup", + "main", + "priority", + &placeholder_prio, + ], + ); // Catch-all: everything else from this container → the egress table. let catch_all = (base + 15).to_string(); ok &= sudo_ok( diff --git a/members/nullnet-client/src/control_channel.rs b/members/nullnet-client/src/control_channel.rs index c203fce..aecf3f9 100644 --- a/members/nullnet-client/src/control_channel.rs +++ b/members/nullnet-client/src/control_channel.rs @@ -2,9 +2,7 @@ use crate::commands::{RtNetLinkHandle, configure_access_port, dnat, egress, remo use crate::ebpf::{FirewallPeers, FirewallVxlanPorts, NetId}; use crate::egress_policy::{PolicyVerdicts, flush_container_conntrack}; use crate::egress_state::{EgressRecord, EgressState}; -use crate::host_mappings::{ - HOSTS_MARKER, HostMappingsState, edit_container_hosts, hosts_file_lock, -}; +use crate::host_mappings::{self, HOSTS_MARKER, HostMappingsState, hosts_file_lock}; use crate::nfqueue::BridgeIpCache; use crate::peers::peer::{Peers, VethKey}; use crate::triggers::TriggersState; @@ -531,6 +529,7 @@ async fn handle_vxlan_setup( if let Some(container) = message.docker_container.as_deref() { triggers_state.mark_active( container, + crate::triggers::EGRESS_DST_IP, crate::triggers::EGRESS_TRIGGER_PORT, vxlan_id, gw, @@ -590,7 +589,7 @@ async fn handle_vxlan_setup( // packet traverses `nat PREROUTING`. The DNAT rule MUST already be // installed by then, so we: // 1. peek the initiator's bridge IP (stashed at `mark_pending`) - // 2. install DNAT with `-s ` + // 2. install DNAT with `-s -d ` // 3. mark_active → wakes the waiter, packet released into the new // rule if let Some(dnat_port) = message.dnat_port @@ -598,14 +597,21 @@ async fn handle_vxlan_setup( && let Ok(overlay_ip) = host_mapping.ip.parse::() { let container_key = message.docker_container.as_deref().unwrap_or(""); - let container_ip = triggers_state.peek_container_ip(container_key, dnat_port); + // `host_mapping.name` is chain[0] for this trigger — the same + // literal name the NFQUEUE listener read from its port→target + // map when it observed the original packet — so recomputing the + // placeholder here independently lands on the exact same + // address it used, without needing it on the wire separately. + let dest_ip = crate::placeholder::ip_for(&host_mapping.name); + let container_ip = triggers_state.peek_container_ip(container_key, dest_ip, dnat_port); // Only promote to Active if the DNAT rule is actually live. Waking // the held packet without it would release the SYN into a missing // rule (→ misroute to the original dest); instead leave it Pending // so the listener drops at ACTIVE_TIMEOUT. - if dnat::install(dnat_port, overlay_ip, container_ip) { + if dnat::install(dnat_port, overlay_ip, container_ip, dest_ip) { triggers_state.mark_active( container_key, + dest_ip, dnat_port, vxlan_id, overlay_ip, @@ -695,30 +701,52 @@ async fn handle_vxlan_teardown( firewall_vxlan_ports.remove(message.vxlan_id); // remove DNAT before tearing the tunnel down so existing flows reset - // cleanly. The `container_ip` matches the `-s` we used at install time. - // `remove_by_vxlan` also matches the egress steer's sentinel-port entry - // (EGRESS_TRIGGER_PORT = 0). That path installs a policy-route steer, not a - // DNAT — its teardown runs via EgressState below — so skip DNAT removal for - // it. Only real backend DNAT ports (>0) go through `dnat::remove`; otherwise - // we'd fire a bogus `iptables -D --dport 0` and a false removal-failed event. - if let Some((_container, port, overlay_ip, container_ip)) = + // cleanly. The `container_ip`/`dst_ip` match the `-s`/`-d` we used at + // install time. `remove_by_vxlan` also matches the egress steer's + // sentinel-port entry (EGRESS_TRIGGER_PORT = 0). That path installs a + // policy-route steer, not a DNAT — its teardown runs via EgressState + // below — so skip DNAT removal for it. Only real backend DNAT ports (>0) + // go through `dnat::remove`; otherwise we'd fire a bogus + // `iptables -D --dport 0` and a false removal-failed event. Whether this + // was a backend-trigger entry (vs. an egress steer) also decides how the + // host mapping below gets torn down. + let mut is_backend_trigger_entry = false; + if let Some((_container, dst_ip, port, overlay_ip, container_ip)) = triggers_state.remove_by_vxlan(message.vxlan_id) && port != crate::triggers::EGRESS_TRIGGER_PORT - && !dnat::remove(port, overlay_ip, container_ip) { - fire_event( - &grpc, - AgentEventKind::DnatRemovalFailed(AgentDnatRemovalFailed { - port: u32::from(port), - overlay_ip: overlay_ip.to_string(), - }), - ); + is_backend_trigger_entry = true; + if !dnat::remove(port, overlay_ip, container_ip, dst_ip) { + fire_event( + &grpc, + AgentEventKind::DnatRemovalFailed(AgentDnatRemovalFailed { + port: u32::from(port), + overlay_ip: overlay_ip.to_string(), + }), + ); + } } - // remove host mapping if one was installed at setup + // Remove the host mapping installed at setup — except for a + // backend-trigger entry, where we re-seed the placeholder instead of + // deleting the line outright. Deleting would leave the initiator unable + // to resolve the name at all: the next `connect(name, ...)` would + // dead-end exactly like an un-seeded bare name does (no packet, no + // re-trigger, chain never rebuilds). Proxy-dependency mappings are + // unaffected — a fresh proxy request rebuilds the chain (and writes the + // real mapping) before forwarding, so there's no gap to cover there. if let Some((host_mapping, docker_container)) = host_mappings_state.take_vxlan(message.vxlan_id) { - let _ = remove_host_mapping(&host_mapping, docker_container.as_deref()); + if is_backend_trigger_entry && let Some(container) = docker_container.as_deref() { + if let Err(e) = crate::placeholder::seed_placeholder(container, &host_mapping.name) { + eprintln!( + "[control_channel] failed to re-seed placeholder for '{}' in {container}: {e:?}", + host_mapping.name + ); + } + } else { + let _ = remove_host_mapping(&host_mapping, docker_container.as_deref()); + } } // teardown VXLAN on this machine @@ -901,41 +929,29 @@ fn add_host_mapping(hm: &HostMapping, docker_container: Option<&str>) -> Result< if let Some(container) = docker_container { // container-targeted: the resolver that needs this name lives inside // the container, so write only there and leave the host's file alone. - // Edited from the host side (see `read_container_hosts`) so a paused - // container is served just as well as a running one. + // Edited from the host side (see `host_mappings::edit_container_hosts`) + // so a paused container is served just as well as a running one. // // Bail rather than write on a failed read: treating a missing file as // empty content would truncate a live `/etc/hosts` down to this one // entry. - edit_container_hosts(container, |current| { - upsert_hosts_entry(current, &hm.name, &entry) + host_mappings::edit_container_hosts(container, |current| { + host_mappings::upsert_hosts_entry(current, &hm.name, &entry) }) .handle_err(location!())?; } else { // host-targeted: upsert into the host's /etc/hosts let content = std::fs::read_to_string(path).handle_err(location!())?; - std::fs::write(path, upsert_hosts_entry(&content, &hm.name, &entry)) - .handle_err(location!())?; + std::fs::write( + path, + host_mappings::upsert_hosts_entry(&content, &hm.name, &entry), + ) + .handle_err(location!())?; } Ok(()) } -fn upsert_hosts_entry(content: &str, name: &str, entry: &str) -> String { - let mut lines: Vec = content.lines().map(ToString::to_string).collect(); - let mut found = false; - for line in &mut lines { - if line.split_whitespace().skip(1).any(|tok| tok == name) { - *line = entry.to_string(); - found = true; - } - } - if !found { - lines.push(entry.to_string()); - } - lines.join("\n") + "\n" -} - fn remove_host_mapping(hm: &HostMapping, docker_container: Option<&str>) -> Result<(), Error> { let path = "/etc/hosts"; @@ -949,40 +965,23 @@ fn remove_host_mapping(hm: &HostMapping, docker_container: Option<&str>) -> Resu // the `docker pause` that follows it race, and losing that race used to // strand the entry, leaving the name pointing at a dead overlay IP for // as long as the container kept being resumed rather than restarted. - edit_container_hosts(container, |current| { - remove_hosts_entry(current, &hm.name, &hm.ip) + host_mappings::edit_container_hosts(container, |current| { + host_mappings::remove_hosts_entry(current, &hm.name, &hm.ip) }) .handle_err(location!())?; } else { // host-targeted: drop this net's line from the host file let content = std::fs::read_to_string(path).handle_err(location!())?; - std::fs::write(path, remove_hosts_entry(&content, &hm.name, &hm.ip)) - .handle_err(location!())?; + std::fs::write( + path, + host_mappings::remove_hosts_entry(&content, &hm.name, &hm.ip), + ) + .handle_err(location!())?; } Ok(()) } -/// Drop the line mapping `name`, but only while it still points at `ip`. -/// -/// NET IDs are recycled, so a teardown can land after a *newer* net has already -/// re-installed the same name at a different overlay IP (`upsert_hosts_entry` -/// keys on the name alone, which is what makes the replacement correct). -/// Matching the IP too makes the late teardown a no-op instead of deleting a -/// mapping that belongs to a live tunnel. -fn remove_hosts_entry(content: &str, name: &str, ip: &str) -> String { - let lines: Vec = content - .lines() - .filter(|line| { - let mut tokens = line.split_whitespace(); - let line_ip = tokens.next(); - !(line_ip == Some(ip) && tokens.any(|tok| tok == name)) - }) - .map(ToString::to_string) - .collect(); - lines.join("\n") + "\n" -} - /// Lowercase hex encoding, used to pass the tunnel's AES key to /// `vxlan-setup.sh`/`vxlan-teardown.sh` as a shell argument. fn hex_encode(bytes: &[u8]) -> String { diff --git a/members/nullnet-client/src/host_mappings.rs b/members/nullnet-client/src/host_mappings.rs index dab57bc..7aa8e4a 100644 --- a/members/nullnet-client/src/host_mappings.rs +++ b/members/nullnet-client/src/host_mappings.rs @@ -197,6 +197,48 @@ fn edit_hosts_file(path: &Path, edit: impl FnOnce(&str) -> String) -> Result ip` into `content`, tagged with [`HOSTS_MARKER`] so a +/// restarted process can sweep it later. Shared by two callers: the reactive +/// real-mapping write once a tunnel is up (`control_channel`'s +/// `add_host_mapping`) and the proactive placeholder seed written before any +/// packet exists (`placeholder::seed_placeholder`) — both need identical +/// upsert semantics and the same crash-safety (locking, tagging), just a +/// different `ip` value and I/O wrapper. +pub(crate) fn upsert_hosts_entry(content: &str, name: &str, entry: &str) -> String { + let mut lines: Vec = content.lines().map(ToString::to_string).collect(); + let mut found = false; + for line in &mut lines { + if line.split_whitespace().skip(1).any(|tok| tok == name) { + *line = entry.to_string(); + found = true; + } + } + if !found { + lines.push(entry.to_string()); + } + lines.join("\n") + "\n" +} + +/// Drop the line mapping `name`, but only while it still points at `ip`. +/// +/// NET IDs are recycled, so a teardown can land after a *newer* net has +/// already re-installed the same name at a different overlay IP +/// (`upsert_hosts_entry` keys on the name alone, which is what makes the +/// replacement correct). Matching the IP too makes the late teardown a no-op +/// instead of deleting a mapping that belongs to a live tunnel. +pub(crate) fn remove_hosts_entry(content: &str, name: &str, ip: &str) -> String { + let lines: Vec = content + .lines() + .filter(|line| { + let mut tokens = line.split_whitespace(); + let line_ip = tokens.next(); + !(line_ip == Some(ip) && tokens.any(|tok| tok == name)) + }) + .map(ToString::to_string) + .collect(); + lines.join("\n") + "\n" +} + #[cfg(test)] mod edit_tests { use super::{HOSTS_MARKER, edit_hosts_file, strip_marked}; @@ -278,3 +320,42 @@ mod tests { assert_eq!(strip_marked(content), content); } } + +#[cfg(test)] +mod hosts_entry_tests { + use super::*; + + #[test] + fn upsert_appends_when_absent() { + let out = upsert_hosts_entry("127.0.0.1 localhost\n", "redis", "203.0.113.7 redis"); + assert_eq!(out, "127.0.0.1 localhost\n203.0.113.7 redis\n"); + } + + #[test] + fn upsert_replaces_existing_line() { + let out = upsert_hosts_entry( + "127.0.0.1 localhost\n203.0.113.7 redis\n", + "redis", + "10.0.0.5 redis", + ); + assert_eq!(out, "127.0.0.1 localhost\n10.0.0.5 redis\n"); + } + + #[test] + fn remove_drops_matching_line_only() { + let out = remove_hosts_entry( + "127.0.0.1 localhost\n10.0.0.5 redis\n10.0.0.6 billing\n", + "redis", + "10.0.0.5", + ); + assert_eq!(out, "127.0.0.1 localhost\n10.0.0.6 billing\n"); + } + + #[test] + fn remove_is_a_noop_when_ip_no_longer_matches() { + // A late teardown for a torn-down tunnel landing after a newer tunnel + // re-mapped the same name at a different IP must not delete it. + let out = remove_hosts_entry("127.0.0.1 localhost\n10.0.0.9 redis\n", "redis", "10.0.0.5"); + assert_eq!(out, "127.0.0.1 localhost\n10.0.0.9 redis\n"); + } +} diff --git a/members/nullnet-client/src/main.rs b/members/nullnet-client/src/main.rs index 36d3e8f..1012f0b 100644 --- a/members/nullnet-client/src/main.rs +++ b/members/nullnet-client/src/main.rs @@ -10,7 +10,6 @@ use crate::forward::receive::receive; use crate::forward::send::send; use crate::host_mappings::HostMappingsState; use crate::local_endpoints::LocalEndpoints; -use crate::nfqueue::{TriggerMap, TriggerOwner}; use crate::peers::peer::Peers; use crate::triggers::TriggersState; use clap::Parser; @@ -42,6 +41,7 @@ mod host_mappings; mod local_endpoints; mod nfqueue; mod peers; +mod placeholder; mod triggers; pub const FORWARD_PORT: u16 = 9999; @@ -162,7 +162,8 @@ async fn main() -> Result<(), Error> { // packet of each new watched-port flow, listener fires backend_trigger // with the resolved initiator container, waits for VxlanSetup to install // DNAT, then verdicts ACCEPT so the original packet hits the new chain. - let (config_tx, config_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (config_tx, config_rx) = + tokio::sync::mpsc::unbounded_channel::>>(); // Poked by the cache's docker-events watcher after every container // start/die so `declare_services` re-runs immediately instead of @@ -182,8 +183,8 @@ async fn main() -> Result<(), Error> { policy_verdicts, ); - // declare services + push the port→trigger-owners map to the NFQUEUE - // listener on each refresh. + // declare services + push the port→target map to the NFQUEUE listener + // on each refresh. tokio::spawn(async move { declare_services(grpc_server, config_tx, docker_changed) .await @@ -292,7 +293,7 @@ async fn grpc_init() -> Result { async fn declare_services( grpc_server: NullnetGrpcInterface, - config_tx: UnboundedSender, + config_tx: UnboundedSender>>, docker_changed: Arc, ) -> Result<(), Error> { let mut last_snapshot: Vec = Vec::new(); @@ -371,22 +372,49 @@ async fn declare_services( }); } - // One port may be claimed by several services on this node, so - // each maps to a list of owners rather than a single service. - let mut trigger_owners: TriggerMap = HashMap::new(); + // More than one target can share a port — either the same + // service's own multiple dependency chains (two plain-HTTPS + // deps both 443, disambiguated by destination), or an + // unrelated service/container that just happens to talk to + // the same port number (disambiguated by source container). + // Group by port so the listener can apply both. + let mut port_to_target: HashMap> = HashMap::new(); for st in response.service_triggers { - for port in st.ports { - let Ok(port) = u16::try_from(port) else { - eprintln!("server returned invalid trigger port {port}; skipping"); + for tp in st.trigger_ports { + let Ok(port) = u16::try_from(tp.port) else { + eprintln!("server returned invalid trigger port {}; skipping", tp.port); continue; }; - trigger_owners.entry(port).or_default().push(TriggerOwner { - service: st.service_name.clone(), - containers: st.containers.clone(), - }); + // Pre-seed a placeholder /etc/hosts entry for the + // dependency's literal name in every replica + // container hosting the initiator service, *before* + // any packet is observed, so a bare container name + // (not just a pre-provisioned DNS alias) produces a + // real first packet for NFQUEUE to catch. Idempotent + // — safe on every reconcile pass; self-heals across + // container restarts (Docker wipes /etc/hosts on + // every start). + for container in &st.containers { + if let Err(e) = + placeholder::seed_placeholder(container, &tp.target_name) + { + eprintln!( + "failed to seed placeholder for '{}' in {container}: {e:?}", + tp.target_name + ); + } + } + port_to_target + .entry(port) + .or_default() + .push(nfqueue::TriggerTarget { + service_name: st.service_name.clone(), + target_name: tp.target_name.clone(), + containers: st.containers.clone(), + }); } } - if config_tx.send(trigger_owners).is_err() { + if config_tx.send(port_to_target).is_err() { // observer task gone; nothing more to do here return Ok(()); } diff --git a/members/nullnet-client/src/nfqueue/cache.rs b/members/nullnet-client/src/nfqueue/cache.rs index 5fed8e2..d18de1a 100644 --- a/members/nullnet-client/src/nfqueue/cache.rs +++ b/members/nullnet-client/src/nfqueue/cache.rs @@ -1,3 +1,4 @@ +use crate::triggers::TriggersState; use std::collections::{HashMap, HashSet}; use std::net::Ipv4Addr; use std::process::Stdio; @@ -280,15 +281,22 @@ fn strip_cidr_to_ipv4(ip_with_mask: &str) -> Option { } /// Spawn the long-running `docker events` watcher. Triggers a refresh after -/// every container start/die and pokes `docker_changed` so the -/// declare-services loop in `main` re-runs immediately. If docker isn't -/// installed or the subprocess can't be spawned, the task logs and exits — -/// listener falls back to the initial cache snapshot. Restarts the -/// subprocess on unexpected exit. -pub fn spawn_events_watcher(cache: BridgeIpCache, docker_changed: Arc) { +/// every container start/die, purges any stale trigger state for that +/// specific container (see `forget_container`'s doc comment — a restart or +/// recreate keeps the container's name but gets a new sandbox, so a stale +/// `TriggersState` entry would otherwise wrongly look `Active` forever), and +/// pokes `docker_changed` so the declare-services loop in `main` re-runs +/// immediately. If docker isn't installed or the subprocess can't be +/// spawned, the task logs and exits — listener falls back to the initial +/// cache snapshot. Restarts the subprocess on unexpected exit. +pub fn spawn_events_watcher( + cache: BridgeIpCache, + docker_changed: Arc, + triggers_state: Arc, +) { tokio::spawn(async move { loop { - if let Err(e) = run_events_loop(&cache, &docker_changed).await { + if let Err(e) = run_events_loop(&cache, &docker_changed, &triggers_state).await { eprintln!("[nfqueue/cache] events watcher: {e}; restarting in 5s"); tokio::time::sleep(std::time::Duration::from_secs(5)).await; } @@ -296,7 +304,11 @@ pub fn spawn_events_watcher(cache: BridgeIpCache, docker_changed: Arc) { }); } -async fn run_events_loop(cache: &BridgeIpCache, docker_changed: &Notify) -> Result<(), String> { +async fn run_events_loop( + cache: &BridgeIpCache, + docker_changed: &Notify, + triggers_state: &TriggersState, +) -> Result<(), String> { let mut child = tokio::process::Command::new("docker") .args([ "events", @@ -308,9 +320,12 @@ async fn run_events_loop(cache: &BridgeIpCache, docker_changed: &Notify) -> Resu "event=die", // `.Action` (start/die) is the modern field; the legacy // `.Status` was removed in newer daemons (template eval errors - // "can't evaluate field Status in type *events.Message"). + // "can't evaluate field Status in type *events.Message"). The + // container name is needed (not just the action) so a + // restart/recreate can purge that one container's stale + // TriggersState entries instead of guessing which one changed. "--format", - "{{.Action}}", + "{{.Action}} {{.Actor.Attributes.name}}", ]) .stdout(Stdio::piped()) .stderr(Stdio::piped()) @@ -342,10 +357,16 @@ async fn run_events_loop(cache: &BridgeIpCache, docker_changed: &Notify) -> Resu .await .map_err(|e| format!("read docker events: {e}"))? { - // We don't parse the line — any container start/die warrants a - // full refresh. Cheap enough: a few processes per event. + // Beyond action+name we don't otherwise parse the line — any + // container start/die warrants a full cache refresh. Cheap enough: + // a few processes per event. println!("[nfqueue/cache] docker event: {line} — refreshing"); cache.refresh().await; + if let Some((_, name)) = line.split_once(' ') + && !name.is_empty() + { + triggers_state.forget_container(name); + } // Cache now reflects the new task; kick the declare-services // loop so the ipset catches up before the new task dials. docker_changed.notify_one(); diff --git a/members/nullnet-client/src/nfqueue/egress_listener.rs b/members/nullnet-client/src/nfqueue/egress_listener.rs index cf98db4..1071ef1 100644 --- a/members/nullnet-client/src/nfqueue/egress_listener.rs +++ b/members/nullnet-client/src/nfqueue/egress_listener.rs @@ -19,7 +19,7 @@ use crate::egress_policy::PolicyVerdicts; use crate::nfqueue::cache::BridgeIpCache; use crate::nfqueue::parse::ipv4_flow; use crate::nfqueue::recv_loop::spawn_queue_loop; -use crate::triggers::{EGRESS_TRIGGER_PORT, TriggerState, TriggersState}; +use crate::triggers::{EGRESS_DST_IP, EGRESS_TRIGGER_PORT, TriggerState, TriggersState}; use nfq::{Message, Verdict}; use nullnet_grpc_lib::NullnetGrpcInterface; use nullnet_grpc_lib::nullnet_grpc::{ @@ -157,13 +157,19 @@ async fn decide_verdict(ctx: &EgressCtx, flow: Option<(Ipv4Addr, Ipv4Addr, u16)> return Verdict::Drop; } - match ctx.triggers_state.state(&container, EGRESS_TRIGGER_PORT) { + match ctx + .triggers_state + .state(&container, EGRESS_DST_IP, EGRESS_TRIGGER_PORT) + { TriggerState::Active => Verdict::Accept, TriggerState::Pending(notify) => wait_for_steer(ctx, &container, notify).await, TriggerState::Fresh => { - let notify = ctx - .triggers_state - .mark_pending(&container, EGRESS_TRIGGER_PORT, src_ip); + let notify = ctx.triggers_state.mark_pending( + &container, + EGRESS_DST_IP, + EGRESS_TRIGGER_PORT, + src_ip, + ); // Register the waiter BEFORE the gRPC round-trip: the server can // dispatch the egress `VxlanSetup` (→ `mark_active`) faster than its // reply to `egress_trigger` returns, and `notify_waiters` only wakes @@ -172,7 +178,8 @@ async fn decide_verdict(ctx: &EgressCtx, flow: Option<(Ipv4Addr, Ipv4Addr, u16)> tokio::pin!(notified); if notified.as_mut().enable() || matches!( - ctx.triggers_state.state(&container, EGRESS_TRIGGER_PORT), + ctx.triggers_state + .state(&container, EGRESS_DST_IP, EGRESS_TRIGGER_PORT), TriggerState::Active ) { @@ -202,7 +209,8 @@ async fn decide_verdict(ctx: &EgressCtx, flow: Option<(Ipv4Addr, Ipv4Addr, u16)> Ok(Err(e)) => { eprintln!("[egress-nfq] egress_trigger {container}: {e}"); report_trigger_send_failed(&ctx.grpc, &container, dst_ip, dst_port, e); - ctx.triggers_state.forget(&container, EGRESS_TRIGGER_PORT); + ctx.triggers_state + .forget(&container, EGRESS_DST_IP, EGRESS_TRIGGER_PORT); Verdict::Drop } Err(_) => { @@ -214,7 +222,8 @@ async fn decide_verdict(ctx: &EgressCtx, flow: Option<(Ipv4Addr, Ipv4Addr, u16)> dst_port, format!("egress_trigger timed out after {TRIGGER_TIMEOUT:?}"), ); - ctx.triggers_state.forget(&container, EGRESS_TRIGGER_PORT); + ctx.triggers_state + .forget(&container, EGRESS_DST_IP, EGRESS_TRIGGER_PORT); Verdict::Drop } } @@ -261,7 +270,8 @@ async fn wait_for_steer(ctx: &EgressCtx, container: &str, notify: Arc) - tokio::pin!(notified); if notified.as_mut().enable() || matches!( - ctx.triggers_state.state(container, EGRESS_TRIGGER_PORT), + ctx.triggers_state + .state(container, EGRESS_DST_IP, EGRESS_TRIGGER_PORT), TriggerState::Active ) { diff --git a/members/nullnet-client/src/nfqueue/listener.rs b/members/nullnet-client/src/nfqueue/listener.rs index 03c6c4b..979648b 100644 --- a/members/nullnet-client/src/nfqueue/listener.rs +++ b/members/nullnet-client/src/nfqueue/listener.rs @@ -1,5 +1,5 @@ use crate::nfqueue::cache::BridgeIpCache; -use crate::nfqueue::parse::ipv4_src_and_dst_port; +use crate::nfqueue::parse::ipv4_flow; use crate::nfqueue::recv_loop::spawn_queue_loop; use crate::triggers::{TriggerState, TriggersState}; use nfq::{Message, Verdict}; @@ -8,6 +8,7 @@ use nullnet_grpc_lib::nullnet_grpc::{ AgentBackendTriggerSendFailed, AgentEvent, agent_event::Event as AgentEventKind, }; use std::collections::HashMap; +use std::net::Ipv4Addr; use std::sync::mpsc::Sender; use std::sync::{Arc, RwLock}; use std::time::Duration; @@ -32,46 +33,79 @@ const TRIGGER_TIMEOUT: Duration = Duration::from_secs(5); /// giving up on the held packet. const ACTIVE_TIMEOUT: Duration = Duration::from_secs(5); -/// A service that declared a trigger on a port, and the containers on this node -/// that host it. -#[derive(Debug, Clone)] -pub struct TriggerOwner { - pub service: String, - /// Real container names, matching the bridge-IP cache's string space. Empty - /// means a server predating `ServiceTrigger.containers`, in which case the - /// owner matches any container — the old port-only behaviour. +/// What a watched trigger port is associated with: the declaring (initiator) +/// service, for reporting; the real container names hosting it on this node +/// (the same string space as the bridge-IP cache and `Container.real_name`) — +/// empty means a server predating this field, matching any source container; +/// and the literal name its chain[0] resolves to — what +/// `placeholder::seed_placeholder` pre-seeds into the initiator container's +/// `/etc/hosts`, and what disambiguates two dependencies that happen to +/// share a port (via their distinct placeholder addresses). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TriggerTarget { + pub service_name: String, + pub target_name: String, pub containers: Vec, } -/// Watched port → the services claiming it on this node. More than one may -/// claim the same port, each through its own replicas. -pub type TriggerMap = HashMap>; - -/// The service whose trigger `container` should fire on `port`, if any. -/// -/// The ipset that queues these packets matches on destination port alone, so a -/// port watched for one service catches every container on the node. Resolving -/// the owner by container is what keeps a co-located container's traffic from -/// being attributed to — and rejected by — someone else's trigger. -fn owner_for<'a>(map: &'a TriggerMap, container: &str, port: u16) -> Option<&'a str> { - let owners = map.get(&port)?; - owners - .iter() - .find(|o| o.containers.iter().any(|c| c == container)) - .or_else(|| owners.iter().find(|o| o.containers.is_empty())) - .map(|o| o.service.as_str()) -} - /// State shared by every per-packet handler. Cloned freely across tokio tasks. #[derive(Clone)] pub struct ListenerCtx { pub grpc: NullnetGrpcInterface, pub cache: BridgeIpCache, - pub trigger_owners: Arc>, + /// More than one target can share a port — either the same service's own + /// multiple dependency chains, or an unrelated service/container that + /// just happens to talk to the same port number (the ipset match is + /// destination-port only, so a watched port catches every container on + /// the node). `resolve_target` disambiguates in two stages: by source + /// container first, then by destination. + pub port_to_target: Arc>>>, pub triggers_state: Arc, pub semaphore: Arc, } +/// Pick the target this packet's (source container, destination) actually +/// belongs to. +/// +/// Two independent scoping axes, applied in sequence: +/// 1. By source container — a candidate explicitly listing `container` +/// always wins over one that doesn't. Only when *no* candidate names +/// `container` do the container-less (legacy) candidates apply — a +/// scoped candidate must beat a legacy catch-all for its own container +/// even though both may share a target_name/destination, which +/// destination-matching alone could never break the tie on. +/// 2. By destination — when more than one candidate survives step 1 (this +/// service's own multiple chains sharing the port, or several +/// same-priority candidates), an exact match against each candidate's +/// own deterministic placeholder address is required — there is no safe +/// default to fall back on, since guessing wrong would route the packet +/// into the wrong tunnel. A lone survivor is trusted unconditionally +/// (covers a destination that doesn't yet match its placeholder, e.g. +/// the very first packet on a freshly re-seeded name). +fn resolve_target<'a>( + targets: &'a [TriggerTarget], + container: &str, + dst_ip: Ipv4Addr, +) -> Option<&'a TriggerTarget> { + let scoped: Vec<&TriggerTarget> = targets + .iter() + .filter(|t| t.containers.iter().any(|c| c == container)) + .collect(); + let candidates: Vec<&TriggerTarget> = if scoped.is_empty() { + targets.iter().filter(|t| t.containers.is_empty()).collect() + } else { + scoped + }; + match candidates.as_slice() { + [] => None, + [only] => Some(only), + many => many + .iter() + .copied() + .find(|t| crate::placeholder::ip_for(&t.target_name) == dst_ip), + } +} + /// Spawn the backend-trigger recv loop (queue 0). Each packet is held until /// `handle_packet` resolves a verdict — see `recv_loop::spawn_queue_loop`. pub fn spawn_recv_thread(ctx: ListenerCtx) { @@ -100,7 +134,7 @@ async fn handle_packet(mut msg: Message, ctx: ListenerCtx, verdict_tx: Sender Verdict { - match ctx.triggers_state.state(container, dst_port) { + let service = target.service_name.as_str(); + match ctx.triggers_state.state(container, dst_ip, dst_port) { TriggerState::Active => Verdict::Accept, TriggerState::Pending(notify) => { // `mark_active` wakes us with `Notify::notify_waiters()`, which @@ -156,7 +203,7 @@ async fn decide_verdict( tokio::pin!(notified); if notified.as_mut().enable() || matches!( - ctx.triggers_state.state(container, dst_port), + ctx.triggers_state.state(container, dst_ip, dst_port), TriggerState::Active ) { @@ -173,7 +220,9 @@ async fn decide_verdict( } } TriggerState::Fresh => { - let notify = ctx.triggers_state.mark_pending(container, dst_port, src_ip); + let notify = ctx + .triggers_state + .mark_pending(container, dst_ip, dst_port, src_ip); // Register BEFORE the gRPC round-trip: the server can dispatch // `VxlanSetup` (→ `mark_active` here) faster than its reply to // `backend_trigger` arrives back, especially on multi-edge @@ -184,7 +233,7 @@ async fn decide_verdict( tokio::pin!(notified); if notified.as_mut().enable() || matches!( - ctx.triggers_state.state(container, dst_port), + ctx.triggers_state.state(container, dst_ip, dst_port), TriggerState::Active ) { @@ -196,6 +245,7 @@ async fn decide_verdict( service.to_string(), u32::from(dst_port), container.to_string(), + target.target_name.clone(), ), ) .await; @@ -219,7 +269,7 @@ async fn decide_verdict( "[nfqueue] backend_trigger '{service}' port {dst_port} container {container}: {e}" ); report_trigger_send_failed(&ctx.grpc, service, dst_port, e); - ctx.triggers_state.forget(container, dst_port); + ctx.triggers_state.forget(container, dst_ip, dst_port); Verdict::Drop } Err(_) => { @@ -232,7 +282,7 @@ async fn decide_verdict( dst_port, format!("backend_trigger timed out after {TRIGGER_TIMEOUT:?}"), ); - ctx.triggers_state.forget(container, dst_port); + ctx.triggers_state.forget(container, dst_ip, dst_port); Verdict::Drop } } @@ -265,76 +315,145 @@ fn report_trigger_send_failed( #[cfg(test)] mod tests { - use super::{TriggerMap, TriggerOwner, owner_for}; + use super::{TriggerTarget, resolve_target}; + use std::net::Ipv4Addr; - fn owner(service: &str, containers: &[&str]) -> TriggerOwner { - TriggerOwner { - service: service.to_string(), + fn target(service: &str, target_name: &str, containers: &[&str]) -> TriggerTarget { + TriggerTarget { + service_name: service.to_string(), + target_name: target_name.to_string(), containers: containers.iter().map(|c| (*c).to_string()).collect(), } } + fn dst_of(target_name: &str) -> Ipv4Addr { + crate::placeholder::ip_for(target_name) + } + /// The instaprotek case: `service` triggers on 8932, which is also /// `api-prod-v2`'s backend port. Every co-located container talks to that /// port, and before scoping each one was attributed to `service` — the /// server then found no matching replica and the client dropped the SYN. #[test] fn foreign_container_on_a_watched_port_is_not_attributed() { - let map: TriggerMap = TriggerMap::from([(8932, vec![owner("service", &["svc_c1"])])]); + let targets = [target("service", "dep", &["svc_c1"])]; + let dst = dst_of("dep"); - assert_eq!(owner_for(&map, "svc_c1", 8932), Some("service")); assert_eq!( - owner_for(&map, "portal_c1", 8932), + resolve_target(&targets, "svc_c1", dst).map(|t| t.service_name.as_str()), + Some("service") + ); + assert_eq!( + resolve_target(&targets, "portal_c1", dst), None, "a container that doesn't host the declaring service must pass through" ); } - #[test] - fn unwatched_port_is_a_miss() { - let map: TriggerMap = TriggerMap::from([(8932, vec![owner("service", &["svc_c1"])])]); - assert_eq!(owner_for(&map, "svc_c1", 9999), None); - } - /// Several replicas of one service on the same node all own its trigger. #[test] fn every_replica_of_the_declaring_service_owns_it() { - let map: TriggerMap = - TriggerMap::from([(8932, vec![owner("service", &["svc_c1", "svc_c2"])])]); - assert_eq!(owner_for(&map, "svc_c1", 8932), Some("service")); - assert_eq!(owner_for(&map, "svc_c2", 8932), Some("service")); + let targets = [target("service", "dep", &["svc_c1", "svc_c2"])]; + let dst = dst_of("dep"); + assert_eq!( + resolve_target(&targets, "svc_c1", dst).map(|t| t.service_name.as_str()), + Some("service") + ); + assert_eq!( + resolve_target(&targets, "svc_c2", dst).map(|t| t.service_name.as_str()), + Some("service") + ); } /// Two services may now claim the same port on one node; each resolves to /// its own, which was impossible while the map was keyed by port alone. #[test] fn two_services_can_share_a_port() { - let map: TriggerMap = TriggerMap::from([( - 8932, - vec![owner("service", &["svc_c1"]), owner("other", &["other_c1"])], - )]); - assert_eq!(owner_for(&map, "svc_c1", 8932), Some("service")); - assert_eq!(owner_for(&map, "other_c1", 8932), Some("other")); - assert_eq!(owner_for(&map, "stranger_c1", 8932), None); + let targets = [ + target("service", "dep-a", &["svc_c1"]), + target("other", "dep-b", &["other_c1"]), + ]; + assert_eq!( + resolve_target(&targets, "svc_c1", dst_of("dep-a")).map(|t| t.service_name.as_str()), + Some("service") + ); + assert_eq!( + resolve_target(&targets, "other_c1", dst_of("dep-b")).map(|t| t.service_name.as_str()), + Some("other") + ); + assert_eq!( + resolve_target(&targets, "stranger_c1", dst_of("dep-a")), + None + ); + } + + /// A server predating `TriggerTarget::containers` sends none, and must + /// keep behaving exactly as before: any container matches. + #[test] + fn container_less_target_matches_anything() { + let targets = [target("service", "dep", &[])]; + assert_eq!( + resolve_target(&targets, "anything", dst_of("dep")).map(|t| t.service_name.as_str()), + Some("service") + ); } - /// A server predating `ServiceTrigger.containers` sends none, and must keep - /// behaving exactly as before: any container matches. + /// Mixed fleet: a scoped target must win over a legacy catch-all for its + /// own container, and the catch-all still covers everyone else. #[test] - fn container_less_owner_matches_anything() { - let map: TriggerMap = TriggerMap::from([(8932, vec![owner("service", &[])])]); - assert_eq!(owner_for(&map, "anything", 8932), Some("service")); + fn scoped_target_wins_over_legacy_catch_all() { + let targets = [ + target("legacy", "dep", &[]), + target("scoped", "dep", &["scoped_c1"]), + ]; + assert_eq!( + resolve_target(&targets, "scoped_c1", dst_of("dep")).map(|t| t.service_name.as_str()), + Some("scoped") + ); + assert_eq!( + resolve_target(&targets, "someone_else", dst_of("dep")) + .map(|t| t.service_name.as_str()), + Some("legacy") + ); } - /// Mixed fleet: a scoped owner must win over a legacy catch-all for its own - /// container, and the catch-all still covers everyone else. + /// The original disambiguation this module was built around: one + /// container hosting two chains sharing a port, told apart only by + /// which placeholder address the packet was actually addressed to. #[test] - fn scoped_owner_wins_over_legacy_catch_all() { - let map: TriggerMap = TriggerMap::from([( - 8932, - vec![owner("legacy", &[]), owner("scoped", &["scoped_c1"])], - )]); - assert_eq!(owner_for(&map, "scoped_c1", 8932), Some("scoped")); - assert_eq!(owner_for(&map, "someone_else", 8932), Some("legacy")); + fn same_container_disambiguates_multiple_chains_by_destination() { + let targets = [ + target("portal", "dep-a", &["portal_c1"]), + target("portal", "dep-b", &["portal_c1"]), + ]; + assert_eq!( + resolve_target(&targets, "portal_c1", dst_of("dep-a")).map(|t| t.target_name.as_str()), + Some("dep-a") + ); + assert_eq!( + resolve_target(&targets, "portal_c1", dst_of("dep-b")).map(|t| t.target_name.as_str()), + Some("dep-b") + ); + } + + /// Both axes combined: container-scoping first narrows to `portal_c1`'s + /// own two candidates (excluding `other`'s, scoped to a different + /// container), then destination-matching picks the right chain among + /// what's left. + #[test] + fn container_scoping_and_destination_matching_compose() { + let targets = [ + target("portal", "dep-a", &["portal_c1"]), + target("portal", "dep-b", &["portal_c1"]), + target("other", "dep-c", &["other_c1"]), + ]; + assert_eq!( + resolve_target(&targets, "portal_c1", dst_of("dep-b")).map(|t| t.target_name.as_str()), + Some("dep-b") + ); + // portal_c1's own two candidates survive container-scoping, but + // neither's placeholder matches a destination that was never one of + // portal's own (dep-c belongs to "other", scoped to "other_c1"). + assert_eq!(resolve_target(&targets, "portal_c1", dst_of("dep-c")), None); } } diff --git a/members/nullnet-client/src/nfqueue/mod.rs b/members/nullnet-client/src/nfqueue/mod.rs index b00045a..2991efe 100644 --- a/members/nullnet-client/src/nfqueue/mod.rs +++ b/members/nullnet-client/src/nfqueue/mod.rs @@ -5,7 +5,7 @@ mod parse; mod recv_loop; pub use cache::BridgeIpCache; -pub use listener::{TriggerMap, TriggerOwner}; +pub use listener::TriggerTarget; use crate::commands::nfqueue as rules; use crate::egress_policy::PolicyVerdicts; @@ -24,8 +24,8 @@ use tokio::sync::mpsc::UnboundedReceiver; /// - Populates the bridge-IP → container-name cache from `docker inspect` /// and keeps it fresh via a `docker events` watcher. /// - Consumes `config_rx` (driven by the services-list refresh in `main`) to -/// keep the kernel ipset in sync and to maintain a port → service lookup -/// for the per-packet handler. +/// keep the kernel ipset in sync and to maintain a port → trigger-target +/// lookup for the per-packet handler. /// - Spawns the recv thread that owns the netfilter queue. Each packet is /// handed off to a tokio task; the recv thread drains verdicts in lockstep /// so packets release back into the netfilter pipeline. @@ -35,12 +35,13 @@ use tokio::sync::mpsc::UnboundedReceiver; pub fn spawn_listener( grpc: NullnetGrpcInterface, triggers_state: Arc, - config_rx: UnboundedReceiver, + config_rx: UnboundedReceiver>>, docker_changed: Arc, cache: BridgeIpCache, verdicts: Arc, ) { - let trigger_owners: Arc> = Arc::new(RwLock::new(HashMap::new())); + let port_to_target: Arc>>> = + Arc::new(RwLock::new(HashMap::new())); // Initial cache populate + long-running docker-events watcher. The // watcher pings `docker_changed` after every refresh so the @@ -49,20 +50,21 @@ pub fn spawn_listener( // might fire a SYN before its trigger port is being watched. { let bridge_cache = cache.clone(); + let triggers_state_for_events = triggers_state.clone(); tokio::spawn(async move { bridge_cache.refresh().await; - cache::spawn_events_watcher(bridge_cache, docker_changed); + cache::spawn_events_watcher(bridge_cache, docker_changed, triggers_state_for_events); }); } - // Config consumer: each services-list refresh produces a port → owners - // map. We diff the ports vs the previous set, push the diff to the ipset - // (so the kernel knows which ports to queue), then atomically replace the - // userspace lookup the handler reads to resolve a packet's owner. + // Config consumer: each services-list refresh produces a port→target + // map. We diff vs the previous, push the diff to the ipset (so the + // kernel knows which ports to queue), then atomically replace our + // userspace port→target lookup that the handler reads. { - let trigger_owners = trigger_owners.clone(); + let port_to_target = port_to_target.clone(); tokio::spawn(async move { - consume_config(config_rx, trigger_owners).await; + consume_config(config_rx, port_to_target).await; }); } @@ -78,7 +80,7 @@ pub fn spawn_listener( let ctx = ListenerCtx { grpc, cache, - trigger_owners, + port_to_target, triggers_state, semaphore: Arc::new(Semaphore::new(HANDLER_CONCURRENCY)), }; @@ -86,8 +88,8 @@ pub fn spawn_listener( } async fn consume_config( - mut config_rx: UnboundedReceiver, - trigger_owners: Arc>, + mut config_rx: UnboundedReceiver>>, + port_to_target: Arc>>>, ) { let mut current_ports: HashSet = HashSet::new(); while let Some(new_map) = config_rx.recv().await { @@ -95,7 +97,7 @@ async fn consume_config( rules::apply_ports_diff(¤t_ports, &new_ports); // Swap the lookup. Sync RwLock; write is brief, never held across // an `.await`. - *trigger_owners.write().unwrap() = new_map; + *port_to_target.write().unwrap() = new_map; current_ports = new_ports; } } diff --git a/members/nullnet-client/src/nfqueue/parse.rs b/members/nullnet-client/src/nfqueue/parse.rs index f618d5f..cd9aedf 100644 --- a/members/nullnet-client/src/nfqueue/parse.rs +++ b/members/nullnet-client/src/nfqueue/parse.rs @@ -2,25 +2,12 @@ use etherparse::{LaxPacketHeaders, NetHeaders, TransportHeader}; use std::net::Ipv4Addr; /// NFQUEUE delivers L3 (no Ethernet) IPv4 packets to userspace. Extract the -/// source IP and the L4 destination port for TCP and UDP. Returns `None` for -/// non-IPv4, non-TCP/UDP, fragmented, or malformed packets. -pub fn ipv4_src_and_dst_port(packet: &[u8]) -> Option<(Ipv4Addr, u16)> { - let headers = LaxPacketHeaders::from_ip(packet).ok()?; - let src_octets = match headers.net? { - NetHeaders::Ipv4(ipv4, _) => ipv4.source, - _ => return None, - }; - let dst_port = match headers.transport? { - TransportHeader::Tcp(tcp) => tcp.destination_port, - TransportHeader::Udp(udp) => udp.destination_port, - _ => return None, - }; - Some((Ipv4Addr::from(src_octets), dst_port)) -} - -/// Like [`ipv4_src_and_dst_port`] but also returns the destination IP. Used by -/// the egress listener, which classifies flows by destination (external vs -/// internal) rather than by port. +/// source IP, destination IP, and L4 destination port for TCP and UDP. +/// Returns `None` for non-IPv4, non-TCP/UDP, fragmented, or malformed +/// packets. The destination IP is what lets the backend-trigger listener +/// disambiguate two dependencies sharing a port (via their distinct +/// placeholder addresses, see `placeholder.rs`) and what the egress listener +/// uses to classify flows (external vs internal). pub fn ipv4_flow(packet: &[u8]) -> Option<(Ipv4Addr, Ipv4Addr, u16)> { let headers = LaxPacketHeaders::from_ip(packet).ok()?; let (src_octets, dst_octets) = match headers.net? { @@ -83,8 +70,8 @@ mod tests { 80, ); assert_eq!( - ipv4_src_and_dst_port(&pkt), - Some((Ipv4Addr::new(172, 17, 0, 5), 80)) + ipv4_flow(&pkt), + Some((Ipv4Addr::new(172, 17, 0, 5), Ipv4Addr::new(10, 0, 0, 1), 80)) ); } @@ -97,14 +84,14 @@ mod tests { 53, ); assert_eq!( - ipv4_src_and_dst_port(&pkt), - Some((Ipv4Addr::new(172, 17, 0, 6), 53)) + ipv4_flow(&pkt), + Some((Ipv4Addr::new(172, 17, 0, 6), Ipv4Addr::new(10, 0, 0, 2), 53)) ); } #[test] fn rejects_too_short() { - assert!(ipv4_src_and_dst_port(&[0x45; 10]).is_none()); + assert!(ipv4_flow(&[0x45; 10]).is_none()); } #[test] @@ -114,7 +101,7 @@ mod tests { let mut buf = vec![0u8; 40]; buf[0] = 0x60; buf[6] = 59; // No-Next-Header so etherparse stops cleanly - assert!(ipv4_src_and_dst_port(&buf).is_none()); + assert!(ipv4_flow(&buf).is_none()); } #[test] @@ -124,6 +111,6 @@ mod tests { buf[0] = 0x45; buf[2..4].copy_from_slice(&total_len); buf[9] = 1; // ICMP — not TCP/UDP - assert!(ipv4_src_and_dst_port(&buf).is_none()); + assert!(ipv4_flow(&buf).is_none()); } } diff --git a/members/nullnet-client/src/placeholder.rs b/members/nullnet-client/src/placeholder.rs new file mode 100644 index 0000000..02f4a36 --- /dev/null +++ b/members/nullnet-client/src/placeholder.rs @@ -0,0 +1,199 @@ +use crate::host_mappings; +use nullnet_liberror::{Error, ErrorHandler, Location, location}; +use std::net::Ipv4Addr; + +/// Default placeholder range: 203.0.113.0/24 (RFC 5737 TEST-NET-3) — +/// reserved for documentation, never assigned to a real host, never on-link +/// for a container's own subnet. A container's routing table has no direct +/// route for it, so any address in this range falls through to the +/// container's default gateway — the path NFQUEUE's `mangle PREROUTING` +/// hook sits on — instead of being resolved on-link. +const DEFAULT_CIDR_BASE: [u8; 3] = [203, 0, 113]; + +/// Env override for the placeholder range, in case an operator's environment +/// does something unusual with the default block. Only the /24's network +/// address matters here; the last octet is always derived per-name. +const CIDR_ENV_VAR: &str = "TRIGGER_PLACEHOLDER_CIDR"; + +/// Deterministic placeholder address for `name`, distinct per name so two +/// backend-trigger dependencies sharing a port (see `triggers::TriggersState`'s +/// `dst_ip`-widened key) still disambiguate by destination address alone. +/// Stable across restarts — this is a pure function of `name`, not a stored +/// allocation, so the client and (once it knows the same name) the +/// control-channel setup/teardown paths always agree on it independently. +/// The last octet comes from `nullnet_grpc_lib::last_octet_for`, shared with +/// nullnet-server so its config-validation can detect a collision before +/// this ever runs. +pub(crate) fn ip_for(name: &str) -> Ipv4Addr { + let [a, b, c] = cidr_base(); + Ipv4Addr::new(a, b, c, nullnet_grpc_lib::last_octet_for(name)) +} + +/// The placeholder block as a CIDR string (e.g. `"203.0.113.0/24"`), for +/// callers that need to treat it as a single internal-ish destination range +/// rather than resolve individual names — see `commands::egress`'s +/// `INTERNAL_RANGES`: a backend-trigger placeholder address is synthetic and +/// never real internet traffic, so it must never be classified as an egress +/// candidate, regardless of which CIDR base is configured. +pub(crate) fn placeholder_cidr() -> String { + let [a, b, c] = cidr_base(); + format!("{a}.{b}.{c}.0/24") +} + +fn cidr_base() -> [u8; 3] { + match std::env::var(CIDR_ENV_VAR) { + Ok(cidr) => parse_cidr_base(&cidr).unwrap_or_else(|| { + eprintln!( + "[placeholder] invalid {CIDR_ENV_VAR} '{cidr}'; falling back to default {DEFAULT_CIDR_BASE:?}" + ); + DEFAULT_CIDR_BASE + }), + Err(_) => DEFAULT_CIDR_BASE, + } +} + +fn parse_cidr_base(cidr: &str) -> Option<[u8; 3]> { + let ip_part = cidr.split('/').next()?; + let octets: Vec = ip_part.split('.').filter_map(|o| o.parse().ok()).collect(); + match octets.as_slice() { + [a, b, c, ..] => Some([*a, *b, *c]), + _ => None, + } +} + +/// Write `name -> ip_for(name)` into `container`'s own `/etc/hosts`, before +/// any packet has been observed on the trigger port it's associated with. +/// Idempotent — safe to call on every declare-services reconcile pass. Edited +/// from the host side (see `host_mappings::edit_container_hosts`) so a paused +/// container is seeded just as well as a running one. +pub(crate) fn seed_placeholder(container: &str, name: &str) -> Result<(), Error> { + warn_if_no_default_route(container); + let ip = ip_for(name); + let entry = format!("{ip} {name} {}", host_mappings::HOSTS_MARKER); + host_mappings::edit_container_hosts(container, |current| { + host_mappings::upsert_hosts_entry(current, name, &entry) + }) + .handle_err(location!())?; + Ok(()) +} + +/// Containers already warned about missing a default route, so the warning +/// prints once per bad spell rather than on every declare-services reconcile +/// pass — and clears once the container recovers, so a later regression (a +/// recreate that drops the primary interface again) warns again instead of +/// staying silent forever. +static WARNED_NO_DEFAULT_ROUTE: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); + +/// Best-effort check: a container with no default route can never reach an +/// off-link placeholder address (see this module's doc comment on why the +/// placeholder block is deliberately never on-link) — every backend-trigger +/// dial from it is doomed before nullnet is even involved, and the only +/// visible symptom would otherwise be a bare `ENETUNREACH`/`EHOSTUNREACH` in +/// the initiator's own app three layers away, indistinguishable from any +/// other network hiccup. Log loudly instead. A `docker exec` failure here +/// (container gone, docker hiccup, `ip` missing in a minimal image) is not +/// itself something to report — this is diagnostic, not load-bearing. +fn warn_if_no_default_route(container: &str) { + let Ok(out) = std::process::Command::new("docker") + .args(["exec", container, "ip", "route", "show", "default"]) + .output() + else { + return; + }; + if !out.status.success() { + return; + } + let mut warned = WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()); + if out.stdout.is_empty() { + if warned.insert(container.to_string()) { + eprintln!( + "[placeholder] container '{container}' has no default route — off-link \ + trigger traffic will fail; check its Docker network attachment \ + (missing eth0, exhausted address pool, etc.) before retrying" + ); + } + } else { + warned.remove(container); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deterministic_across_calls() { + assert_eq!(ip_for("redis"), ip_for("redis")); + } + + #[test] + fn distinct_names_get_distinct_addresses() { + assert_ne!(ip_for("auth"), ip_for("billing")); + } + + #[test] + fn stays_within_default_range() { + let ip = ip_for("redis"); + let [a, b, c, d] = ip.octets(); + assert_eq!([a, b, c], DEFAULT_CIDR_BASE); + assert!((1..=254).contains(&d)); + } + + #[test] + fn placeholder_cidr_matches_default_base() { + assert_eq!(placeholder_cidr(), "203.0.113.0/24"); + } + + #[test] + fn parses_cidr_base_from_env_value() { + assert_eq!(parse_cidr_base("198.51.100.0/24"), Some([198, 51, 100])); + assert_eq!(parse_cidr_base("198.51.100.5"), Some([198, 51, 100])); + assert_eq!(parse_cidr_base("not-a-cidr"), None); + } + + #[test] + fn no_default_route_warning_dedups_then_clears_on_recovery() { + let mut warned = WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()); + warned.clear(); + drop(warned); + + // First sighting of a bad container: newly inserted, warns. + assert!( + WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert("flaky".to_string()) + ); + // Same container again while still bad: already present, would not + // re-warn — this is what keeps every reconcile pass from spamming. + assert!( + !WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert("flaky".to_string()) + ); + // Recovery clears it, so a later regression warns again instead of + // staying silent forever. + WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove("flaky"); + assert!( + WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert("flaky".to_string()) + ); + + WARNED_NO_DEFAULT_ROUTE + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clear(); + } +} diff --git a/members/nullnet-client/src/triggers.rs b/members/nullnet-client/src/triggers.rs index bf28c83..e558ee0 100644 --- a/members/nullnet-client/src/triggers.rs +++ b/members/nullnet-client/src/triggers.rs @@ -16,11 +16,21 @@ const PENDING_TIMEOUT: Duration = Duration::from_secs(10); /// sentinel never collides with them in the shared `TriggersState`. pub const EGRESS_TRIGGER_PORT: u16 = 0; -/// Per-(initiator_container, port) lifecycle. `container_ip` is the bridge IP -/// the NFQUEUE listener observed when the trigger fired; it is carried through -/// to `Active` so DNAT install/remove can match the right `-s` source. Legacy -/// callers that don't know the container IP pass `Ipv4Addr::UNSPECIFIED`, -/// which the dnat module treats as "no source filter". +/// `dst_ip` sentinel for egress triggers, which aren't destination-scoped at +/// all (steering matches every destination for the initiator container). +/// Mirrors `Ipv4Addr::UNSPECIFIED`'s existing "no filter" meaning elsewhere +/// in this codebase (e.g. `dnat`'s `container_ip`). +pub const EGRESS_DST_IP: Ipv4Addr = Ipv4Addr::UNSPECIFIED; + +/// Per-(initiator_container, dst_ip, port) lifecycle. `dst_ip` disambiguates +/// two backend-trigger dependencies that share a port (e.g. two plain-HTTPS +/// deps, both 443) — each target name gets its own deterministic placeholder +/// address (see `placeholder.rs`), so the destination the packet was +/// actually addressed to is enough to tell them apart. `container_ip` is the +/// bridge IP the NFQUEUE listener observed when the trigger fired; it is +/// carried through to `Active` so DNAT install/remove can match the right +/// `-s` source. Legacy callers that don't know the container IP pass +/// `Ipv4Addr::UNSPECIFIED`, which the dnat module treats as "no source filter". pub enum Lifecycle { Pending { since: Instant, @@ -45,18 +55,23 @@ pub enum TriggerState { Active, } +/// Key: `(initiator_container, dst_ip, port)`. `dst_ip` is `EGRESS_DST_IP` +/// for egress entries (not destination-scoped) or the observed/placeholder +/// destination for backend-trigger entries. +type Key = (String, Ipv4Addr, u16); + #[derive(Default)] pub struct TriggersState { - by_key: Mutex>, + by_key: Mutex>, } impl TriggersState { - /// Snapshot the state for `(container, port)`. The lock is dropped before - /// returning so callers can `.await` on the returned `Notify` without - /// holding it. - pub fn state(&self, container: &str, port: u16) -> TriggerState { + /// Snapshot the state for `(container, dst_ip, port)`. The lock is dropped + /// before returning so callers can `.await` on the returned `Notify` + /// without holding it. + pub fn state(&self, container: &str, dst_ip: Ipv4Addr, port: u16) -> TriggerState { let by_key = self.by_key.lock().unwrap(); - match by_key.get(&(container.to_string(), port)) { + match by_key.get(&(container.to_string(), dst_ip, port)) { Some(Lifecycle::Active { .. }) => TriggerState::Active, Some(Lifecycle::Pending { since, notify, .. }) if since.elapsed() < PENDING_TIMEOUT => { TriggerState::Pending(notify.clone()) @@ -68,9 +83,15 @@ impl TriggersState { /// Insert (or refresh) a `Pending` entry and return the `Notify` the /// caller awaits. If an in-flight `Pending` already exists, its `Notify` /// is returned so concurrent fires share one wake-up. - pub fn mark_pending(&self, container: &str, port: u16, container_ip: Ipv4Addr) -> Arc { + pub fn mark_pending( + &self, + container: &str, + dst_ip: Ipv4Addr, + port: u16, + container_ip: Ipv4Addr, + ) -> Arc { let mut by_key = self.by_key.lock().unwrap(); - let key = (container.to_string(), port); + let key = (container.to_string(), dst_ip, port); if let Some(Lifecycle::Pending { since, notify, .. }) = by_key.get(&key) && since.elapsed() < PENDING_TIMEOUT { @@ -94,9 +115,9 @@ impl TriggersState { /// (which wakes on `mark_active`'s `notify_waiters`) finds the DNAT rule /// live by the time it traverses `nat PREROUTING`. Returns /// `Ipv4Addr::UNSPECIFIED` if no entry exists. - pub fn peek_container_ip(&self, container: &str, port: u16) -> Ipv4Addr { + pub fn peek_container_ip(&self, container: &str, dst_ip: Ipv4Addr, port: u16) -> Ipv4Addr { let by_key = self.by_key.lock().unwrap(); - match by_key.get(&(container.to_string(), port)) { + match by_key.get(&(container.to_string(), dst_ip, port)) { Some(Lifecycle::Pending { container_ip, .. }) | Some(Lifecycle::Active { container_ip, .. }) => *container_ip, None => Ipv4Addr::UNSPECIFIED, @@ -110,13 +131,14 @@ impl TriggersState { pub fn mark_active( &self, container: &str, + dst_ip: Ipv4Addr, port: u16, vxlan_id: u32, overlay_ip: Ipv4Addr, container_ip: Ipv4Addr, ) { let mut by_key = self.by_key.lock().unwrap(); - let key = (container.to_string(), port); + let key = (container.to_string(), dst_ip, port); let notify = match by_key.remove(&key) { Some(Lifecycle::Pending { notify, .. } | Lifecycle::Active { notify, .. }) => notify, None => Arc::new(Notify::new()), @@ -136,21 +158,43 @@ impl TriggersState { notify.notify_waiters(); } - /// Drop the entry so the next observed packet on this `(container, port)` retriggers. - pub fn forget(&self, container: &str, port: u16) { + /// Drop the entry so the next observed packet on this + /// `(container, dst_ip, port)` retriggers. + pub fn forget(&self, container: &str, dst_ip: Ipv4Addr, port: u16) { self.by_key .lock() .unwrap() - .remove(&(container.to_string(), port)); + .remove(&(container.to_string(), dst_ip, port)); + } + + /// Drop every entry for `container`, regardless of `dst_ip`/`port`. + /// Called when Docker reports the container restarted or was recreated: + /// its name is stable across a recreate, but the sandbox (and therefore + /// the source IP any installed DNAT's `-s` matched on) is not. A stale + /// `Active` entry would otherwise short-circuit `decide_verdict` straight + /// to `Accept` for the new instance's packets — skipping re-trigger + /// entirely — even though the tunnel it points at no longer matches + /// anything real. This is a coarser hammer than `forget`/`remove_by_vxlan` + /// (no attempt to also tear down whatever DNAT the stale entry implied; + /// the container-side network is being rebuilt anyway), but it's what + /// restores self-healing without a manual `nullnet-client` restart. + pub fn forget_container(&self, container: &str) { + self.by_key + .lock() + .unwrap() + .retain(|(c, ..), _| c != container); } /// Find the `Active` entry for `vxlan_id`, remove it, and return - /// `(container, port, overlay_ip, container_ip)` so the caller can tear - /// down DNAT with the matching `-s`. - pub fn remove_by_vxlan(&self, vxlan_id: u32) -> Option<(String, u16, Ipv4Addr, Ipv4Addr)> { + /// `(container, dst_ip, port, overlay_ip, container_ip)` so the caller + /// can tear down DNAT with the matching `-s`/`-d`. + pub fn remove_by_vxlan( + &self, + vxlan_id: u32, + ) -> Option<(String, Ipv4Addr, u16, Ipv4Addr, Ipv4Addr)> { let mut by_key = self.by_key.lock().unwrap(); - let key = by_key.iter().find_map(|((c, p), lc)| match lc { - Lifecycle::Active { vxlan_id: v, .. } if *v == vxlan_id => Some((c.clone(), *p)), + let key = by_key.iter().find_map(|((c, d, p), lc)| match lc { + Lifecycle::Active { vxlan_id: v, .. } if *v == vxlan_id => Some((c.clone(), *d, *p)), _ => None, })?; // The lock is held across the `iter().find_map` and the `remove` @@ -162,7 +206,7 @@ impl TriggersState { overlay_ip, container_ip, .. - }) => Some((key.0, key.1, overlay_ip, container_ip)), + }) => Some((key.0, key.1, key.2, overlay_ip, container_ip)), Some(Lifecycle::Pending { .. }) | None => { unreachable!("find_map matched Active for {key:?}; lock held across remove") } @@ -179,23 +223,27 @@ mod tests { const IP: Ipv4Addr = Ipv4Addr::new(172, 17, 0, 5); const OVERLAY: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); + /// Stand-in placeholder/destination address — same role a real deployment + /// gets from `placeholder::ip_for(target_name)`. + const DST: Ipv4Addr = Ipv4Addr::new(203, 0, 113, 7); + const DST2: Ipv4Addr = Ipv4Addr::new(203, 0, 113, 42); #[tokio::test] async fn pending_then_active_wakes_waiter() { let state = Arc::new(TriggersState::default()); - let notify = state.mark_pending("c1", 80, IP); - assert_eq!(state.peek_container_ip("c1", 80), IP); + let notify = state.mark_pending("c1", DST, 80, IP); + assert_eq!(state.peek_container_ip("c1", DST, 80), IP); let state_clone = state.clone(); let waiter = tokio::spawn(async move { notify.notified().await; - matches!(state_clone.state("c1", 80), TriggerState::Active) + matches!(state_clone.state("c1", DST, 80), TriggerState::Active) }); tokio::time::sleep(Duration::from_millis(20)).await; // The control-channel pattern: peek for install, install DNAT, then // mark_active to wake. peek above already returned IP for the install. - state.mark_active("c1", 80, 42, OVERLAY, IP); + state.mark_active("c1", DST, 80, 42, OVERLAY, IP); let ok = timeout(Duration::from_secs(1), waiter) .await @@ -213,17 +261,17 @@ mod tests { // synchronous state recheck after `.enable()` — it must observe // the Active state set by mark_active and short-circuit the await. let state = Arc::new(TriggersState::default()); - let notify = state.mark_pending("c1", 80, IP); + let notify = state.mark_pending("c1", DST, 80, IP); // mark_active fires BEFORE the listener registers a waiter — this // is exactly the lost-wake race. - state.mark_active("c1", 80, 42, OVERLAY, IP); + state.mark_active("c1", DST, 80, 42, OVERLAY, IP); // The listener's race-protected pattern. let notified = notify.notified(); tokio::pin!(notified); let trip_via_enable = notified.as_mut().enable(); - let trip_via_state = matches!(state.state("c1", 80), TriggerState::Active); + let trip_via_state = matches!(state.state("c1", DST, 80), TriggerState::Active); assert!( trip_via_enable || trip_via_state, "race-fix must observe Active even when mark_active fired before enable" @@ -249,15 +297,18 @@ mod tests { // during `backend_trigger`'s round-trip — after enable, before the // listener resumes awaiting. let state = Arc::new(TriggersState::default()); - let notify = state.mark_pending("c1", 80, IP); + let notify = state.mark_pending("c1", DST, 80, IP); let notified = notify.notified(); tokio::pin!(notified); notified.as_mut().enable(); - assert!(matches!(state.state("c1", 80), TriggerState::Pending(_))); + assert!(matches!( + state.state("c1", DST, 80), + TriggerState::Pending(_) + )); // mark_active fires AFTER enable but before await. - state.mark_active("c1", 80, 42, OVERLAY, IP); + state.mark_active("c1", DST, 80, 42, OVERLAY, IP); let result = timeout(Duration::from_millis(50), notified).await; assert!( @@ -269,44 +320,98 @@ mod tests { #[tokio::test] async fn concurrent_pending_share_notify() { let state = Arc::new(TriggersState::default()); - let n1 = state.mark_pending("c1", 80, IP); - let n2 = state.mark_pending("c1", 80, IP); + let n1 = state.mark_pending("c1", DST, 80, IP); + let n2 = state.mark_pending("c1", DST, 80, IP); assert!(Arc::ptr_eq(&n1, &n2), "same key must reuse the Notify"); } #[tokio::test] async fn distinct_containers_are_independent() { let state = TriggersState::default(); - let _ = state.mark_pending("c1", 80, IP); - let _ = state.mark_pending("c2", 80, Ipv4Addr::new(172, 17, 0, 6)); - assert!(matches!(state.state("c1", 80), TriggerState::Pending(_))); - assert!(matches!(state.state("c2", 80), TriggerState::Pending(_))); - state.mark_active("c1", 80, 7, OVERLAY, IP); - assert!(matches!(state.state("c1", 80), TriggerState::Active)); - assert!(matches!(state.state("c2", 80), TriggerState::Pending(_))); + let _ = state.mark_pending("c1", DST, 80, IP); + let _ = state.mark_pending("c2", DST, 80, Ipv4Addr::new(172, 17, 0, 6)); + assert!(matches!( + state.state("c1", DST, 80), + TriggerState::Pending(_) + )); + assert!(matches!( + state.state("c2", DST, 80), + TriggerState::Pending(_) + )); + state.mark_active("c1", DST, 80, 7, OVERLAY, IP); + assert!(matches!(state.state("c1", DST, 80), TriggerState::Active)); + assert!(matches!( + state.state("c2", DST, 80), + TriggerState::Pending(_) + )); + } + + #[tokio::test] + async fn same_port_different_dst_ip_are_independent() { + // Two backend-trigger dependencies sharing a port (e.g. two + // plain-HTTPS deps, both 443) from the same initiator container — + // disambiguated purely by their distinct placeholder/destination IPs. + let state = TriggersState::default(); + let _ = state.mark_pending("c1", DST, 443, IP); + let _ = state.mark_pending("c1", DST2, 443, IP); + assert!(matches!( + state.state("c1", DST, 443), + TriggerState::Pending(_) + )); + assert!(matches!( + state.state("c1", DST2, 443), + TriggerState::Pending(_) + )); + state.mark_active("c1", DST, 443, 7, OVERLAY, IP); + assert!(matches!(state.state("c1", DST, 443), TriggerState::Active)); + assert!(matches!( + state.state("c1", DST2, 443), + TriggerState::Pending(_) + )); } #[tokio::test] - async fn remove_by_vxlan_returns_container_port_and_ip() { + async fn remove_by_vxlan_returns_container_dst_ip_port_and_ip() { let state = TriggersState::default(); - let _ = state.mark_pending("c1", 80, IP); - state.mark_active("c1", 80, 42, OVERLAY, IP); + let _ = state.mark_pending("c1", DST, 80, IP); + state.mark_active("c1", DST, 80, 42, OVERLAY, IP); let removed = state.remove_by_vxlan(42).expect("entry should exist"); - assert_eq!(removed, ("c1".to_string(), 80, OVERLAY, IP)); - assert!(matches!(state.state("c1", 80), TriggerState::Fresh)); + assert_eq!(removed, ("c1".to_string(), DST, 80, OVERLAY, IP)); + assert!(matches!(state.state("c1", DST, 80), TriggerState::Fresh)); } #[tokio::test] async fn peek_returns_unspecified_when_absent() { let state = TriggersState::default(); - assert_eq!(state.peek_container_ip("c1", 80), Ipv4Addr::UNSPECIFIED); + assert_eq!( + state.peek_container_ip("c1", DST, 80), + Ipv4Addr::UNSPECIFIED + ); } #[tokio::test] async fn forget_drops_entry() { let state = TriggersState::default(); - let _ = state.mark_pending("c1", 80, IP); - state.forget("c1", 80); - assert!(matches!(state.state("c1", 80), TriggerState::Fresh)); + let _ = state.mark_pending("c1", DST, 80, IP); + state.forget("c1", DST, 80); + assert!(matches!(state.state("c1", DST, 80), TriggerState::Fresh)); + } + + #[tokio::test] + async fn forget_container_drops_active_and_pending_across_ports_and_dst_ips() { + let state = TriggersState::default(); + state.mark_active("c1", DST, 80, 7, OVERLAY, IP); + let _ = state.mark_pending("c1", DST2, 443, IP); + let _ = state.mark_pending("c2", DST, 80, IP); + + state.forget_container("c1"); + + assert!(matches!(state.state("c1", DST, 80), TriggerState::Fresh)); + assert!(matches!(state.state("c1", DST2, 443), TriggerState::Fresh)); + // A different container's state is untouched. + assert!(matches!( + state.state("c2", DST, 80), + TriggerState::Pending(_) + )); } } diff --git a/members/nullnet-grpc-lib/proto/nullnet_grpc.proto b/members/nullnet-grpc-lib/proto/nullnet_grpc.proto index 133a265..0bd5c54 100644 --- a/members/nullnet-grpc-lib/proto/nullnet_grpc.proto +++ b/members/nullnet-grpc-lib/proto/nullnet_grpc.proto @@ -232,20 +232,36 @@ message Listener { } // Response to ServicesList: per-declared-service, the set of trigger ports -// the client should observe via eBPF. When traffic is observed on one of -// these ports, the client fires BackendTrigger(service_name, port). +// the client should observe via NFQUEUE. When traffic is observed on one of +// these ports, the client fires BackendTrigger(service_name, port, target_name). message ServicesListResponse { repeated ServiceTrigger service_triggers = 1; } message ServiceTrigger { string service_name = 1; - repeated uint32 ports = 2; + reserved 2; + reserved "ports"; + repeated TriggerPort trigger_ports = 3; // Real container names hosting this service on the receiving node — the same // string space as the client's bridge-IP cache and as Container.real_name. - // Scopes the trigger to its own replicas; empty means a server that predates - // this field, and the client then falls back to matching any container. - repeated string containers = 3; + // Scopes both the port and the target-chain lookups to this service's own + // replicas: `target_name` (per TriggerPort) should be pre-seeded into each + // one, and a container not in this list never gets attributed to this + // trigger even if it happens to share the port. Empty means a server that + // predates this field, and the client falls back to matching any container. + repeated string containers = 4; +} + +// One trigger port this service's initiator container should be observed on. +message TriggerPort { + uint32 port = 1; + // chain[0] for this port's dependency chain — the literal name the + // initiator resolves (e.g. a bare Docker container name). Lets the client + // pre-seed a placeholder `/etc/hosts` entry for it before any packet is + // observed, so a bare name (not just a pre-provisioned DNS alias) produces + // a real first packet to trigger on. + string target_name = 2; } message HostMapping { @@ -303,6 +319,11 @@ message BackendTriggerRequest { // sender_ip. With docker, this disambiguates the replica when multiple // replicas of the same service live on the same host. string initiator_container = 3; + // The target_name (see TriggerPort) the observed packet's destination + // placeholder IP resolved back to. Lets the server pick the right chain + // when two of this service's triggers share the same port, instead of + // relying on port alone. + string target_name = 4; } // Egress trigger — fired by the client on the first NEW flow from a registered diff --git a/members/nullnet-grpc-lib/src/lib.rs b/members/nullnet-grpc-lib/src/lib.rs index a0f0d0b..afe2602 100644 --- a/members/nullnet-grpc-lib/src/lib.rs +++ b/members/nullnet-grpc-lib/src/lib.rs @@ -1,4 +1,5 @@ mod control_tls_verifier; +mod placeholder; mod proto; use crate::control_tls_verifier::PinnedCa; @@ -8,6 +9,7 @@ use crate::nullnet_grpc::{ EgressPolicyCheck, EgressTriggerRequest, Empty, IngressPolicyCheck, MsgId, NetMessage, NetType, PortMappingBundle, ProxyRequest, ServiceReport, ServicesListResponse, Upstream, }; +pub use placeholder::last_octet_for; pub use proto::*; use std::path::Path; use tokio::sync::mpsc; @@ -114,6 +116,7 @@ impl NullnetGrpcInterface { service_name: String, port: u32, initiator_container: String, + target_name: String, ) -> Result<(), String> { self.client .clone() @@ -121,6 +124,7 @@ impl NullnetGrpcInterface { service_name, port, initiator_container, + target_name, })) .await .map(|_| ()) diff --git a/members/nullnet-grpc-lib/src/placeholder.rs b/members/nullnet-grpc-lib/src/placeholder.rs new file mode 100644 index 0000000..6016cea --- /dev/null +++ b/members/nullnet-grpc-lib/src/placeholder.rs @@ -0,0 +1,44 @@ +/// Deterministic last octet (1..=254, avoiding the network/broadcast +/// addresses) of a backend-trigger dependency's placeholder address, derived +/// from its name. Shared between nullnet-client (which builds the actual +/// `/etc/hosts` address from it — see its `placeholder.rs`) and +/// nullnet-server (which uses it at config-validation time to catch two +/// distinct chain[0] names on the same port that would land on the same +/// address) so both sides can never disagree on the mapping. +/// +/// Not collision-free — a non-cryptographic hash folded into 254 buckets +/// will alias distinct names well before 254 of them are in play. Callers +/// are responsible for deciding whether a collision matters to them. +pub fn last_octet_for(name: &str) -> u8 { + let hash = fnv1a(name.as_bytes()); + u8::try_from(hash % 254).unwrap_or(0) + 1 +} + +fn fnv1a(bytes: &[u8]) -> u32 { + let mut hash: u32 = 0x811c_9dc5; + for &b in bytes { + hash ^= u32::from(b); + hash = hash.wrapping_mul(0x0100_0193); + } + hash +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deterministic_across_calls() { + assert_eq!(last_octet_for("redis"), last_octet_for("redis")); + } + + #[test] + fn distinct_names_can_get_distinct_octets() { + assert_ne!(last_octet_for("auth"), last_octet_for("billing")); + } + + #[test] + fn stays_within_range() { + assert!((1..=254).contains(&last_octet_for("redis"))); + } +} diff --git a/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs b/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs index f40dc71..a49591e 100644 --- a/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs +++ b/members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs @@ -222,26 +222,42 @@ pub struct Listener { pub path: ::prost::alloc::string::String, } /// Response to ServicesList: per-declared-service, the set of trigger ports -/// the client should observe via eBPF. When traffic is observed on one of -/// these ports, the client fires BackendTrigger(service_name, port). +/// the client should observe via NFQUEUE. When traffic is observed on one of +/// these ports, the client fires BackendTrigger(service_name, port, target_name). #[derive(Clone, PartialEq, ::prost::Message)] pub struct ServicesListResponse { #[prost(message, repeated, tag = "1")] pub service_triggers: ::prost::alloc::vec::Vec, } -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +#[derive(Clone, PartialEq, ::prost::Message)] pub struct ServiceTrigger { #[prost(string, tag = "1")] pub service_name: ::prost::alloc::string::String, - #[prost(uint32, repeated, tag = "2")] - pub ports: ::prost::alloc::vec::Vec, + #[prost(message, repeated, tag = "3")] + pub trigger_ports: ::prost::alloc::vec::Vec, /// Real container names hosting this service on the receiving node — the same /// string space as the client's bridge-IP cache and as Container.real_name. - /// Scopes the trigger to its own replicas; empty means a server that predates - /// this field, and the client then falls back to matching any container. - #[prost(string, repeated, tag = "3")] + /// Scopes both the port and the target-chain lookups to this service's own + /// replicas: `target_name` (per TriggerPort) should be pre-seeded into each + /// one, and a container not in this list never gets attributed to this + /// trigger even if it happens to share the port. Empty means a server that + /// predates this field, and the client falls back to matching any container. + #[prost(string, repeated, tag = "4")] pub containers: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, } +/// One trigger port this service's initiator container should be observed on. +#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] +pub struct TriggerPort { + #[prost(uint32, tag = "1")] + pub port: u32, + /// chain\[0\] for this port's dependency chain — the literal name the + /// initiator resolves (e.g. a bare Docker container name). Lets the client + /// pre-seed a placeholder `/etc/hosts` entry for it before any packet is + /// observed, so a bare name (not just a pre-provisioned DNS alias) produces + /// a real first packet to trigger on. + #[prost(string, tag = "2")] + pub target_name: ::prost::alloc::string::String, +} #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct HostMapping { #[prost(string, tag = "1")] @@ -299,6 +315,12 @@ pub struct BackendTriggerRequest { /// replicas of the same service live on the same host. #[prost(string, tag = "3")] pub initiator_container: ::prost::alloc::string::String, + /// The target_name (see TriggerPort) the observed packet's destination + /// placeholder IP resolved back to. Lets the server pick the right chain + /// when two of this service's triggers share the same port, instead of + /// relying on port alone. + #[prost(string, tag = "4")] + pub target_name: ::prost::alloc::string::String, } /// Egress trigger — fired by the client on the first NEW flow from a registered /// service to an external (non-nullnet) destination. The server builds (or reuses) diff --git a/members/nullnet-server/src/http_server/services.rs b/members/nullnet-server/src/http_server/services.rs index f058545..c722784 100644 --- a/members/nullnet-server/src/http_server/services.rs +++ b/members/nullnet-server/src/http_server/services.rs @@ -24,7 +24,7 @@ struct ServiceJson { registered: bool, replicas: Vec, proxy_dependencies: Vec>, - triggers: HashMap>, + triggers: HashMap>>, #[serde(skip_serializing_if = "Option::is_none")] timeout_secs: Option, #[serde(skip_serializing_if = "Option::is_none")] diff --git a/members/nullnet-server/src/nullnet_grpc_impl.rs b/members/nullnet-server/src/nullnet_grpc_impl.rs index 21856bc..e21af39 100644 --- a/members/nullnet-server/src/nullnet_grpc_impl.rs +++ b/members/nullnet-server/src/nullnet_grpc_impl.rs @@ -21,7 +21,8 @@ use nullnet_grpc_lib::nullnet_grpc::{ AgentEvent, BackendTriggerRequest, CertBundle, EgressDestinationReport, EgressPolicyCheck, EgressPolicyVerdict, EgressTriggerRequest, Empty, IngressPolicyCheck, IngressPolicyVerdict, MsgId, Net, NetMessage, NetType, PortMapping, PortMappingBundle, ProxyRequest, ServiceReport, - ServiceTrigger, ServicesListResponse, Upstream, agent_event::Event as AgentEventKind, + ServiceTrigger, ServicesListResponse, TriggerPort, Upstream, + agent_event::Event as AgentEventKind, }; use nullnet_liberror::{Error, ErrorHandler, Location, location}; use std::collections::{HashMap, HashSet}; @@ -86,13 +87,16 @@ fn build_port_mapping_bundle(stacks: &StackMap) -> PortMappingBundle { /// Build the trigger config for one node: the triggers of the services it /// declared as hosting, each carrying the real container names hosting that -/// service *there*. +/// service *there* and, per port, every dependency chain's target name. /// /// The client's NFQUEUE watch is a destination-port match, so a watched port -/// catches every container on the node. The container list is what lets it tell -/// the declaring service's own traffic from a co-located container that merely -/// talks to the same port — the latter used to be attributed to the declaring -/// service, rejected by `handle_backend_trigger`, and dropped. +/// catches every container on the node. The container list is what lets the +/// client tell the declaring service's own traffic from a co-located +/// container that merely talks to the same port — the latter used to be +/// attributed to the declaring service, rejected by `handle_backend_trigger`, +/// and dropped. Once attributed, more than one chain can still share that +/// port (see `TriggerPort`), disambiguated by the packet's destination +/// against each candidate's own placeholder address. /// /// A service the node hosts as a bare process contributes no containers and is /// skipped: its trigger can never fire (the NFQUEUE path passes host traffic @@ -125,12 +129,31 @@ pub(crate) fn build_service_triggers( if triggers.is_empty() { return None; } - let mut ports: Vec = triggers.keys().map(|p| u32::from(*p)).collect(); - ports.sort_unstable(); + // More than one chain can share a port; emit one TriggerPort per + // chain so the client learns every target_name for that port and + // can seed a placeholder for (and later disambiguate) each one + // independently. + let mut trigger_ports: Vec = triggers + .iter() + .flat_map(|(port, chains)| { + chains.iter().filter_map(move |chain| { + // chain[0] is what the initiator actually resolves; a + // trigger with an empty chain can't be pre-seeded + // (nothing to target) and can't build a chain either, + // so it's already inert — skip it here too. + let target_name = chain.first()?.clone(); + Some(TriggerPort { + port: u32::from(*port), + target_name, + }) + }) + }) + .collect(); + trigger_ports.sort_unstable_by_key(|tp| tp.port); containers.sort_unstable(); Some(ServiceTrigger { service_name: name.to_string(), - ports, + trigger_ports, containers: containers.into_iter().map(ToString::to_string).collect(), }) }) @@ -663,6 +686,7 @@ impl NullnetGrpcImpl { service_ip: IpAddr, service_docker: Option<&str>, port: u16, + target_name: &str, ) -> Result>, Error> { let guard = self.services.read().await; let stack_map = guard @@ -681,6 +705,7 @@ impl NullnetGrpcImpl { service_ip, service_docker, port, + target_name, stack_map, ) else { return Ok(None); @@ -737,8 +762,14 @@ impl NullnetGrpcImpl { } else { Some(req.initiator_container) }; - self.handle_backend_trigger(&req.service_name, port, sender_ip, container.as_deref()) - .await?; + self.handle_backend_trigger( + &req.service_name, + port, + sender_ip, + container.as_deref(), + &req.target_name, + ) + .await?; Ok(Response::new(Empty {})) } @@ -748,6 +779,7 @@ impl NullnetGrpcImpl { port: u16, sender_ip: IpAddr, initiator_container: Option<&str>, + target_name: &str, ) -> Result<(), Error> { println!( "Received backend trigger for '{initiator_name}' (port {port}) from {sender_ip} (container: {})", @@ -789,13 +821,18 @@ impl NullnetGrpcImpl { .handle_err(location!())?; let initiator_ip = replica.ip(); let initiator_docker = replica.docker_container().map(String::from); + // `chain_for` picks the right chain among possibly several + // sharing this port, by matching `target_name` (chain[0]) — the + // literal name the client's placeholder resolved. A single + // chain on this port is used regardless (covers pre-disambiguation + // clients sending an empty target_name); with several sharing the + // port, no match means genuinely ambiguous, not a guess. let first_dep = reg - .triggers() - .get(&port) - .and_then(|chain| chain.first()) + .chain_for(port, target_name) + .and_then(|c| c.first()) .cloned(); println!( - "[trigger] triggers map for '{initiator_name}': {:?}; first_dep for port {port}: {first_dep:?}", + "[trigger] triggers map for '{initiator_name}': {:?}; target_name='{target_name}'; first_dep for port {port}: {first_dep:?}", reg.triggers() ); @@ -829,6 +866,7 @@ impl NullnetGrpcImpl { initiator_ip, initiator_docker.as_deref(), port, + target_name, ) .await } @@ -840,9 +878,17 @@ impl NullnetGrpcImpl { initiator_ip: IpAddr, initiator_docker: Option<&str>, port: u16, + target_name: &str, ) -> Result<(), Error> { let Some(mut chain) = self - .build_backend_dep_chain(stack, initiator_name, initiator_ip, initiator_docker, port) + .build_backend_dep_chain( + stack, + initiator_name, + initiator_ip, + initiator_docker, + port, + target_name, + ) .await? else { println!( diff --git a/members/nullnet-server/src/services/changes.rs b/members/nullnet-server/src/services/changes.rs index 909becf..a5f5a2d 100644 --- a/members/nullnet-server/src/services/changes.rs +++ b/members/nullnet-server/src/services/changes.rs @@ -416,31 +416,33 @@ fn collect_backend_chain_edges( let Some(triggers) = services.get(initiator_name).map(ServiceInfo::triggers) else { return edges; }; - for chain in triggers.values() { - if let Some(dep) = only_through - && !chain.iter().any(|d| d == dep) - { - continue; - } - let mut current_name = initiator_name.to_string(); - let mut current_ip = initiator_ip; - let mut current_docker: Option = initiator_docker.map(String::from); - for dep_name in chain { - let hop = emit_edge_and_probe_hop( - &mut edges, - ¤t_name, - current_ip, - current_docker.as_deref(), - dep_name, - services, - ); - match hop { - Some((ip, docker)) => { - current_name.clone_from(dep_name); - current_ip = ip; - current_docker = docker; + for chains in triggers.values() { + for chain in chains { + if let Some(dep) = only_through + && !chain.iter().any(|d| d == dep) + { + continue; + } + let mut current_name = initiator_name.to_string(); + let mut current_ip = initiator_ip; + let mut current_docker: Option = initiator_docker.map(String::from); + for dep_name in chain { + let hop = emit_edge_and_probe_hop( + &mut edges, + ¤t_name, + current_ip, + current_docker.as_deref(), + dep_name, + services, + ); + match hop { + Some((ip, docker)) => { + current_name.clone_from(dep_name); + current_ip = ip; + current_docker = docker; + } + None => break, } - None => break, } } } diff --git a/members/nullnet-server/src/services/input.rs b/members/nullnet-server/src/services/input.rs index a578df5..be2ec76 100644 --- a/members/nullnet-server/src/services/input.rs +++ b/members/nullnet-server/src/services/input.rs @@ -2,7 +2,9 @@ use crate::events::Event as ServerEvent; use crate::orchestrator::Orchestrator; use crate::services::changes::{ServiceChange, apply_changes, detect_config_changes}; use crate::services::clients::Client; -use crate::services::service_info::{CountryPolicy, ServiceInfo}; +use crate::services::service_info::{ + CountryPolicy, ServiceInfo, TriggerMap, placeholder_collision, +}; use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use nullnet_grpc_lib::nullnet_grpc::ServiceProtocol; use nullnet_liberror::{Error, ErrorHandler, Location, location}; @@ -293,7 +295,43 @@ impl ServicesToml { )) .handle_err(location!()); } - let triggers = s.triggers.into_iter().map(|t| (t.port, t.chain)).collect(); + // Group by port: more than one chain can share a port (two + // dependencies reached on the same real port, e.g. two + // plain-HTTPS deps both 443), disambiguated at trigger time by + // chain[0] — so within one port, every chain's chain[0] must be + // distinct, or the server could never tell them apart either. + let mut triggers: TriggerMap = HashMap::new(); + for t in s.triggers { + let chains = triggers.entry(t.port).or_default(); + if let Some(dup) = chains + .iter() + .find(|c: &&Vec| c.first() == t.chain.first()) + { + return Err(format!( + "service '{}': two triggers on port {} both resolve as '{}' — chains \ + sharing a port must have distinct chain[0] names", + s.name, + t.port, + dup.first().map(String::as_str).unwrap_or("") + )) + .handle_err(location!()); + } + chains.push(t.chain); + } + // Distinct chain[0] names are required above, but the client + // disambiguates a same-port first packet by hashing chain[0] into + // a placeholder address (see `placeholder_collision`), and that + // hash isn't collision-free. Two different names landing on the + // same address would be just as undetectable to the client as a + // literal duplicate, so reject it here too. + if let Some((port, a, b)) = placeholder_collision(&triggers) { + return Err(format!( + "service '{}': triggers '{a}' and '{b}' on port {port} hash to the same \ + placeholder address — rename one of them so the client can tell them apart", + s.name + )) + .handle_err(location!()); + } ret_val.insert( s.name, ServiceInfo::new( @@ -696,10 +734,91 @@ chain = ["dep.b"] // Not proxy-reachable... assert_eq!(map["backend.only"].timeout(), None); // ...yet it carries its triggers and proxy deps verbatim. - assert_eq!(map["backend.only"].triggers()[&5555], vec!["dep.b"]); + assert_eq!( + map["backend.only"].triggers()[&5555], + vec![vec!["dep.b".to_string()]] + ); assert_eq!(map["backend.only"].proxy_deps(), vec![vec!["dep.a"]]); } + #[test] + fn two_triggers_sharing_a_port_with_distinct_targets_both_parse() { + // Two dependencies reached on the same real port (e.g. two + // plain-HTTPS deps, both 443) — distinguishable by chain[0]. + let toml_str = r#" +[[services]] +name = "portal" + +[[services.triggers]] +port = 443 +chain = ["auth"] + +[[services.triggers]] +port = 443 +chain = ["billing"] +"#; + let parsed: ServicesToml = toml::from_str(toml_str).unwrap(); + let map = parsed.services_map().unwrap(); + + let mut chains = map["portal"].triggers()[&443].clone(); + chains.sort(); + assert_eq!( + chains, + vec![vec!["auth".to_string()], vec!["billing".to_string()]] + ); + } + + #[test] + fn two_triggers_sharing_a_port_with_same_target_is_rejected() { + // Same port, same chain[0] — genuinely ambiguous, nothing could ever + // disambiguate which chain a trigger on this port means. + let toml_str = r#" +[[services]] +name = "portal" + +[[services.triggers]] +port = 443 +chain = ["auth"] + +[[services.triggers]] +port = 443 +chain = ["auth"] +"#; + let parsed: ServicesToml = toml::from_str(toml_str).unwrap(); + let err = format!("{:?}", parsed.services_map().unwrap_err()); + assert!(err.contains("port 443"), "unexpected error: {err}"); + assert!(err.contains("auth"), "unexpected error: {err}"); + } + + #[test] + fn two_triggers_sharing_a_port_with_colliding_placeholder_is_rejected() { + // Distinct chain[0] names, but "tax" and "wallet" hash to the same + // placeholder octet under `last_octet_for` — the client's + // `resolve_target` couldn't tell these two apart on the wire any + // better than a literal duplicate could. + assert_eq!( + nullnet_grpc_lib::last_octet_for("tax"), + nullnet_grpc_lib::last_octet_for("wallet") + ); + let toml_str = r#" +[[services]] +name = "portal" + +[[services.triggers]] +port = 443 +chain = ["tax"] + +[[services.triggers]] +port = 443 +chain = ["wallet"] +"#; + let parsed: ServicesToml = toml::from_str(toml_str).unwrap(); + let err = format!("{:?}", parsed.services_map().unwrap_err()); + assert!(err.contains("port 443"), "unexpected error: {err}"); + assert!(err.contains("tax"), "unexpected error: {err}"); + assert!(err.contains("wallet"), "unexpected error: {err}"); + } + #[test] fn parses_parallel_proxy_branches() { let toml_str = r#" diff --git a/members/nullnet-server/src/services/service_info.rs b/members/nullnet-server/src/services/service_info.rs index 4465acc..dbd9685 100644 --- a/members/nullnet-server/src/services/service_info.rs +++ b/members/nullnet-server/src/services/service_info.rs @@ -6,6 +6,42 @@ use std::collections::{HashMap, HashSet}; use std::net::{IpAddr, Ipv4Addr}; use std::time::{Duration, Instant}; +/// Backend-triggered chains keyed by the trigger port observed on the +/// initiator's host. More than one chain can share a port — two dependencies +/// that happen to be reached on the same real port (e.g. two plain-HTTPS +/// deps, both 443) — disambiguated at trigger time by `target_name` +/// (chain[0]), which the client already resolves independently from the +/// packet's destination (its placeholder address — see nullnet-client's +/// `placeholder.rs`). +pub(crate) type TriggerMap = HashMap>>; + +/// First pair of chains sharing a port whose `chain[0]` names hash to the +/// same placeholder address, if any. Two *distinct* names on the same port +/// are normally fine — `chain_for` disambiguates by name, not address — but +/// the client picks which chain a first packet belongs to by matching the +/// packet's destination against each candidate's placeholder address (see +/// nullnet-client's `nfqueue/listener.rs::resolve_target` and +/// `placeholder.rs`). If two names collide under that hash, the client +/// can't tell them apart either and will silently pick whichever candidate +/// comes first, wiring the wrong dependency's traffic through. Distinct +/// `chain[0]` names are already required (see `services_map`'s dup check); +/// this catches the rarer case where two different names still land on the +/// same address. +pub(crate) fn placeholder_collision(triggers: &TriggerMap) -> Option<(u16, String, String)> { + for (&port, chains) in triggers { + for i in 0..chains.len() { + for other in &chains[i + 1..] { + let a = chains[i].first().map(String::as_str).unwrap_or(""); + let b = other.first().map(String::as_str).unwrap_or(""); + if nullnet_grpc_lib::last_octet_for(a) == nullnet_grpc_lib::last_octet_for(b) { + return Some((port, a.to_string(), b.to_string())); + } + } + } + } + None +} + /// Service names that participate in backend trigger chains: those that declare /// triggers ("have backend deps") and every service named in a trigger chain /// ("is a backend dep"). These are never paused — backend-dep networks aren't @@ -21,7 +57,7 @@ pub(crate) fn backend_involved_services( continue; } pinned.insert(name.clone()); - for dep in triggers.values().flatten() { + for dep in triggers.values().flatten().flatten() { pinned.insert(dep.clone()); } } @@ -62,7 +98,7 @@ impl ServiceInfo { #[allow(clippy::too_many_arguments)] pub(crate) fn new( proxy_deps: Vec>, - triggers: HashMap>, + triggers: TriggerMap, timeout: Option, max_networks: Option, protocol: ServiceProtocol, @@ -250,7 +286,7 @@ impl ServiceInfo { } } - pub(crate) fn triggers(&self) -> &HashMap> { + pub(crate) fn triggers(&self) -> &TriggerMap { match self { ServiceInfo::Unregistered(unreg) => &unreg.triggers, ServiceInfo::Registered(reg) => ®.triggers, @@ -260,7 +296,12 @@ impl ServiceInfo { /// True iff `other` appears in any of this service's dep lists (proxy or backend). pub(crate) fn deps_contain(&self, other: &str) -> bool { self.proxy_deps().iter().flatten().any(|d| d == other) - || self.triggers().values().flatten().any(|d| d == other) + || self + .triggers() + .values() + .flatten() + .flatten() + .any(|d| d == other) } } @@ -271,7 +312,7 @@ pub(crate) struct UnregisteredServiceInfo { proxy_deps: Vec>, /// Backend-triggered chains keyed by the trigger port observed on the /// initiator's host. One linear chain per port; no implicit fan-out. - triggers: HashMap>, + triggers: TriggerMap, /// Whether the proxy is reachable for this service, with the associated timeout. timeout: Option, /// Maximum number of networks for this service. @@ -290,7 +331,7 @@ impl UnregisteredServiceInfo { #[allow(clippy::too_many_arguments)] fn new( proxy_deps: Vec>, - triggers: HashMap>, + triggers: TriggerMap, timeout: Option, max_networks: Option, protocol: ServiceProtocol, @@ -384,8 +425,10 @@ pub(crate) struct RegisteredServiceInfo { /// is one linear branch; all branches are brought up in parallel. proxy_deps: Vec>, /// Backend-triggered chains keyed by the trigger port observed on the - /// initiator's host. One linear chain per port; no implicit fan-out. - triggers: HashMap>, + /// initiator's host. More than one chain may share a port, disambiguated + /// at trigger time by `chain_for`'s `target_name` match; no implicit + /// fan-out otherwise. + triggers: TriggerMap, /// Whether the proxy is reachable for this service, with the associated timeout. timeout: Option, /// Maximum number of networks for this service. @@ -427,17 +470,37 @@ impl RegisteredServiceInfo { .collect() } - /// Build the chain of edges for the trigger at `port`, if one exists. - /// Each chain starts at this service's replica. + /// Select the trigger chain for `port` that matches `target_name` — the + /// literal name (chain[0]) the initiator resolved via its placeholder. + /// When only one chain exists for that port, it's used regardless of + /// `target_name` — covers the common case, and callers that don't (yet) + /// know a target_name (an older client, or the empty string). With + /// multiple chains sharing a port, an exact match is required — an + /// empty or unmatched `target_name` is genuinely ambiguous and returns + /// `None` rather than guessing which dependency was meant. + pub(crate) fn chain_for(&self, port: u16, target_name: &str) -> Option<&Vec> { + let chains = self.triggers.get(&port)?; + match chains.as_slice() { + [only] => Some(only), + many => many + .iter() + .find(|c| c.first().map(String::as_str) == Some(target_name)), + } + } + + /// Build the chain of edges for the trigger at `port` matching + /// `target_name`, if one exists. Each chain starts at this service's + /// replica. pub(crate) fn backend_dependency_chain( &self, service_name: &str, service_ip: IpAddr, service_docker: Option<&str>, port: u16, + target_name: &str, services: &HashMap, ) -> Option> { - let chain = self.triggers.get(&port)?; + let chain = self.chain_for(port, target_name)?; Some(build_linear_chain( chain, service_name.to_string(), @@ -651,7 +714,7 @@ impl RegisteredServiceInfo { &self.replicas } - pub(crate) fn triggers(&self) -> &HashMap> { + pub(crate) fn triggers(&self) -> &TriggerMap { &self.triggers } @@ -804,3 +867,56 @@ fn build_linear_chain( } chain } + +#[cfg(test)] +mod chain_selection_tests { + use super::*; + + fn registered_with_triggers(triggers: TriggerMap) -> RegisteredServiceInfo { + RegisteredServiceInfo { + proxy_deps: Vec::new(), + triggers, + timeout: None, + max_networks: None, + protocol: ServiceProtocol::Http, + listen_port: None, + egress_policy: CountryPolicy::None, + ingress_policy: CountryPolicy::None, + replicas: Vec::new(), + } + } + + #[test] + fn single_chain_ignores_target_name() { + let reg = registered_with_triggers(HashMap::from([(443, vec![vec!["auth".to_string()]])])); + // Even an empty or unrelated target_name still resolves the only + // chain — covers pre-disambiguation clients and the common case. + assert_eq!(reg.chain_for(443, ""), Some(&vec!["auth".to_string()])); + assert_eq!( + reg.chain_for(443, "billing"), + Some(&vec!["auth".to_string()]) + ); + } + + #[test] + fn multiple_chains_require_exact_target_name_match() { + let reg = registered_with_triggers(HashMap::from([( + 443, + vec![vec!["auth".to_string()], vec!["billing".to_string()]], + )])); + assert_eq!(reg.chain_for(443, "auth"), Some(&vec!["auth".to_string()])); + assert_eq!( + reg.chain_for(443, "billing"), + Some(&vec!["billing".to_string()]) + ); + // Ambiguous — no guessing. + assert_eq!(reg.chain_for(443, ""), None); + assert_eq!(reg.chain_for(443, "unknown"), None); + } + + #[test] + fn unknown_port_returns_none() { + let reg = registered_with_triggers(HashMap::new()); + assert_eq!(reg.chain_for(443, "auth"), None); + } +} diff --git a/members/nullnet-server/src/tests.rs b/members/nullnet-server/src/tests.rs index 1f9d78e..66ddfb0 100644 --- a/members/nullnet-server/src/tests.rs +++ b/members/nullnet-server/src/tests.rs @@ -141,7 +141,7 @@ async fn trigger_backend_chain( port: u16, ) { server - .handle_backend_trigger(initiator_name, port, initiator_ip, None) + .handle_backend_trigger(initiator_name, port, initiator_ip, None, "") .await .expect("backend trigger failed"); } @@ -162,6 +162,7 @@ async fn setup_backend_chain_for_replica( initiator_ip, initiator_docker, port, + "", ) .await .expect("setup_backend_chain failed"); @@ -1981,7 +1982,7 @@ async fn triggers_changed_swap_A_trigger() { assert_graphviz(&guard, TRIGGERS_CHANGED, "after_swap_A_trigger.dot"); assert_eq!( stack_view(&guard)["A"].triggers().get(&5555), - Some(&vec!["D".to_string()]) + Some(&vec![vec!["D".to_string()]]) ); drop(guard); @@ -2267,7 +2268,7 @@ async fn backend_trigger_disambiguates_colocated_replica_by_container() { let server = backend_disambiguation_setup().await; server - .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), Some("a2")) + .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), Some("a2"), "") .await .expect("backend trigger for a2 should succeed"); @@ -2308,7 +2309,7 @@ async fn backend_trigger_unknown_container_errors_without_ip_fallback() { let server = backend_disambiguation_setup().await; let result = server - .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), Some("ghost")) + .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), Some("ghost"), "") .await; assert!( @@ -2325,7 +2326,7 @@ async fn backend_trigger_without_container_falls_back_to_ip_only() { let server = backend_disambiguation_setup().await; server - .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), None) + .handle_backend_trigger("A", 5555, ip(1, 1, 1, 1), None, "") .await .expect("IP-only trigger should succeed"); @@ -2858,7 +2859,7 @@ async fn backend_involved_replicas_never_suspended() { "initiator".to_string(), ServiceInfo::new( vec![], - HashMap::from([(8080u16, vec!["dep".to_string()])]), + HashMap::from([(8080u16, vec![vec!["dep".to_string()]])]), Some(30), None, ServiceProtocol::Http, @@ -3187,7 +3188,14 @@ async fn trigger_carries_the_containers_that_own_it() { assert_eq!(triggers.len(), 1, "only S declares a trigger"); assert_eq!(triggers[0].service_name, "S"); - assert_eq!(triggers[0].ports, vec![8932]); + assert_eq!( + triggers[0] + .trigger_ports + .iter() + .map(|tp| (tp.port, tp.target_name.as_str())) + .collect::>(), + vec![(8932, "P")] + ); assert_eq!( triggers[0].containers, vec!["stack_s.1.abc"], diff --git a/members/nullnet-server/ui/src/pages/Config.tsx b/members/nullnet-server/ui/src/pages/Config.tsx index 5316319..1f0d169 100644 --- a/members/nullnet-server/ui/src/pages/Config.tsx +++ b/members/nullnet-server/ui/src/pages/Config.tsx @@ -203,7 +203,7 @@ export default function Config() { )} {status.kind === 'error' && ( - Parse error + Rejected {status.msg} )} diff --git a/members/nullnet-server/ui/src/pages/Services.tsx b/members/nullnet-server/ui/src/pages/Services.tsx index 231895a..a5f6826 100644 --- a/members/nullnet-server/ui/src/pages/Services.tsx +++ b/members/nullnet-server/ui/src/pages/Services.tsx @@ -140,12 +140,14 @@ export default function Services() { {svc.max_networks} )} - {Object.entries(svc.triggers).map(([port, chain]) => ( - - Trigger :{port} - {chain.join(' → ')} - - ))} + {Object.entries(svc.triggers).flatMap(([port, chains]) => + chains.map((chain, i) => ( + + Trigger :{port} + {chain.join(' → ')} + + )) + )} {svc.proxy_dependencies.flat().length > 0 && ( Dependencies diff --git a/members/nullnet-server/ui/src/types.ts b/members/nullnet-server/ui/src/types.ts index bc5b414..e123bf5 100644 --- a/members/nullnet-server/ui/src/types.ts +++ b/members/nullnet-server/ui/src/types.ts @@ -15,7 +15,10 @@ export interface ServiceJson { registered: boolean; replicas: ReplicaJson[]; proxy_dependencies: string[][]; - triggers: Record; + // Port -> the chains sharing it. Usually one chain per port; more than one + // means two dependencies reached on the same real port, disambiguated at + // trigger time by chain[0]. + triggers: Record; timeout_secs?: number; max_networks?: number; }