Optional service host-pinning

CHANGELOG.md

@@ -4,6 +4,7 @@ All Nullnet releases with the relative changes are documented in this file.
 
 ## [UNRELEASED]
 ### Added
+- Optional service host pinning and automatic proxy TCP/UDP listen-port firewall allowances ([#184](https://github.com/NullNet-ai/nullnet/pull/184) — fixes [#177](https://github.com/NullNet-ai/nullnet/issues/177))
 - Per-service egress/ingress traffic filters: arbitrary AND/OR/group combinations of Country, Organization, Src IP (ingress), and Dst IP (egress) conditions, evaluated via `rpn-predicate-interpreter` — replaces the country-only egress/ingress policy ([#171](https://github.com/NullNet-ai/nullnet/pull/171) — fixes [#143](https://github.com/NullNet-ai/nullnet/issues/143))
 - Persist ingress and egress sessions to SQLite and show the full history on the Sessions page, filterable by status, service, direction, and policy verdict, with its own retention window ([#170](https://github.com/NullNet-ai/nullnet/pull/170) — fixes [#156](https://github.com/NullNet-ai/nullnet/issues/156))
 ### Changed

SETUP.md

@@ -218,6 +218,8 @@ The repository should be cloned under `/root` so the provided `setup-*.sh` scrip
     the container name; matched against a running container
   - `process_path` — a listening process's exe path (`/proc/<pid>/exe`); matched against a host
     (non-Docker) service
+- `host_ip` optionally limits either match to one node's control-channel IPv4 address
+  (for example, `host_ip = "192.168.1.103"` for that host's SSH service). Omit it to match all hosts.
 - `timeout` controls proxy-reachability: when present the service is a proxy-reachable entry point
   with that per-client idle timeout in seconds (`0` disables the timeout); omit it to keep the
   service off the proxy (backend-only)
@@ -233,6 +235,9 @@ The repository should be cloned under `/root` so the provided `setup-*.sh` scrip
   the external port nullnet-proxy binds directly and forwards raw traffic from. `listen_port` must
   be globally unique per protocol across every stack (the server refuses to start, or rejects a
   hot-reload, if two services claim the same `protocol`/`listen_port` pair)
+- TCP/UDP `listen_port` values are automatically allowed through the proxy host's eBPF ingress
+  firewall at startup and refreshed within the client's 10-second service-report interval.
+  Backend `port` values are not opened; explicit firewall allowlists still apply.
 - `egress_filter`/`ingress_filter` restrict traffic with an AND/OR combination of conditions,
   evaluated via `rpn-predicate-interpreter` (the same postfix-expression engine
   `appguard-server/src/firewall/` uses). `groups` is OR-of-ANDs: every condition within a group must

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

@@ -5,8 +5,8 @@
 //! interface. Structural allow is nullnet control-plane (gRPC) + data-plane
 //! (VXLAN/forward to known peers) + ARP; a CT map then permits established
 //! returns. ICMP is always allowed (both directions). Everything else is an
-//! explicit, env-driven allow: the four `{INGRESS,EGRESS}_ALLOW_{TCP,UDP}_PORTS`
-//! lists (→ `ALLOW_PORTS` map). On the egress-gateway host all outbound is
+//! server-decided allow: explicit port lists plus TCP/UDP proxy listen ports
+//! (→ `ALLOW_PORTS` map). On the egress-gateway host all outbound is
 //! additionally allowed and tracked. Peers are added/removed from the `PEERS` map,
 //! and each VXLAN tunnel's per-tunnel dstport (paired with its specific peer) from
 //! the `VXLAN_PORTS` map, by the control channel as edges come and go.
@@ -18,7 +18,7 @@
 use aya::Ebpf;
 use aya::maps::{HashMap as AyaHashMap, MapData};
 use nullnet_liberror::{Error, ErrorHandler, Location, location};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::net::Ipv4Addr;
 use std::sync::{Arc, Mutex};
 
@@ -28,10 +28,9 @@ const EGRESS_PROG: &str = "nullnet_fw_egress";
 const PROTO_TCP: u8 = 6;
 const PROTO_UDP: u8 = 17;
 
-/// Explicit firewall allow policy, decided globally by the server and delivered
-/// in the `NetworkType` response at startup. Nothing is hardcoded: every
-/// host-service port a node accepts or initiates to is listed here. nullnet's
-/// own control/data plane and CT returns are always allowed.
+/// Server-decided firewall policy delivered at startup. Ingress allowances
+/// are refreshed with service reports; control/data plane and CT returns
+/// are always allowed.
 pub struct FirewallConfig {
     pub server_ip: Ipv4Addr,
     pub control_port: u16,
@@ -65,6 +64,7 @@ pub struct Firewall {
     #[allow(dead_code)]
     bpf: Ebpf,
     pub peers: Arc<FirewallPeers>,
+    pub allow_ports: Arc<Mutex<FirewallAllowPorts>>,
     pub vxlan_ports: Arc<FirewallVxlanPorts>,
 }
 
@@ -225,7 +225,7 @@ pub fn enable(iface: &str, cfg: &FirewallConfig) -> Result<Firewall, Error> {
         cfg.egress_gateway
     );
 
-    populate_allow_ports(&mut bpf, cfg)?;
+    let allow_ports = populate_allow_ports(&mut bpf, cfg)?;
 
     let peers_map: AyaHashMap<MapData, u32, u8> = bpf
         .take_map("PEERS")
@@ -243,6 +243,7 @@ pub fn enable(iface: &str, cfg: &FirewallConfig) -> Result<Firewall, Error> {
 
     Ok(Firewall {
         bpf,
+        allow_ports: Arc::new(Mutex::new(allow_ports)),
         peers: Arc::new(FirewallPeers::new(peers_map)),
         vxlan_ports: Arc::new(FirewallVxlanPorts::new(vxlan_ports_map)),
     })
@@ -267,10 +268,37 @@ fn attach_classifier(
     Ok(())
 }
 
-/// Fill the ALLOW_PORTS map from the four explicit env lists. Each port is keyed
-/// by direction + protocol (see `allow_key`); nothing is added implicitly. The map
-/// is taken only to populate it; the attached programs keep it alive kernel-side.
-fn populate_allow_ports(bpf: &mut Ebpf, cfg: &FirewallConfig) -> Result<(), Error> {
+/// Retained map handle for refreshing server-decided ingress allowances.
+pub struct FirewallAllowPorts {
+    map: AyaHashMap<MapData, u32, u8>,
+    ingress: HashSet<u32>,
+}
+
+impl FirewallAllowPorts {
+    pub fn update_ingress(&mut self, tcp: &[u32], udp: &[u32]) -> Result<(), Error> {
+        let mut desired = HashSet::new();
+        for (proto, ports) in [(PROTO_TCP, tcp), (PROTO_UDP, udp)] {
+            for &port in ports {
+                let port = u16::try_from(port).handle_err(location!())?;
+                desired.insert(allow_key(false, proto, port));
+            }
+        }
+        let removed: Vec<_> = self.ingress.difference(&desired).copied().collect();
+        for key in removed {
+            self.map.remove(&key).handle_err(location!())?;
+            self.ingress.remove(&key);
+        }
+        for key in desired {
+            if !self.ingress.contains(&key) {
+                self.map.insert(key, 0u8, 0).handle_err(location!())?;
+                self.ingress.insert(key);
+            }
+        }
+        Ok(())
+    }
+}
+
+fn populate_allow_ports(bpf: &mut Ebpf, cfg: &FirewallConfig) -> Result<FirewallAllowPorts, Error> {
     let mut map: AyaHashMap<MapData, u32, u8> = bpf
         .take_map("ALLOW_PORTS")
         .ok_or("ALLOW_PORTS map not found in bytecode")
@@ -283,12 +311,17 @@ fn populate_allow_ports(bpf: &mut Ebpf, cfg: &FirewallConfig) -> Result<(), Erro
         (true, PROTO_TCP, &cfg.egress_tcp),
         (true, PROTO_UDP, &cfg.egress_udp),
     ];
+    let mut ingress = HashSet::new();
     for (is_egress, proto, ports) in sets {
         for &p in ports {
-            let _ = map.insert(allow_key(is_egress, proto, p), 0u8, 0);
+            let key = allow_key(is_egress, proto, p);
+            map.insert(key, 0u8, 0).handle_err(location!())?;
+            if !is_egress {
+                ingress.insert(key);
+            }
         }
     }
-    Ok(())
+    Ok(FirewallAllowPorts { map, ingress })
 }
 
 fn raise_memlock_rlimit() {

members/nullnet-client/src/main.rs

@@ -224,10 +224,12 @@ async fn main() -> Result<(), Error> {
         liveness,
     );
 
+    let firewall_allow_ports = ebpf_firewall.allow_ports.clone();
+
     // declare services + push the port→trigger-owners map to the NFQUEUE
     // listener on each refresh.
     tokio::spawn(async move {
-        declare_services(grpc_server, config_tx, docker_changed)
+        declare_services(grpc_server, config_tx, docker_changed, firewall_allow_ports)
             .await
             .expect("Failed to declare services");
     });
@@ -336,6 +338,7 @@ async fn declare_services(
     grpc_server: NullnetGrpcInterface,
     config_tx: UnboundedSender<TriggerMap>,
     docker_changed: Arc<Notify>,
+    firewall_allow_ports: Arc<std::sync::Mutex<ebpf::FirewallAllowPorts>>,
 ) -> Result<(), Error> {
     let mut last_snapshot: Vec<String> = Vec::new();
     loop {
@@ -399,6 +402,23 @@ async fn declare_services(
                 });
             }
             Ok(response) => {
+                let firewall_result = firewall_allow_ports.lock().unwrap().update_ingress(
+                    &response.ingress_allow_tcp_ports,
+                    &response.ingress_allow_udp_ports,
+                );
+                if let Err(e) = firewall_result {
+                    eprintln!("Failed to refresh eBPF ingress ports: {e:?}");
+                    let _ = grpc_server
+                        .report_event(AgentEvent {
+                            event: Some(AgentEventKind::FirewallRulesLoadFailed(
+                                AgentFirewallRulesLoadFailed {
+                                    path: "ebpf ingress ports".to_string(),
+                                    error_message: format!("{e:?}"),
+                                },
+                            )),
+                        })
+                        .await;
+                }
                 if snapshot != last_snapshot {
                     last_snapshot = snapshot;
                     let grpc = grpc_server.clone();

members/nullnet-grpc-lib/proto/nullnet_grpc.proto

@@ -263,6 +263,8 @@ message Listener {
 // these ports, the client fires BackendTrigger(service_name, port).
 message ServicesListResponse {
   repeated ServiceTrigger service_triggers = 1;
+  repeated uint32 ingress_allow_tcp_ports = 2;
+  repeated uint32 ingress_allow_udp_ports = 3;
 }
 
 message ServiceTrigger {
@@ -580,4 +582,4 @@ message AgentProxyRequestRouted     { string service_name = 1; string client_ip
 message AgentTcpListenerBindFailed  { uint32 listen_port = 1; string service_name = 2; string error_message = 3; }
 message AgentUdpListenerBindFailed  { uint32 listen_port = 1; string service_name = 2; string error_message = 3; }
 message AgentTcpUpstreamConnectFailed { string service_name = 1; string client_ip = 2; string error_message = 3; }
-message AgentUdpUpstreamConnectFailed { string service_name = 1; string client_ip = 2; string error_message = 3; }
+message AgentUdpUpstreamConnectFailed { string service_name = 1; string client_ip = 2; string error_message = 3; }

members/nullnet-grpc-lib/src/proto/nullnet_grpc.rs

@@ -228,6 +228,10 @@ pub struct Listener {
 pub struct ServicesListResponse {
     #[prost(message, repeated, tag = "1")]
     pub service_triggers: ::prost::alloc::vec::Vec<ServiceTrigger>,
+    #[prost(uint32, repeated, tag = "2")]
+    pub ingress_allow_tcp_ports: ::prost::alloc::vec::Vec<u32>,
+    #[prost(uint32, repeated, tag = "3")]
+    pub ingress_allow_udp_ports: ::prost::alloc::vec::Vec<u32>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ServiceTrigger {

members/nullnet-server/src/db/migrations/2026-09-08-000001_service_host_ip/down.sql

@@ -0,0 +1 @@
+ALTER TABLE services DROP COLUMN host_ip;

members/nullnet-server/src/db/migrations/2026-09-08-000001_service_host_ip/up.sql

@@ -0,0 +1 @@
+ALTER TABLE services ADD COLUMN host_ip TEXT;

members/nullnet-server/src/db/models.rs

@@ -71,6 +71,7 @@ pub(crate) struct ServiceRow {
     pub(crate) name: String,
     pub(crate) docker_container: Option<String>,
     pub(crate) process_path: Option<String>,
+    pub(crate) host_ip: Option<String>,
     pub(crate) port: Option<i32>,
     pub(crate) timeout: Option<i64>,
     pub(crate) max_networks: Option<i32>,
@@ -89,6 +90,7 @@ pub(crate) struct NewServiceRow<'a> {
     pub(crate) name: &'a str,
     pub(crate) docker_container: Option<&'a str>,
     pub(crate) process_path: Option<&'a str>,
+    pub(crate) host_ip: Option<&'a str>,
     pub(crate) port: Option<i32>,
     pub(crate) timeout: Option<i64>,
     pub(crate) max_networks: Option<i32>,

members/nullnet-server/src/db/schema.rs

@@ -32,6 +32,7 @@ diesel::table! {
         name -> Text,
         docker_container -> Nullable<Text>,
         process_path -> Nullable<Text>,
+        host_ip -> Nullable<Text>,
         port -> Nullable<Integer>,
         timeout -> Nullable<BigInt>,
         max_networks -> Nullable<Integer>,

members/nullnet-server/src/db/stacks.rs

@@ -18,6 +18,7 @@ pub(crate) struct ServiceInsert<'a> {
     pub(crate) name: &'a str,
     pub(crate) docker_container: Option<&'a str>,
     pub(crate) process_path: Option<&'a str>,
+    pub(crate) host_ip: Option<&'a str>,
     pub(crate) port: Option<i32>,
     pub(crate) timeout: Option<i64>,
     pub(crate) max_networks: Option<i32>,
@@ -168,6 +169,7 @@ impl StackRepository {
                 name: s.name,
                 docker_container: s.docker_container,
                 process_path: s.process_path,
+                host_ip: s.host_ip,
                 port: s.port,
                 timeout: s.timeout,
                 max_networks: s.max_networks,
@@ -283,6 +285,7 @@ mod tests {
             name,
             docker_container: Some("my-app_web"),
             process_path: None,
+            host_ip: Some("192.0.2.1"),
             port: Some(8080),
             timeout: Some(0),
             max_networks: None,
@@ -306,6 +309,7 @@ mod tests {
         let services = repo.services_for("alpha").await.unwrap();
         assert_eq!(services.len(), 1);
         assert_eq!(services[0].name, "web");
+        assert_eq!(services[0].host_ip.as_deref(), Some("192.0.2.1"));
         assert_eq!(services[0].docker_container.as_deref(), Some("my-app_web"));
 
         let ids: Vec<i32> = services.iter().map(|s| s.id).collect();

members/nullnet-server/src/http_server/config.rs

@@ -7,10 +7,12 @@
 
 use super::AppState;
 use crate::events::Event as ServerEvent;
+use crate::services::changes::{ServiceChange, apply_changes};
 use crate::services::input::{
     RouteMap, ServicesToml, StackMap, apply_config_update, detect_name_conflicts,
     detect_port_conflicts, detect_route_conflicts,
 };
+use crate::services::service_info::ServiceInfo;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use nullnet_liberror::Error;
@@ -130,11 +132,35 @@ pub(super) async fn reload_and_apply(state: &AppState) -> Result<(), Error> {
     let route_conflicts = detect_route_conflicts(&loaded_routes);
     let name_conflicts = detect_name_conflicts(&loaded_services);
     if conflicts.is_empty() && route_conflicts.is_empty() && name_conflicts.is_empty() {
+        let mut index = state.match_index.write().await;
         {
             let mut services_mut = state.services.write().await;
             apply_config_update(&mut services_mut, loaded_services, &state.orchestrator).await;
+            for (stack, entries) in &loaded_index {
+                let stack_map = services_mut.get_mut(stack).unwrap();
+                let mut changes = Vec::new();
+                for entry in entries {
+                    if let Some(host_ip) = entry.host_ip
+                        && let Some(ServiceInfo::Registered(reg)) = stack_map.get(&entry.name)
+                    {
+                        for replica in reg.replicas() {
+                            if replica.ip() != std::net::IpAddr::V4(host_ip) {
+                                changes.push(ServiceChange::ReplicaRemoved {
+                                    name: entry.name.clone(),
+                                    ip: replica.ip(),
+                                    docker_container: replica
+                                        .docker_container()
+                                        .map(str::to_string),
+                                });
+                            }
+                        }
+                    }
+                }
+                apply_changes(changes, stack_map, None, &state.orchestrator, stack).await;
+            }
         }
-        *state.match_index.write().await = loaded_index;
+        *index = loaded_index;
+        drop(index);
         *state.routes.write().await = loaded_routes;
         state.config_changed.notify_one();
         state.port_mappings_changed.notify_one();