mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

log the agent bridge's forwarded flows in the hive_egress table

Browse source 
A forward chain in the hive-owned nft table logs each new flow from the agent bridge to the uplink (v4 and v6) and counts every packet. It never blocks. The mesh interface is skipped. Logging is rate limited per source address; a flow that limit cannot log gets one line from a shared limit with its own prefix. services.hyperhive.network.egress.forward selects accept, log (default) or drop; drop is refused at evaluation until the interception it needs exists.
This commit is contained in:
atlas 2026-11-12 08:56:44 +01:00
commit bd79b58e78
3 changed files with 321 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

99
nix/host-modules/hive-egress.nix
View file

@ -5,11 +5,13 @@
# host lives here, in a table no firewall backend writes. Only
# `networking.nftables.flushRuleset` removes it; the unit re-applies it then.
#
# The table holds one chain: traffic from the agent bridge to host
# The table holds two chains. `input`: traffic from the agent bridge to host
# loopback is dropped. The kernel does the same by default (a packet for
# 127.0.0.0/8 arriving on a non-loopback interface is a martian while
# `route_localnet` is 0); this rule is the second layer, and the assertion
# below keeps the first one on.
# below keeps the first one on. `forward` (`egress.forward = "log"`): traffic
# the host routes from the agent bridge to anywhere else is logged and
# counted, never blocked.
{
lib,
config,
@ -19,20 +21,63 @@
let
cfg = config.services.hyperhive.network;
nft = lib.getExe pkgs.nftables;
br = cfg.bridgeName;
# The mesh interface (swarm-wireguard.nix): inter-hive traffic is not egress.
meshInterface = "wg-hive";
# Logging is per source address, so one busy agent cannot use up the budget
# of another: a burst of 20 lines, then one a second, and an address is
# forgotten after a minute without a new flow. The rule accepts after it
# logs. A flow the rule cannot log falls through to the unmetered rule below
# it: the address was over its rate, or the set is full (`update` then fails
# and the rule stops matching), which a container spoofing source addresses
# can bring about. Only that case reaches the unmetered rule, so a flow
# gets one line at most. The unmetered rule has one budget for all
# addresses, 5 lines a second after a burst of 50, about 100 MB of journal a
# day when it never rests.
logRate = "1/second burst 20 packets";
unmeteredRate = "5/second burst 50 packets";
fromBridge = ''iifname "${br}" oifname != "${br}"'';
logRule =
family: addr:
''${fromBridge} ct state new update @log_${family} { ${addr} saddr limit rate ${logRate} } log prefix "hive-egress-would-drop " accept'';
# Priority `filter - 10` runs before the NixOS firewall's forward handling,
# which sits at `filter` on both backends (`FORWARD` of the iptables backend,
# the `forward` chain of `nixos-fw`). An `accept` here ends this chain only,
# so the chain neither allows what those refuse nor refuses anything itself.
# Traffic to the host's own addresses (squid, dnsmasq, the gateway) is
# delivered locally and never reaches the forward hook.
forwardChain = ''
set log_v4 { type ipv4_addr; flags dynamic; timeout 1m; size 4096; }
set log_v6 { type ipv6_addr; flags dynamic; timeout 1m; size 4096; }
chain forward {
type filter hook forward priority filter - 10; policy accept;
iifname "${br}" oifname "${meshInterface}" accept
${fromBridge} counter
${logRule "v4" "ip"}
${logRule "v6" "ip6"}
${fromBridge} ct state new limit rate ${unmeteredRate} log prefix "hive-egress-would-drop-unmetered "
${fromBridge} accept
}
'';
# One transaction: `add` + `delete` drop an old copy of the table and the
# definition recreates it, so the file loads the same on first start, on a
# reload and after something else flushed the ruleset.
ruleset = pkgs.writeText "hive-egress.nft" ''
rulesetText = ''
add table inet hive_egress
delete table inet hive_egress
table inet hive_egress {
chain input {
type filter hook input priority filter - 10; policy accept;
iifname "${cfg.bridgeName}" ip daddr 127.0.0.0/8 counter drop
iifname "${cfg.bridgeName}" ip6 daddr ::1 counter drop
iifname "${br}" ip daddr 127.0.0.0/8 counter drop
iifname "${br}" ip6 daddr ::1 counter drop
}
${lib.optionalString (cfg.egress.forward == "log") forwardChain}
}
'';
@ -48,10 +93,52 @@ let
"1"
true
];
ruleset = pkgs.writeText "hive-egress.nft" cfg.egress.ruleset;
in
{
options.services.hyperhive.network.egress.forward = lib.mkOption {
type = lib.types.enum [
"accept"
"log"
"drop"
];
default = "log";
description = ''
What the `inet hive_egress` table does with traffic the host forwards
from the agent bridge to anywhere but the bridge and the mesh.
- `accept`: nothing; the table has no forward chain.
- `log`: each new flow is logged with the prefix `hive-egress-would-drop `,
rate limited per source address, and counted. A flow the per-source limit
cannot log gets a line with the prefix `hive-egress-would-drop-unmetered `
from a limit shared by all addresses. Nothing is blocked. The source
address maps to an agent through the hive's DHCP leases.
- `drop`: not available yet; the evaluation refuses it.
'';
};
options.services.hyperhive.network.egress.ruleset = lib.mkOption {
type = lib.types.lines;
internal = true;
readOnly = true;
default = rulesetText;
defaultText = lib.literalMD "the `inet hive_egress` table definition";
description = ''
The nft file the `hive-egress` unit loads, as text.
'';
};
config = lib.mkIf cfg.enable {
assertions = [
{
assertion = cfg.egress.forward != "drop";
message = ''
services.hyperhive.network.egress.forward = "drop" is not available yet:
the interception that lets an agent's web traffic through squid has not
landed, so a drop would cut every direct connection with no replacement.
Use "log" or "accept".
'';
}
{
assertion = !(lib.any (k: isOn (config.boot.kernel.sysctl.${k} or 0)) routeLocalnetKeys);
message = ''
@ -64,7 +151,7 @@ in
];
systemd.services.hive-egress = {
description = "hive nftables table (agent bridge to host loopback drop)";
description = "hive nftables table (agent bridge egress log, host loopback drop)";
wantedBy = [ "multi-user.target" ];
wants = [ "network-pre.target" ];
before = [ "network-pre.target" ];

74
nix/module-eval/core-toggle.nix
View file

@ -70,6 +70,18 @@ let
{ boot.kernel.sysctl."net.ipv4.conf.all.route_localnet" = 1; }
] { network.enable = true; };
# The two settings of `egress.forward` the ordinary host (`log`) does not
# render, for ../host-modules/hive-egress.nix: `accept` leaves the table
# without a forward chain, `drop` is refused by an assertion.
egressForwardAccept = hive {
network.enable = true;
network.egress.forward = "accept";
};
egressForwardDrop = hive {
network.enable = true;
network.egress.forward = "drop";
};
# Every hive defaults the agents' queue address to the queue's name, so the
# only hive without one is a hive told to have none. Also the host still
# setting the four removed host options; no case on its queue reads them.
@ -1521,6 +1533,68 @@ let
&& bare.networking.firewall.extraInputRules == ""
&& !(centralToggleOff.systemd.services ? hive-egress);
}
{
# The forward chain is in the ordinary host's table and logs the bridge's
# forwarded flows without a verdict that blocks. The ruleset is read as
# text; the last rule of the chain is the accept that keeps the log from
# changing any outcome.
name = "the default egress table logs the bridge's forwarded flows and blocks none";
ok =
let
text = bare.services.hyperhive.network.egress.ruleset;
chain = lib.last (lib.splitString "chain forward" text);
rules = lib.filter (l: l != "") (map lib.trim (lib.splitString "\n" chain));
idx = pat: lib.lists.findFirstIndex (lib.hasInfix pat) 1000 rules;
in
lib.hasInfix "chain forward" text
&& lib.hasInfix "type filter hook forward priority filter - 10; policy accept;" chain
&& lib.hasInfix ''iifname "hive-br0" oifname != "hive-br0" ct state new update @log_v4 { ip saddr limit rate'' chain
&& lib.hasInfix ''iifname "hive-br0" oifname != "hive-br0" ct state new update @log_v6 { ip6 saddr limit rate'' chain
&& lib.hasInfix ''log prefix "hive-egress-would-drop "'' chain
&& lib.hasInfix ''iifname "hive-br0" oifname "wg-hive" accept'' chain
# The per-source rules accept after logging, and the unmetered rule
# comes after both: a flow gets one line, and one the per-source rules
# cannot log still gets the unmetered one.
&& lib.all (l: lib.hasSuffix " accept" l) (lib.filter (lib.hasInfix "update @log_") rules)
&& lib.hasInfix ''iifname "hive-br0" oifname != "hive-br0" ct state new limit rate 5/second burst 50 packets log prefix "hive-egress-would-drop-unmetered "'' chain
&& idx "update @log_v4" < idx "update @log_v6"
&& idx "update @log_v6" < idx "would-drop-unmetered"
&& lib.last (lib.filter (l: l != "}") rules) == ''iifname "hive-br0" oifname != "hive-br0" accept''
&& !(lib.any (l: lib.hasSuffix " drop" l || lib.hasSuffix " reject" l) rules)
&& lib.hasInfix "chain input" text;
}
{
# `accept` renders the table without the forward chain and without the
# log prefix; the input chain is the presence arm that the text it
# reads is the table.
name = "egress.forward accept leaves the table without a forward chain";
ok =
let
text = egressForwardAccept.services.hyperhive.network.egress.ruleset;
in
lib.hasInfix "chain input" text
&& !(lib.hasInfix "chain forward" text)
&& !(lib.hasInfix "hive-egress-would-drop" text)
&& !(lib.any (a: !a.assertion) egressForwardAccept.assertions);
}
{
# `drop` is refused by an assertion that names the option and says it is
# not available; `bare` and the `accept` host, which set `log` and
# `accept`, are the controls that the lookup finds nothing for them.
name = "egress.forward drop is refused as not available yet";
ok =
let
refused =
m:
lib.any (
a:
!a.assertion
&& lib.hasInfix "services.hyperhive.network.egress.forward" a.message
&& lib.hasInfix "not available yet" a.message
) m.assertions;
in
refused egressForwardDrop && !(refused bare) && !(refused egressForwardAccept);
}
{
# The host that turns `route_localnet` on is refused by an assertion
# naming it; the ordinary host, which leaves it alone, is not.

106
nix/vm-tests/hive-squid.nix
View file

@ -10,7 +10,9 @@
# covers the real option wiring.
#
# It also covers ../host-modules/hive-egress.nix: `proxy` loads the real module, so the
# agent's dials to the host's loopback run against the table the module installs.
# agent's dials to the host's loopback, and its direct connects to the uplink
# (`proxy` forwards and masquerades like the hive host), run against the table
# the module installs.
{ pkgs }:
let
bridgeIp = "192.168.1.1";
@ -98,6 +100,10 @@ let
address = "fd00:99::${toString host}";
prefixLength = 64;
};
bridgeUla = host: {
address = "fd00:98::${toString host}";
prefixLength = 64;
};
# Plain TCP on 443: a CONNECT is answered before any TLS, which is all the
# test reads.
@ -151,6 +157,14 @@ pkgs.testers.runNixOSTest {
++ privateAddrs 1
);
networking.interfaces.eth2.ipv6.addresses = [ (ulaAddr 1) ];
networking.interfaces.eth1.ipv6.addresses = [ (bridgeUla 1) ];
# What hive-network.nix sets for the agent bridge: forwarding, and NAT
# for everything that leaves it. The agent's direct path out exists.
networking.nat = {
enable = true;
internalInterfaces = [ "eth1" ];
};
boot.kernel.sysctl."net.ipv6.conf.all.forwarding" = 1;
# Names the proxy resolves, each a different destination class.
networking.hosts = {
# `other.t.local` and the self-check's deny canary resolve, are
@ -225,6 +239,13 @@ pkgs.testers.runNixOSTest {
++ privateAddrs 2
);
networking.interfaces.eth1.ipv6.addresses = [ (ulaAddr 2) ];
networking.interfaces.eth1.ipv6.routes = [
{
address = "fd00:98::";
prefixLength = 64;
via = "fd00:99::1";
}
];
networking.firewall.allowedTCPPorts = [ 443 ];
};
@ -240,6 +261,26 @@ pkgs.testers.runNixOSTest {
prefixLength = 24;
}
];
networking.interfaces.eth1.ipv4.routes = [
{
address = "203.0.113.0";
prefixLength = 24;
via = bridgeIp;
}
{
address = "198.51.100.0";
prefixLength = 24;
via = bridgeIp;
}
];
networking.interfaces.eth1.ipv6.addresses = [ (bridgeUla 3) ];
networking.interfaces.eth1.ipv6.routes = [
{
address = "fd00:99::";
prefixLength = 64;
via = "fd00:98::1";
}
];
};
};
@ -735,5 +776,68 @@ pkgs.testers.runNixOSTest {
proxy.succeed("nft delete table inet hive_egress")
assert not dial("127.0.0.5"), "route_localnet=0 refuses it alone"
proxy.succeed("systemctl reload hive-egress.service")
def would_drop(pattern):
# The forward chain's log lines in the kernel log that match `pattern`.
out = proxy.succeed("journalctl -k -o cat --no-pager | grep hive-egress-would-drop || true")
return [line for line in out.splitlines() if re.search(pattern, line)]
with checked("a direct connect from the agent bridge still works in log mode, and the flow is logged with the agent's address"):
# Through squid, to the gateway on the bridge and out of the proxy itself
# first: none of them is forwarded traffic, so none may be logged.
assert connect("forge.t.local", ATLAS) == "200", "through squid"
assert dial("${bridgeIp}"), "the gateway on the bridge IP"
assert dial("203.0.113.2"), "a direct connect to an outside IP works"
proxy.wait_until_succeeds(
"journalctl -k -o cat --no-pager | grep -E "
"'hive-egress-would-drop IN=eth1 OUT=eth2 .*SRC=192\\.168\\.1\\.3 DST=203\\.0\\.113\\.2 .*PROTO=TCP .*DPT=443 '"
)
agent.succeed("ping -c1 -W2 203.0.113.2")
proxy.wait_until_succeeds(
"journalctl -k -o cat --no-pager | grep -E "
"'hive-egress-would-drop IN=eth1 OUT=eth2 .*SRC=192\\.168\\.1\\.3 DST=203\\.0\\.113\\.2 .*PROTO=ICMP '"
)
# Control: the same journal holds no line for the traffic that is not forwarded.
assert not would_drop(r"DST=192\.168\.1\.1 "), "the bridge IP is the host's own address"
assert not would_drop(r"DPT=3128 "), "squid's port is the host's own address"
assert not would_drop(r"SRC=203\.0\.113\.1 "), "squid's own connections leave the host and are not forwarded"
out = proxy.succeed("nft list chain inet hive_egress forward")
found = re.search(r'oifname != "eth1" counter packets (\d+) bytes \d+$', out, re.M)
assert found and int(found.group(1)) > 0, out
with checked("the same holds over IPv6"):
agent.wait_until_succeeds("ping -6 -c1 -W2 fd00:98::1")
assert dial("fd00:99::2"), "a direct v6 connect works"
proxy.wait_until_succeeds(
"journalctl -k -o cat --no-pager | grep -E "
"'hive-egress-would-drop IN=eth1 OUT=eth2 .*SRC=fd00:0*98:(0+:)*0*3 DST=fd00:0*99:(0+:)*0*2 .*PROTO=TCP .*DPT=443 '"
)
with checked("a flow to the mesh interface is not logged and one to the uplink is"):
proxy.succeed("ip link add wg-hive type dummy && ip link set wg-hive up")
proxy.succeed("ip route add 198.51.100.0/24 dev wg-hive")
agent.execute("echo x > /dev/udp/198.51.100.7/9; echo x > /dev/udp/203.0.113.77/9")
proxy.wait_until_succeeds(
"journalctl -k -o cat --no-pager | grep -E "
"'hive-egress-would-drop IN=eth1 OUT=eth2 .*DST=203\\.0\\.113\\.77 '"
)
assert not would_drop(r"DST=198\.51\.100\.7 "), "the mesh interface is not egress"
with checked("one busy agent address is cut to a burst plus a trickle, the rest still leave a trace, and every flow is logged once at most"):
agent.execute("for i in $(seq 1 100); do echo x > /dev/udp/203.0.113.78/$((3000 + i)); done")
proxy.wait_until_succeeds(
"journalctl -k -o cat --no-pager | grep -E 'hive-egress-would-drop-unmetered .*DST=203\\.0\\.113\\.78 '"
)
time.sleep(3)
lines = would_drop(r"DST=203\.0\.113\.78 ")
metered = [line for line in lines if "hive-egress-would-drop IN=" in line]
unmetered = [line for line in lines if "hive-egress-would-drop-unmetered IN=" in line]
assert 1 <= len(metered) <= 30, f"100 new flows: {len(metered)} per-source lines"
assert 1 <= len(unmetered) <= 80, f"100 new flows: {len(unmetered)} unmetered lines"
ports = [re.search(r"DPT=(\d+) ", line).group(1) for line in lines]
assert len(ports) == len(set(ports)), "a flow was logged twice"
out = proxy.succeed("nft list chain inet hive_egress forward")
found = re.search(r'oifname != "eth1" counter packets (\d+) bytes \d+$', out, re.M)
assert found and int(found.group(1)) >= 100, out
'';
}