openvibers/openvibe.link Commits

Enforce worker.egress public and openvibe-only with a filtered job veth (#24)

OP OpenVibers committed 4 Oct 2026 · fb0d6e67f6 · 9 files, +581 −31

local checks ✓ 1/1 · CI ✓ 9/9 · review SHIP by DeepSeek harness

https://github.com/OpenVibers/OpenVibe.Node/pull/24

Changes

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 3d43a58..d7f119c 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -188,7 +188,7 @@ Otherwise it advertises no class and refuses every well-formed job with `class n
| direction | type | fields |
|------------------|----------------|-------------------------------------------------------------------------------------------|
--| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, the default)} |
++| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`: no network whatever the Node's `worker.egress`; absent: `worker.egress` applies, `none` by default)} |
| server → device | `job_cancel` | `id` |
| server → device | `job_exit_ack` | `id` |
| device → server | `job_started` | `id`, `started_ms` |
ddiff --git a/docs/worker.md b/docs/worker.md
iindex da1d787..2802fcb 100644
---- a/docs/worker.md
++++ b/docs/worker.md
@@@ -84,11 +84,20 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
(`RLIMIT_AS`, always set: there is no "none"; a runtime that reserves a large address space up front, such as
Node.js, the JVM or race-instrumented Go, needs a large value, e.g. 64 TiB for the last). `mem_bytes` is the job cgroup's `memory.max` (swap off: `memory.swap.max` is set to 0 and read back; without swap accounting, which makes that
control, the probe and each job are refused unless the host has no swap at all; an OOM kills the whole job).
--- `egress`: what a job may reach on the network. `none` (the default) is the only policy enforced: a network
-- namespace with nothing but a loopback that is down. `public` (internet but no private, link-local or Node
-- addresses) and `openvibe-only` (the platform's endpoints only) need an egress proxy in the job's namespace that is
-- not built yet (follow-up): with either, the probe fails and the worker stays off, never a silent `none` nor an
-- open network.
++- `egress`: the most a job may reach on the network; all three policies are enforced (see "Egress" below). `none`
++ (the default): a network namespace with nothing but a loopback that is down. `public`: IPv4 to public addresses
++ only, through a veth the Node NATs; private (`10/8`, `172.16/12`, `192.168/16`), shared (`100.64/10`, most
++ WireGuard meshes), loopback, link-local (`169.254/16`, cloud metadata included), multicast, documentation and
++ reserved ranges, `egress_deny`, the host itself (every address it holds) and anything not routed through the
++ host's default route (a WireGuard peer, a container bridge) are refused; there is no IPv6 (`fe80::/10` included).
++ `openvibe-only`: `public`, further restricted to `egress_allow`. A job runs under the **stricter** of its own `net`
++ and this: a job naming `net: "deny"` gets `none` whatever `egress` says; a job without `net` gets `egress`. A policy
++ the host cannot enforce fails the probe and keeps the worker off: never a silent `none` nor an open network.
++- `egress_allow` (with `openvibe-only` only, and required by it): the IPv4 CIDRs a job may reach, the OpenVibe
++ network's own ranges. They are configured, not guessed nor resolved from names: the Node trusts no DNS answer for
++ them. The refused ranges above still win over an entry that overlaps them.
++- `egress_deny` (with `public` or `openvibe-only`): more IPv4 CIDRs to refuse, on top of those always refused: the
++ Node's own cell or WireGuard peers when they hold public addresses.
The classes `function` and `code` — advertised as the reserved Fabric capabilities `worker:function` and `worker:code` in
`status.capabilities.worker` (the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — are each offered only when `enabled` is true, the OS is Linux
@@@ -97,8 +106,11 @@ the way a job is started and checks every control from the host: the sandbox was
mount, network, PID, IPC, UTS and cgroup namespaces of its own, `NoNewPrivs` and a seccomp filter; nothing in its
root but `/tmp`, `/proc` and the device nodes is mounted writable; it is in its own cgroup with every controller
(cpu, memory, pids, io) and every block device holds its `io.max`; its rlimits (`RLIMIT_AS` included) are set; its
--root holds none of `/var`, `/etc/shadow`, `/etc/ssl/private`, `/etc/pki/tls/private` and the Node's directories; its CPU limit can be set and its usage read; `egress` is `none`; no function's `artifact_dir`
--is refused (above). If any check fails the Node logs why (`worker off: …`) and refuses every job,
++root holds none of `/var`, `/etc/shadow`, `/etc/ssl/private`, `/etc/pki/tls/private` and the Node's directories; its CPU limit can be set and its usage read; no function's `artifact_dir`
++is refused (above). It does so under `none` and, when `egress` is `public` or `openvibe-only`, again under that
++policy, whose veth must then stand as the process's only interface up, with the default route through it and the
++Node's nftables table written (and first: the Node runs as root, `ip`, `nsenter` and `nft` are installed, IPv4
++forwarding is on). If any check fails the Node logs why (`worker off: …`) and refuses every job,
exactly as with the worker off: there is no weaker mode. A job whose sandbox fails at run time ends `failed` without
its function having run.
@@@ -159,7 +171,8 @@ In six lines:
`artifact_dir` made as it starts and a size-capped `/tmp` of its own: none of the Node's files, its credential or
its control socket, nor the host's private keys.
4. The network namespace has only a loopback that is down: no IP network, not even to the Node or `localhost`
-- (`egress` `none`; the other policies are not enforced yet, so they keep the worker off).
++ (`egress` `none`, or a job naming `net: "deny"`). Under `public` or `openvibe-only` it also holds one veth to the
++ host, through which only what that policy allows leaves (see "Egress"); the loopback stays down.
5. Nothing of the Node's environment reaches it.
6. CPU, memory, processes and disk I/O are enforced by its cgroup v2 and rlimits; ttl, wall, CPU and output also by a
per-job watchdog that SIGKILLs the whole group; `RLIMIT_CPU` backs the CPU cap per process.
@@@ -207,8 +220,32 @@ What it does **not** protect against:
- **The `args` and the result are trusted to the function**: the worker passes the server's JSON unchanged; validating
it is the function's job.
--Not built yet (follow-ups): the `public` and `openvibe-only` egress policies (an egress proxy reached through a veth
--or a socket bound into the job's namespace, filtering by destination).
++### Egress
++
++How `public` and `openvibe-only` are enforced (`internal/worker/egress_linux.go`):
++
++- **A veth per job.** Before the job starts, the Node makes a veth pair `ovw<n>`/`ovw<n>j` and gives the two ends a
++ `/30` of `169.254.240.0/20` (link-local: an address no job may reach, so no job reaches another's). Once the
++ sandbox stands, and before the function executes, it moves `ovw<n>j` into the job's network namespace, addresses it
++ with a default route through `ovw<n>` and reads the namespace back: that interface alone up (the loopback stays
++ down), that route and its `/30` alone. The job cannot change any of it: it holds no capability. Deleted when the
++ job ends.
++- **NAT and filtering in the host's namespace**, in the Node's own nftables table `inet openvibe_worker`, written whole
++ in one transaction each time a job's veth is made (a firewall reload that dropped it is undone by the next job;
++ other tables are left alone, and a firewall that drops forwarded traffic still does). What a job sends is
++ masqueraded out of the host's default route. Refused, with a TCP reset (otherwise ICMP administratively prohibited),
++ so a job sees "connection refused" at once: anything addressed to the host itself; IPv6; anything not leaving by an
++ interface of the host's IPv4 default route; the refused ranges listed under `egress`; `egress_deny`; and, under
++ `openvibe-only`, anything outside `egress_allow`. Nothing new reaches a job; only replies do.
++- **Needs**: the Node running as root (`run_as`), `ip` and `nsenter` (iproute2, util-linux) and `nft` (nftables) in
++ `/usr/sbin`, `/sbin`, `/usr/bin` or `/bin`, IPv4 forwarding on (`sysctl net.ipv4.ip_forward=1`: the Node does not
++ turn it on, which would make the host a router for every interface; the owner does) and an IPv4 default route.
++ Without one of them the probe fails and so does every job.
++- **No DNS.** The Node runs no resolver for jobs, and the host's (`127.0.0.53` on most systems) is out of a job's
++ reach. A job resolves names only through a resolver its policy lets it reach by address (any public one under
++ `public`; one listed in `egress_allow` under `openvibe-only`), or connects by address.
++- **`openvibe-only` is configured, not guessed.** No authoritative list of the OpenVibe network's ranges exists in
++ this repository or OpenVibe.Contracts, so `egress_allow` must name them; without it the config does not load.
## Refusals and exit reasons
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex bfff9b4..ccfbf1b 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -15,6 +15,7 @@ import (
"errors"
"fmt"
"io/fs"
++ "net/netip"
"os"
"path/filepath"
"regexp"
@@@ -123,9 +124,18 @@ type WorkerConfig struct {
// job's private root holds: for development only.
AllowSameUser bool `json:"allow_same_user,omitempty"`
Caps WorkerCaps `json:"caps,omitzero"`
-- // Egress is what a job may reach on the network: none (the default; a network namespace with only a loopback that
-- // is down), public or openvibe-only. Only none is enforced yet: with another value the worker stays off.
++ // Egress is the most a job may reach on the network, all three enforced (docs/worker.md): none (the default; a
++ // network namespace with only a loopback that is down), public (IPv4 to public addresses, through a veth the Node
++ // NATs; private, shared, link-local and reserved ranges, EgressDeny and the host itself refused) or openvibe-only
++ // (public restricted to EgressAllow). A job runs under the stricter of its own net and this: net "deny" is none
++ // whatever Egress says. A policy the host cannot enforce fails the probe, so the worker stays off.
Egress string `json:"egress,omitempty"`
++ // EgressAllow is the IPv4 CIDRs an openvibe-only job may reach (the OpenVibe network's own ranges, configured
++ // rather than guessed); required with openvibe-only and refused with any other policy.
++ EgressAllow []string `json:"egress_allow,omitempty"`
++ // EgressDeny is more IPv4 CIDRs a public or openvibe-only job is refused (the Node's cell, WireGuard peers with
++ // public addresses), on top of the ranges always refused.
++ EgressDeny []string `json:"egress_deny,omitempty"`
// NodeDirs are the Node's own directories (config, credential, state, control socket), which no job may see; the
// Node sets them, the config file does not.
NodeDirs []string `json:"-"`
@@@ -392,6 +402,19 @@ func (w WorkerConfig) validate() error {
default:
return fmt.Errorf("config: worker.egress %q is not none, public or openvibe-only", w.Egress)
}
++ for key, cidrs := range map[string][]string{"egress_allow": w.EgressAllow, "egress_deny": w.EgressDeny} {
++ for _, c := range cidrs {
++ if p, err := netip.ParsePrefix(c); err != nil || !p.Addr().Is4() || p != p.Masked() {
++ return fmt.Errorf("config: worker.%s %q is not an IPv4 CIDR such as 203.0.113.0/24 (jobs have no IPv6)", key, c)
++ }
++ }
++ }
++ switch {
++ case w.Egress == EgressOpenVibeOnly && len(w.EgressAllow) == 0:
++ return errors.New("config: worker.egress openvibe-only needs worker.egress_allow, the CIDRs a job may reach")
++ case w.Egress != EgressOpenVibeOnly && len(w.EgressAllow) > 0:
++ return errors.New("config: worker.egress_allow applies to egress openvibe-only only")
++ }
if w.RunAs != nil && (w.RunAs.UID == 0 || w.RunAs.GID == 0) {
return errors.New("config: worker.run_as must not be root (uid or gid 0)")
}
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
iindex ae413c4..838333c 100644
---- a/internal/config/config_test.go
++++ b/internal/config/config_test.go
@@@ -121,3 +121,30 @@ func TestWorkerEntryClass(t *testing.T) {
t.Fatal("a function entry with no class and one with class function were both accepted")
}
}
++
++// TestWorkerEgress: the three policies load; openvibe-only needs its allowlist, which no other policy takes; both
++// lists hold IPv4 CIDRs only.
++func TestWorkerEgress(t *testing.T) {
++ ok := []WorkerConfig{{}, {Egress: EgressNone}, {Egress: EgressPublic, EgressDeny: []string{"203.0.113.0/24"}},
++ {Egress: EgressOpenVibeOnly, EgressAllow: []string{"198.51.100.7/32"}, EgressDeny: []string{"198.51.100.0/24"}}}
++ for _, w := range ok {
++ c := Default()
++ c.Worker = w
++ if err := c.Validate(); err != nil {
++ t.Fatalf("%+v: %v", w, err)
++ }
++ }
++ bad := []WorkerConfig{{Egress: "open"}, {Egress: EgressOpenVibeOnly},
++ {Egress: EgressPublic, EgressAllow: []string{"198.51.100.0/24"}}, {EgressAllow: []string{"198.51.100.0/24"}},
++ {Egress: EgressOpenVibeOnly, EgressAllow: []string{"openvibe.network"}},
++ {Egress: EgressOpenVibeOnly, EgressAllow: []string{"2001:db8::/32"}},
++ {Egress: EgressOpenVibeOnly, EgressAllow: []string{"198.51.100.7/24"}},
++ {Egress: EgressPublic, EgressDeny: []string{"10.0.0.1"}}}
++ for _, w := range bad {
++ c := Default()
++ c.Worker = w
++ if err := c.Validate(); err == nil {
++ t.Fatalf("%+v was accepted", w)
++ }
++ }
++}
ddiff --git a/internal/worker/egress_linux.go b/internal/worker/egress_linux.go
nnew file mode 100644
iindex 0000000..02b1b4d
---- /dev/null
++++ b/internal/worker/egress_linux.go
@@@ -0,0 +1,298 @@
++//go:build linux
++
++package worker
++
++import (
++ "bytes"
++ "encoding/binary"
++ "errors"
++ "fmt"
++ "net/netip"
++ "os"
++ "os/exec"
++ "path/filepath"
++ "regexp"
++ "slices"
++ "strconv"
++ "strings"
++ "sync"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/config"
++)
++
++// A job under worker.egress public or openvibe-only (docs/worker.md, "Egress") gets, in its network namespace, one end
++// of a veth pair whose other end, ovw<n>, stays in the host's; the two ends hold a /30 of vethPool and the job's
++// default route goes through the host's end. The nftables table openvibe_worker, in the host's namespace, masquerades
++// what a job sends and refuses (a TCP reset, otherwise ICMP administratively prohibited) everything else: anything to
++// the host itself, IPv6, a route other than the host's default one, the nonPublic ranges, worker.egress_deny and, for
++// openvibe-only, whatever lies outside worker.egress_allow. Nothing new reaches a job, only replies. The table is
++// written whole, in one transaction, each time a job's veth is made, so a firewall reload that dropped it is undone by
++// the next job. A job naming net "deny", or under egress none, gets no veth: its loopback, down, is all it has.
++
++const (
++ egressTable = "openvibe_worker"
++ vethPrefix = "ovw"
++ vethPool = "169.254.240.0/20" // link-local: an address a job is refused, so a job reaches no other's veth
++ vethMax = 1024 // the /30s in vethPool
++)
++
++// nonPublic is the IPv4 ranges a job under public or openvibe-only is always refused: this network, private (RFC
++// 1918), shared (100.64/10: carrier NAT and most WireGuard meshes), loopback, link-local (cloud metadata included),
++// IETF protocol assignments, documentation, the 6to4 relay, benchmarking, multicast and reserved.
++var nonPublic = []string{"0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8", "169.254.0.0/16", "172.16.0.0/12",
++ "192.0.0.0/24", "192.0.2.0/24", "192.88.99.0/24", "192.168.0.0/16", "198.18.0.0/15", "198.51.100.0/24",
++ "203.0.113.0/24", "224.0.0.0/4", "240.0.0.0/4"}
++
++// egressToolDirs is where the Node finds ip, nsenter and nft, which it runs as root (not from $PATH); tests change it.
++var egressToolDirs = []string{"/usr/sbin", "/sbin", "/usr/bin", "/bin"}
++
++// errEgressTool: a program the egress policies need is not installed.
++var errEgressTool = errors.New("not installed")
++
++var (
++ vethMu sync.Mutex
++ vethUsed = map[int]bool{}
++ vethNext int // round robin: a pair just deleted is not made again at once
++)
++
++type egressTools struct{ ip, nsenter, nft string }
++
++// egressReady refuses an egress policy the host cannot enforce: the Node must run as root, find ip, nsenter and nft,
++// and the host must forward IPv4 (net.ipv4.ip_forward=1, which the Node does not set: it is the owner's choice).
++func egressReady() (egressTools, error) {
++ var t egressTools
++ if os.Geteuid() != 0 {
++ return t, errors.New("needs the Node to run as root (worker.run_as)")
++ }
++ for _, x := range []struct {
++ name string
++ path *string
++ }{{"ip", &t.ip}, {"nsenter", &t.nsenter}, {"nft", &t.nft}} {
++ for _, d := range egressToolDirs {
++ p := filepath.Join(d, x.name)
++ if st, err := os.Stat(p); err == nil && st.Mode().IsRegular() && st.Mode()&0o111 != 0 {
++ *x.path = p
++ break
++ }
++ }
++ if *x.path == "" {
++ return t, fmt.Errorf("needs %s (%s): %w", x.name, strings.Join(egressToolDirs, ", "), errEgressTool)
++ }
++ }
++ if b, err := os.ReadFile("/proc/sys/net/ipv4/ip_forward"); err != nil || strings.TrimSpace(string(b)) != "1" {
++ return t, errors.New("needs IPv4 forwarding on the host (sysctl net.ipv4.ip_forward=1)")
++ }
++ return t, nil
++}
++
++// egressNet is a job's veth pair: made before the job starts, its job end moved into the job's namespace once the
++// sandbox stands.
++type egressNet struct {
++ tools egressTools
++ n int // ovw<n>; -1 once released
++ made bool // the pair exists (this Node made it)
++ host, job netip.Addr
++}
++
++func (e *egressNet) hostIf() string { return vethPrefix + strconv.Itoa(e.n) }
++func (e *egressNet) jobIf() string { return e.hostIf() + "j" }
++
++// newEgress writes the table for cfg and makes a job's veth pair, its host end addressed and up. Every error wraps
++// what the host lacks, so the probe and the job fail on it.
++func newEgress(cfg config.WorkerConfig) (*egressNet, error) {
++ t, err := egressReady()
++ if err != nil {
++ return nil, err
++ }
++ up, err := uplinks(t.ip)
++ if err != nil {
++ return nil, err
++ }
++ if err := runTool(ruleset(cfg, up), t.nft, "-f", "-"); err != nil {
++ return nil, fmt.Errorf("writing the nftables table: %w", err)
++ }
++ e := &egressNet{tools: t, n: -1}
++ vethMu.Lock()
++ for i := range vethMax {
++ if n := (vethNext + i) % vethMax; !vethUsed[n] {
++ e.n, vethUsed[n], vethNext = n, true, n+1
++ break
++ }
++ }
++ vethMu.Unlock()
++ if e.n < 0 {
++ return nil, errors.New("every job veth is in use")
++ }
++ b := netip.MustParsePrefix(vethPool).Addr().As4()
++ base := binary.BigEndian.Uint32(b[:]) + uint32(4*e.n)
++ e.host, e.job = addr4(base+1), addr4(base+2)
++ if err := runTool("", t.ip, "link", "add", e.hostIf(), "type", "veth", "peer", "name", e.jobIf()); err != nil {
++ e.close()
++ return nil, err
++ }
++ e.made = true
++ batch := fmt.Sprintf("addr add %s/30 dev %s\nlink set %s up\n", e.host, e.hostIf(), e.hostIf())
++ if err := runTool(batch, t.ip, "-batch", "-"); err != nil {
++ e.close()
++ return nil, err
++ }
++ return e, nil
++}
++
++func addr4(v uint32) netip.Addr {
++ var b [4]byte
++ binary.BigEndian.PutUint32(b[:], v)
++ return netip.AddrFrom4(b)
++}
++
++// attach moves the job end of the pair into the network namespace of pid (the sandbox init, waiting to execute the
++// function), addresses it with a default route through the host end, then reads the namespace back: the job end is
++// its only interface up (the loopback stays down) and its routes are that one and its /30's.
++func (e *egressNet) attach(pid int) error {
++ if err := runTool("", e.tools.ip, "link", "set", e.jobIf(), "netns", strconv.Itoa(pid)); err != nil {
++ return err
++ }
++ in := []string{"--net=/proc/" + strconv.Itoa(pid) + "/ns/net", "--", e.tools.ip}
++ batch := fmt.Sprintf("addr add %s/30 dev %s\nlink set %s up\nroute add default via %s dev %s\n",
++ e.job, e.jobIf(), e.jobIf(), e.host, e.jobIf())
++ if err := runTool(batch, e.tools.nsenter, append(in, "-batch", "-")...); err != nil {
++ return err
++ }
++ links, err := toolOutput(e.tools.nsenter, append(in, "-o", "link", "show", "up")...)
++ if err != nil {
++ return err
++ }
++ var up []string
++ for _, l := range strings.Split(strings.TrimSpace(links), "\n") {
++ if f := strings.Fields(l); len(f) > 1 {
++ name, _, _ := strings.Cut(strings.TrimSuffix(f[1], ":"), "@")
++ up = append(up, name)
++ }
++ }
++ if !slices.Equal(up, []string{e.jobIf()}) {
++ return fmt.Errorf("the job's interfaces up are %v, not %s alone", up, e.jobIf())
++ }
++ routes, err := toolOutput(e.tools.nsenter, append(in, "-4", "route", "show")...)
++ if err != nil {
++ return err
++ }
++ want := []string{"default via " + e.host.String() + " dev " + e.jobIf(),
++ netip.PrefixFrom(e.job, 30).Masked().String() + " dev " + e.jobIf()}
++ got := strings.Split(strings.TrimSpace(routes), "\n")
++ if len(got) != len(want) || !slices.ContainsFunc(got, func(l string) bool { return routeIs(l, want[0]) }) ||
++ !slices.ContainsFunc(got, func(l string) bool { return routeIs(l, want[1]) }) {
++ return fmt.Errorf("the job's routes are %q, not %q", got, want)
++ }
++ return nil
++}
++
++func routeIs(line, route string) bool {
++ line = strings.TrimSpace(line)
++ return line == route || strings.HasPrefix(line, route+" ")
++}
++
++// close deletes the pair (gone already if the job's namespace is) and releases its number.
++func (e *egressNet) close() {
++ if e.n < 0 {
++ return
++ }
++ if e.made {
++ _ = runTool("", e.tools.ip, "link", "del", e.hostIf())
++ }
++ vethMu.Lock()
++ delete(vethUsed, e.n)
++ vethMu.Unlock()
++ e.n, e.made = -1, false
++}
++
++// ifName is an interface name nft may hold in a string.
++var ifName = regexp.MustCompile(`^[A-Za-z0-9_.:-]{1,15}$`)
++
++// uplinks are the interfaces of the host's IPv4 default routes, the only ones a job's traffic may leave by (so a
++// WireGuard peer, a container bridge or a LAN reached by its own route is refused even at a public address).
++func uplinks(ip string) ([]string, error) {
++ out, err := toolOutput(ip, "-4", "route", "show", "default")
++ if err != nil {
++ return nil, err
++ }
++ var devs []string
++ f := strings.Fields(out)
++ for i := 0; i+1 < len(f); i++ {
++ if d := f[i+1]; f[i] == "dev" && !slices.Contains(devs, d) {
++ if !ifName.MatchString(d) || strings.HasPrefix(d, vethPrefix) {
++ return nil, fmt.Errorf("the default route's interface %q is not usable", d)
++ }
++ devs = append(devs, d)
++ }
++ }
++ if len(devs) == 0 {
++ return nil, errors.New("the host has no IPv4 default route")
++ }
++ return devs, nil
++}
++
++// ruleset is the nft script that replaces the table openvibe_worker for cfg, jobs leaving by uplinks.
++func ruleset(cfg config.WorkerConfig, uplinks []string) string {
++ var b strings.Builder
++ w := func(format string, a ...any) { fmt.Fprintf(&b, format+"\n", a...) }
++ w("table inet %s", egressTable)
++ w("delete table inet %s", egressTable)
++ w("table inet %s {", egressTable)
++ w("\tchain refuse {")
++ w("\t\tmeta l4proto tcp reject with tcp reset")
++ w("\t\treject with icmpx admin-prohibited")
++ w("\t}")
++ w("\tchain job {")
++ w("\t\tmeta nfproto != ipv4 goto refuse")
++ w("\t\toifname != { \"%s\" } goto refuse", strings.Join(uplinks, "\", \""))
++ for _, c := range append(slices.Clone(nonPublic), cfg.EgressDeny...) {
++ w("\t\tip daddr %s goto refuse", c)
++ }
++ if cfg.Egress == config.EgressOpenVibeOnly {
++ for _, c := range cfg.EgressAllow {
++ w("\t\tip daddr %s accept", c)
++ }
++ w("\t\tgoto refuse")
++ }
++ w("\t\taccept")
++ w("\t}")
++ w("\tchain forward {")
++ w("\t\ttype filter hook forward priority -10; policy accept;")
++ w("\t\tiifname \"%s*\" goto job", vethPrefix)
++ w("\t\toifname \"%s*\" ct state established,related accept", vethPrefix)
++ w("\t\toifname \"%s*\" goto refuse", vethPrefix)
++ w("\t}")
++ w("\tchain input {")
++ w("\t\ttype filter hook input priority -10; policy accept;")
++ w("\t\tiifname \"%s*\" goto refuse", vethPrefix)
++ w("\t}")
++ w("\tchain postrouting {")
++ w("\t\ttype nat hook postrouting priority 100; policy accept;")
++ w("\t\tip saddr %s oifname != \"%s*\" masquerade", vethPool, vethPrefix)
++ w("\t}")
++ w("}")
++ return b.String()
++}
++
++// runTool runs one of egressTools with stdin, failing with its output.
++func runTool(stdin, path string, args ...string) error {
++ cmd := exec.Command(path, args...)
++ cmd.Env = []string{"PATH=/usr/sbin:/usr/bin:/sbin:/bin"}
++ if stdin != "" {
++ cmd.Stdin = strings.NewReader(stdin)
++ }
++ if out, err := cmd.CombinedOutput(); err != nil {
++ return fmt.Errorf("%s %s: %w: %s", filepath.Base(path), strings.Join(args, " "), err, bytes.TrimSpace(out))
++ }
++ return nil
++}
++
++func toolOutput(path string, args ...string) (string, error) {
++ cmd := exec.Command(path, args...)
++ cmd.Env = []string{"PATH=/usr/sbin:/usr/bin:/sbin:/bin"}
++ out, err := cmd.Output()
++ if err != nil {
++ return "", fmt.Errorf("%s %s: %w", filepath.Base(path), strings.Join(args, " "), err)
++ }
++ return string(out), nil
++}
ddiff --git a/internal/worker/run_linux_test.go b/internal/worker/run_linux_test.go
iindex 40043d8..0ead914 100644
---- a/internal/worker/run_linux_test.go
++++ b/internal/worker/run_linux_test.go
@@@ -75,6 +75,32 @@ func TestHelperProcess(t *testing.T) {
_, derr := net.DialTimeout("tcp", "192.0.2.1:9", time.Second)
_ = json.NewEncoder(result).Encode(map[string]any{"env": os.Environ(), "cwd": cwd, "pid": os.Getpid(),
"uid": os.Getuid(), "loopback": fmt.Sprint(lerr), "dial": fmt.Sprint(derr)})
++ case "egress": // how each address answers a TCP connect: connected, refused, unreachable or the error
++ got := map[string]string{}
++ var mu sync.Mutex
++ var wg sync.WaitGroup
++ for _, a := range []string{"10.0.0.1:80", "169.254.169.254:80", "169.254.240.1:22", "1.1.1.1:53"} {
++ wg.Add(1)
++ go func() {
++ defer wg.Done()
++ c, err := net.DialTimeout("tcp", a, 3*time.Second)
++ r := fmt.Sprint(err)
++ switch {
++ case err == nil:
++ c.Close()
++ r = "connected"
++ case errors.Is(err, syscall.ECONNREFUSED):
++ r = "refused"
++ case errors.Is(err, syscall.ENETUNREACH):
++ r = "unreachable"
++ }
++ mu.Lock()
++ got[a] = r
++ mu.Unlock()
++ }()
++ }
++ wg.Wait()
++ _ = json.NewEncoder(result).Encode(got)
case "escape": // every attempt must fail but the write to /tmp; the result is each attempt's error
node := os.Getenv("OPENVIBE_TEST_NODE_DIR")
try := map[string]error{}
@@@ -1208,16 +1234,23 @@ func TestForkBomb(t *testing.T) {
}
}
--// TestEgressPolicy: only egress none is enforced; with another policy the probe fails (the class is not advertised)
--// and a job fails without running.
--func TestEgressPolicy(t *testing.T) {
++// TestEgressUnenforceable: an egress policy the host cannot enforce (here ip, nsenter and nft are not found, or the
++// Node is not root) fails the probe, so the class is not advertised, and a job fails without running: never a silent
++// none nor an open network.
++func TestEgressUnenforceable(t *testing.T) {
++ dirs := egressToolDirs
++ egressToolDirs = []string{t.TempDir()}
++ t.Cleanup(func() { egressToolDirs = dirs })
for _, e := range []string{config.EgressPublic, config.EgressOpenVibeOnly} {
s := &sink{}
cfg := helperConfig(t, config.WorkerCaps{}, "env")
cfg.Egress = e
++ if e == config.EgressOpenVibeOnly {
++ cfg.EgressAllow = []string{"9.9.9.9/32"}
++ }
w := New(cfg, s.send, quiet())
t.Cleanup(w.Close)
-- if err := w.Probe(); err == nil || !strings.Contains(err.Error(), "egress") {
++ if err := w.Probe(); err == nil || !strings.Contains(err.Error(), "worker.egress "+e) {
t.Fatalf("%s: probe %v, want refused", e, err)
}
j := testJob(1, "env")
@@@ -1228,6 +1261,83 @@ func TestEgressPolicy(t *testing.T) {
}
}
++// TestEgress: under none a job has no route at all; under public it has one, but 10.0.0.1, 169.254.169.254 (cloud
++// metadata) and the host's end of its veth are refused by the Node's table (refused, not unreachable nor a timeout);
++// under openvibe-only an unlisted public host is refused too; a job naming net "deny" under public gets none. It needs
++// root, ip, nsenter, nft and IPv4 forwarding: it skips without the programs even where OPENVIBE_WORKER_TESTS=require.
++func TestEgress(t *testing.T) {
++ none := map[string]string{"10.0.0.1:80": "unreachable", "169.254.169.254:80": "unreachable",
++ "169.254.240.1:22": "unreachable", "1.1.1.1:53": "unreachable"}
++ refused := map[string]string{"10.0.0.1:80": "refused", "169.254.169.254:80": "refused", "169.254.240.1:22": "refused"}
++ for _, c := range []struct {
++ name, egress string
++ allow []string
++ net string
++ want map[string]string
++ }{
++ {"none", "", nil, "", none},
++ {"public", config.EgressPublic, nil, "", refused},
++ {"openvibe-only", config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}, "",
++ map[string]string{"10.0.0.1:80": "refused", "169.254.169.254:80": "refused", "1.1.1.1:53": "refused"}},
++ {"deny under public", config.EgressPublic, nil, protocol.NetDeny, none},
++ } {
++ t.Run(c.name, func(t *testing.T) {
++ cfg := helperConfig(t, config.WorkerCaps{}, "egress")
++ cfg.Egress, cfg.EgressAllow = c.egress, c.allow
++ if c.egress != "" {
++ if _, err := egressReady(); errors.Is(err, errEgressTool) {
++ t.Skip(err)
++ }
++ }
++ w, s := newRunWorkerConfig(t, cfg)
++ j := testJob(1, "egress")
++ j.Net = c.net
++ run(t, w, j)
++ ex := s.exit(t, j.ID)
++ var got map[string]string
++ if err := json.Unmarshal(ex.Result, &got); err != nil || ex.Reason != protocol.ExitExited {
++ t.Fatalf("%+v %s: %v", ex, ex.Result, err)
++ }
++ for a, want := range c.want {
++ if got[a] != want {
++ t.Errorf("%s: %s, want %s (all: %v)", a, got[a], want, got)
++ }
++ }
++ })
++ }
++}
++
++// TestEgressRuleset: the table refuses IPv6, other routes than the default one, the private, shared, link-local and
++// reserved ranges and egress_deny before anything is accepted; openvibe-only then accepts egress_allow alone.
++func TestEgressRuleset(t *testing.T) {
++ cfg := config.WorkerConfig{Egress: config.EgressPublic, EgressDeny: []string{"203.0.113.0/24", "198.51.100.0/24"}}
++ pub := ruleset(cfg, []string{"eth0", "wlan0"})
++ cfg.Egress, cfg.EgressAllow = config.EgressOpenVibeOnly, []string{"9.9.9.9/32"}
++ ovo := ruleset(cfg, []string{"eth0"})
++ for _, r := range []string{pub, ovo} {
++ job := r[strings.Index(r, "chain job {"):]
++ job = job[:strings.Index(job, "\n\t}")]
++ accept := strings.Index(job, "accept")
++ for _, want := range []string{"meta nfproto != ipv4 goto refuse", "ip daddr 10.0.0.0/8 goto refuse",
++ "ip daddr 100.64.0.0/10 goto refuse", "ip daddr 169.254.0.0/16 goto refuse",
++ "ip daddr 172.16.0.0/12 goto refuse", "ip daddr 192.168.0.0/16 goto refuse",
++ "ip daddr 203.0.113.0/24 goto refuse", "ip daddr 198.51.100.0/24 goto refuse"} {
++ if i := strings.Index(job, want); i < 0 || i > accept {
++ t.Fatalf("%q missing or after the first accept in\n%s", want, job)
++ }
++ }
++ if !strings.Contains(r, "iifname \"ovw*\" goto refuse") || !strings.Contains(r, "delete table inet openvibe_worker") {
++ t.Fatalf("the host itself is not refused, or the table is not replaced whole:\n%s", r)
++ }
++ }
++ if !strings.Contains(pub, `oifname != { "eth0", "wlan0" } goto refuse`) || strings.Contains(pub, "9.9.9.9") {
++ t.Fatalf("public:\n%s", pub)
++ }
++ if !strings.Contains(ovo, "ip daddr 9.9.9.9/32 accept\n\t\tgoto refuse\n\t\taccept") {
++ t.Fatalf("openvibe-only does not refuse what egress_allow does not list:\n%s", ovo)
++ }
++}
++
// TestRootPlan: the private root holds the device nodes; no bind is, holds or lies in /etc/ssl, /etc/pki (whose CA
// certificates are copied, not bound), the Node's directories nor /var; a system path holding a Node directory is
// refused.
ddiff --git a/internal/worker/worker.go b/internal/worker/worker.go
iindex 68e1252..52acc6f 100644
---- a/internal/worker/worker.go
++++ b/internal/worker/worker.go
@@@ -60,6 +60,17 @@ type plan struct {
root string // the host directory (empty, the job's) its private root is mounted on
fn config.FunctionConfig // zero for the probe
limits protocol.JobLimits
++ egress string // the job's network policy (egressFor); "" is none
++}
++
++// egressFor is the network policy a job runs under: the stricter of its net and the host's worker.egress. A job
++// naming net "deny" (the only value it may name) runs under none whatever the host allows; without net, the host's
++// policy applies.
++func egressFor(host, net string) string {
++ if net != "" || host == config.EgressNone {
++ return ""
++ }
++ return host
}
type job struct {
@@@ -391,7 +402,7 @@ func (w *Worker) execute(jb *job) protocol.JobExit {
cmd.Dir, cmd.Env = work, jobEnv(jb.fn, work, id)
cmd.Stdin = bytes.NewReader(append(append([]byte(nil), jb.req.Args...), '\n'))
cmd.Stdout, cmd.ExtraFiles = outW, []*os.File{resW}
-- sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits})
++ sb, err := w.isolate(cmd, w.cfg, plan{root: dir, fn: jb.fn, limits: jb.limits, egress: egressFor(w.cfg.Egress, jb.req.Net)})
if err != nil {
outW.Close()
resW.Close()
ddiff --git a/internal/worker/worker_linux.go b/internal/worker/worker_linux.go
iindex 133bd7d..60df46b 100644
---- a/internal/worker/worker_linux.go
++++ b/internal/worker/worker_linux.go
@@@ -29,8 +29,8 @@ var pageSize = int64(os.Getpagesize())
const capSysAdmin = 21
// isolate turns cmd into the start of a sandboxed job (see sandbox_linux.go): the Node's own executable re-run as the
--// sandbox init, in its own process group, in new user, mount, network (a loopback that is down), PID, IPC, UTS and
--// cgroup namespaces, inside the job's cgroup v2 (made here), with PR_SET_PDEATHSIG SIGKILL. It runs as worker.run_as
++// sandbox init, in its own process group, in new user, mount, network (a loopback that is down, and under egress public
++// or openvibe-only a veth, egress_linux.go), PID, IPC, UTS and cgroup namespaces, inside the job's cgroup v2 (made here), with PR_SET_PDEATHSIG SIGKILL. It runs as worker.run_as
// (no supplementary groups; the Node must be root) or, with allow_same_user, as the Node's own non-root user. cmd's
// path, args, env and dir become the function's, run once the sandbox stands; cmd.ExtraFiles holds at most fd 3. The
// function's artifact directory is opened here and copied by the init (stageArtifact).
@@@ -97,6 +97,11 @@ func isolate(cmd *exec.Cmd, cfg config.WorkerConfig, p plan) (*sandbox, error) {
if art != nil {
files = append(files, art) // artifactFD
}
++ if p.egress != "" {
++ if sb.egress, err = newEgress(cfg); err != nil {
++ return fail(fmt.Errorf("worker.egress %s: %w", p.egress, err))
++ }
++ }
if sb.cg, err = newJobCgroup(caps, p.limits); err != nil {
return fail(err)
}
@@@ -131,12 +136,9 @@ func specFile(spec sandboxSpec) (*os.File, error) {
return f, nil
}
--// policy refuses what the sandbox cannot enforce: an egress other than none, an architecture without a seccomp
--// allowlist.
++// policy refuses what the sandbox cannot enforce: an architecture without a seccomp allowlist. (An egress policy the
++// host cannot enforce fails in newEgress.)
func policy(cfg config.WorkerConfig) error {
-- if e := cfg.Egress; e != "" && e != config.EgressNone {
-- return fmt.Errorf("worker.egress %s is not enforced yet (only none is)", e)
-- }
_, err := seccompFilter()
return err
}
@@@ -170,9 +172,10 @@ func jobAttr(cfg config.WorkerConfig) (*syscall.SysProcAttr, error) {
}
// sandbox is a job's sandbox as the Node holds it: the status pipe of its init, the pipe that lets it execute the
--// function, the descriptors the child inherits (closed once it started) and its cgroup.
++// function, the descriptors the child inherits (closed once it started), its cgroup and its veth (nil under none).
type sandbox struct {
cg *jobCgroup
++ egress *egressNet
status *os.File
goW *os.File
child []*os.File
@@@ -186,8 +189,8 @@ func (s *sandbox) started() {
s.child = nil
}
--// ready waits until the sandbox init stands confined, about to execute the function, and checks from the host that
--// it runs with no_new_privs and the seccomp filter, that nothing but its /tmp, /proc and device nodes is mounted
++// ready waits until the sandbox init stands confined, about to execute the function, moves the job's veth into its
++// network namespace (under egress public or openvibe-only) and checks from the host that it runs with no_new_privs and the seccomp filter, that nothing but its /tmp, /proc and device nodes is mounted
// writable and that every block device has the job's io.max. The init waits for run meanwhile, so it cannot have
// exited. An error means the function must not run.
func (s *sandbox) ready(pid int) error {
@@@ -200,6 +203,11 @@ func (s *sandbox) ready(pid int) error {
msg, _ := io.ReadAll(s.status)
return fmt.Errorf("the sandbox could not be set up: %s%s", b[:], msg)
}
++ if s.egress != nil {
++ if err := s.egress.attach(pid); err != nil {
++ return fmt.Errorf("worker.egress: the job's network: %w", err)
++ }
++ }
// The spec, the function's environment included, must not show to every host user in the init's argv.
if c, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/cmdline"); err != nil || string(c) != sandboxArg0+"\x00" {
return fmt.Errorf("the sandbox init's command line is not its name alone (%q, %v)", c, err)
@@@ -254,6 +262,9 @@ func (s *sandbox) close() {
if s.cg != nil {
s.cg.remove()
}
++ if s.egress != nil {
++ s.egress.close()
++ }
}
// probe starts /bin/sh the way a job is started and checks every control from the host: the sandbox was set up;
@@@ -261,12 +272,20 @@ func (s *sandbox) close() {
// filter; nothing but its /tmp, /proc and device nodes is mounted writable; it is in its job cgroup; its rlimits
// (RLIMIT_AS included: max_vm_bytes must be set) are set; every block device has its io.max; its root holds none of
// the Node's directories nor hiddenPaths (/var, the host's private keys); its CPU limit can be set and its usage read.
--// It also checks that no declared function's artifact directory is refused. Any failure keeps the class off: there is
--// no weaker mode.
++// It does so for every policy a job may run under: none (a job naming net "deny") and worker.egress, whose veth must
++// then stand as the only interface up with the default route through it, the Node's nftables table written. It also
++// checks that no declared function's artifact directory is refused. Any failure keeps the class off: there is no
++// weaker mode, never none in place of public nor an open network.
func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
if err := policy(cfg); err != nil {
return err
}
++ egress := egressFor(cfg.Egress, "")
++ if egress != "" {
++ if _, err := egressReady(); err != nil {
++ return fmt.Errorf("worker.egress %s: %w", egress, err)
++ }
++ }
binds, links, err := rootPlan(cfg.NodeDirs)
if err != nil {
return err
@@@ -283,6 +302,17 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
return fmt.Errorf("%s %s@%s: %w", f.EffectiveClass(), f.Name, f.Version, err)
}
}
++ if err := probeSandbox(cfg, isolate, ""); err != nil {
++ return err
++ }
++ if egress != "" {
++ return probeSandbox(cfg, isolate, egress)
++ }
++ return nil
++}
++
++// probeSandbox is probe's check of one sandbox, under egress.
++func probeSandbox(cfg config.WorkerConfig, isolate isolateFunc, egress string) error {
root, err := os.MkdirTemp("", "openvibe-probe-")
if err != nil {
return err
@@@ -296,7 +326,7 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
caps := cfg.Caps.WithDefaults()
cmd := exec.Command("/bin/sh", "-c", "read x")
cmd.Env, cmd.Dir = []string{"PATH=/usr/bin:/bin"}, "/tmp"
-- sb, err := isolate(cmd, cfg, plan{root: root, limits: protocol.JobLimits{CPUMS: 1000, MemBytes: caps.MaxMemBytes}})
++ sb, err := isolate(cmd, cfg, plan{root: root, limits: protocol.JobLimits{CPUMS: 1000, MemBytes: caps.MaxMemBytes}, egress: egress})
if err != nil {
return err
}
ddiff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go
iindex 13fa6b3..3b10939 100644
---- a/internal/worker/worker_test.go
++++ b/internal/worker/worker_test.go
@@@ -288,3 +288,17 @@ func TestUsageHelperProcess(t *testing.T) {
time.Sleep(time.Duration(args.HoldMS) * time.Millisecond)
os.Exit(0)
}
++
++// TestEgressFor: a job runs under the stricter of its net and the host's egress: net "deny" is none whatever the host
++// allows; without net, the host's policy.
++func TestEgressFor(t *testing.T) {
++ for _, c := range []struct{ host, net, want string }{
++ {"", "", ""}, {config.EgressNone, "", ""}, {config.EgressPublic, "", config.EgressPublic},
++ {config.EgressOpenVibeOnly, "", config.EgressOpenVibeOnly}, {config.EgressPublic, protocol.NetDeny, ""},
++ {config.EgressOpenVibeOnly, protocol.NetDeny, ""}, {config.EgressPublic, "allow", ""},
++ } {
++ if got := egressFor(c.host, c.net); got != c.want {
++ t.Errorf("egressFor(%q, %q) = %q, want %q", c.host, c.net, got, c.want)
++ }
++ }
++}