The firewall setting reaches the mirror machine and the store's members

The leading control plane tells the mirror over its API (PUT /v1/firewall,
kept as .firewall.json beside the models and applied by the same path unit
control planes use), and keeps the state in the store's cluster_state for
members that run no llmgw, which a root timer reads from the member's own
copy. The mirror also lets in the network of every node and control plane
the leader knows. Tried on the PoC: off, on and a network added, on every
kind of machine.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-09 08:53:17 +02:00
1 parent 5be3da505b
commit 8df78a4d41
20 files changed
+772 -34

No files matched your search

+12 -3
View File
@@ -796,10 +796,19 @@ leader 30s later with /health answering throughout; rollback did the same.
facility state from the store (it kept what it read at start).
- Retires `appendGuardIfMissing` (server init): the import does it.
### E2. Generated files from settings (L) — firewall on/off done 2026-10-06
### E2. Generated files from settings (L) — firewall on/off done 2026-10-06, every machine 2026-10-09
`firewall.enabled` in the cluster's settings reaches nodes over the agent
stream and control planes through llmgw-firewall-sync.path. Left: the mirror
machine and store members, the allowed set from membership, the other files.
stream and control planes through llmgw-firewall-sync.path. Since
2026-10-09 the mirror machine too (the leader PUTs /v1/firewall, the mirror
writes `.firewall.json`, the same path unit applies it) and store members
that run no llmgw (the leader keeps it in `cluster_state`, a root timer
reads the member's own copy with psql); `firewall_networks` reach all of
them, and the mirror lets in the network of every node and control plane
the leader knows. Tried on the PoC: off and on, and a network added, on
control planes, mirror, store member and nodes. Left: the allowed set
replaced by membership rather than added to (a network taken out of the
settings stays let in), the other files (the mirror's unit flags,
keepalived, etcd and Patroni configuration).
Every machine runs the agent (root) with roles — node, control plane, mirror,
store member — and the agent renders and applies what was written once by
+8
View File
@@ -65,6 +65,14 @@ APIs change between minors without a deprecation period.
as `llmgw_org`, `llmgw_team`, `llmgw_project`, `llmgw_key` (the key in a
sensitive `secret`) and `llmgw_membership`.
- **The firewall setting reaches every machine.** A separate mirror and the
store's members follow `firewall.enabled` and `firewall_networks` as nodes
and control planes did: the leading control plane tells the mirror over
its API, and keeps the state in the store for members that run no llmgw,
which a timer on each reads. The mirror also lets in the network of every
node the leader knows. `mirror init` and `server init --patroni` install
what follows it; run them again on a cluster set up before.
### Changed
- **The store's PostgreSQL is sized to its machine's memory.** Before, it ran
+23 -7
View File
@@ -2196,13 +2196,29 @@ llmgw ctl config set firewall.enabled false # every machine's table taken out
llmgw ctl config set firewall.enabled true # and put back as it was
```
It is in force within seconds: each node's agent takes its table out or puts
it back, and each control plane's `llmgw-firewall-sync.path` (installed by
`server init` and `server add`; `sudo llmgw firewall sync --install` on one
set up before) does the same there. The rules files stay, so turning it back
on restores them. Nodes added while it is off get none. A separate mirror
machine and the store's members keep theirs for now. `firewall_networks`
changes reach the nodes the same way.
It is in force on every machine within a minute, each in its own way:
- **nodes**: the agent takes its table out or puts it back;
- **control planes**: `llmgw-firewall-sync.path`, installed by `server
init` and `server add` (`sudo llmgw firewall sync --install` on one set up
before);
- **a separate mirror machine**: the leading control plane tells the mirror,
which writes it to `.firewall.json` in its directory, and the same path
unit applies it. `mirror init` installs it; on a mirror set up before, run
`sudo llmgw mirror init` again with the same flags;
- **store members that are nothing else**: they run no llmgw, so
`llmgw-firewall-follow.timer` reads what the leading control plane keeps in
the store, from the member's own copy, once a minute. `server init
--patroni` installs it; run it again on a cluster set up before.
The rules files stay, so turning it back on restores them. Nodes added while
it is off get none. Networks added to `firewall_networks` reach every machine
the same way. One taken out of it stays let in until it is taken out of
`/etc/llmgw/firewall.nft` by hand: the rules are only ever added to.
The leading control plane also tells the mirror machine where every node and
control plane is, so a node on a network the mirror had not let in can fetch
its weights without a `firewall_networks` entry.
Only those ports are touched. The table's policy is to accept, so whatever
else the machine serves — and whatever firewall it already has — is left as
+49 -21
View File
@@ -7,12 +7,14 @@ import (
"fmt"
"net/netip"
"os"
"strings"
"time"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/firewall"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
)
// firewallStateFile is where a control plane writes what the cluster's
@@ -20,11 +22,9 @@ import (
// the server itself runs as llmgw and cannot change the ruleset.
const firewallStateFile = "/var/lib/llmgw/firewall.json"
// firewallState is what is written there.
type firewallState struct {
Enabled bool `json:"enabled"`
Networks []string `json:"networks,omitempty"`
}
// firewallState is what is written there: the same as the mirror is told
// and keeps beside its models.
type firewallState = mirror.FirewallState
// writeFirewallState records the settings' firewall for this machine's sync
// unit. Without the state directory — a server not run by systemd — there
@@ -46,25 +46,34 @@ func writeFirewallState(cfg config.Server) error {
// The units that apply it: a path unit watching the file, and the oneshot it
// starts.
const (
firewallSyncPathUnit = "llmgw-firewall-sync.path"
firewallSyncUnitFile = `[Unit]
Description=llmgw firewall: apply the cluster's settings on this control plane
const firewallSyncPathUnit = "llmgw-firewall-sync.path"
// firewallSyncUnits are the two units for a state file: on a control plane
// the one its server writes, on a mirror the one the control plane has the
// mirror write.
func firewallSyncUnits(state string) (service, pathUnit string) {
arg := ""
if state != firewallStateFile {
arg = " --state " + state
}
service = `[Unit]
Description=llmgw firewall: apply the cluster's settings on this machine
[Service]
Type=oneshot
ExecStart=/usr/local/bin/llmgw firewall sync
ExecStart=/usr/local/bin/llmgw firewall sync` + arg + `
`
firewallSyncPathFile = `[Unit]
Description=llmgw firewall: follow the cluster's settings on this control plane
pathUnit = `[Unit]
Description=llmgw firewall: follow the cluster's settings on this machine
[Path]
PathChanged=` + firewallStateFile + `
PathChanged=` + state + `
[Install]
WantedBy=multi-user.target
`
)
return service, pathUnit
}
func newFirewallCmd() *cobra.Command {
c := &cobra.Command{
@@ -76,6 +85,7 @@ func newFirewallCmd() *cobra.Command {
"which " + firewallSyncPathUnit + " applies with 'llmgw firewall sync'.",
}
var install bool
state := firewallStateFile
sync := &cobra.Command{
Use: "sync",
Short: "Apply what the cluster's settings say of the firewall, here (root)",
@@ -85,21 +95,21 @@ func newFirewallCmd() *cobra.Command {
return errors.New("the firewall is changed as root: run it with sudo")
}
if install {
if err := installFirewallSync(cmd.Context(), firewall.Local{}); err != nil {
if err := installFirewallSync(cmd.Context(), firewall.Local{}, state); err != nil {
return err
}
fmt.Fprintln(cmd.OutOrStdout(), firewallSyncPathUnit+" follows the cluster's settings here")
if _, err := os.Stat(firewallStateFile); err != nil {
return nil // the server writes it when it starts
if _, err := os.Stat(state); err != nil {
return nil // written when the settings are first told
}
}
b, err := os.ReadFile(firewallStateFile)
b, err := os.ReadFile(state)
if err != nil {
return err
}
var st firewallState
if err := json.Unmarshal(b, &st); err != nil {
return fmt.Errorf("%s: %w", firewallStateFile, err)
return fmt.Errorf("%s: %w", state, err)
}
ctx, cancel := context.WithTimeout(cmd.Context(), 2*time.Minute)
defer cancel()
@@ -112,14 +122,17 @@ func newFirewallCmd() *cobra.Command {
},
}
sync.Flags().BoolVar(&install, "install", false, "first install the units that run this whenever the settings change")
sync.Flags().StringVar(&state, "state", firewallStateFile,
"the file the settings are written to: a control plane's, or "+mirror.FirewallFile+" in a mirror's directory")
c.AddCommand(sync)
return c
}
// installFirewallSync writes the sync units on a machine and has the path
// unit follow the state file.
func installFirewallSync(ctx context.Context, r firewall.Runner) error {
body := firewallSyncUnitFile + "\x00" + firewallSyncPathFile
func installFirewallSync(ctx context.Context, r firewall.Runner, state string) error {
service, pathUnit := firewallSyncUnits(state)
body := service + "\x00" + pathUnit
out, err := r.Run(ctx, `set -e
t=$(mktemp -d); trap 'rm -rf "$t"' EXIT
cat > "$t/all"
@@ -172,5 +185,20 @@ func syncFirewall(ctx context.Context, r firewall.Runner, st firewallState) (str
return "", err
}
}
// The cluster's machines: a node on a network this one has not let in
// yet, which the mirror's files are for.
if len(st.Toward) > 0 {
added, err := firewall.Admit(ctx, r, st.Toward)
if err != nil {
return "", err
}
if len(added) > 0 {
shown := make([]string, len(added))
for i, p := range added {
shown[i] = p.String()
}
return "llmgw's firewall is on here, and now lets in " + strings.Join(shown, ", "), nil
}
}
return "llmgw's firewall is on here", nil
}
+32
View File
@@ -2,8 +2,11 @@ package cli
import (
"context"
"net/netip"
"strings"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/firewall"
)
type scriptLog struct {
@@ -49,3 +52,32 @@ func TestAControlPlaneFollowsTheFirewallSetting(t *testing.T) {
t.Errorf("on: %q %v", msg, err)
}
}
func TestAMachineLetsInTheClusterMachinesItIsTold(t *testing.T) {
// The rules hold 10.0.0.0/24; a node on 192.168.7.0/24 is let in, as
// the route to it says, and one already in is not asked about.
r := &scriptLog{reply: func(s string) string {
switch {
case strings.Contains(s, "echo loaded"):
return "loaded\n"
case strings.Contains(s, "cat /etc/llmgw/firewall.nft"):
return firewallRulesFor("10.0.0.0/24")
case strings.Contains(s, "ip -o route get 192.168.7.20"):
return "192.168.7.20 dev eth1 src 192.168.7.1 uid 0\n2: eth1 inet 192.168.7.1/24 brd 192.168.7.255 scope global eth1\n"
case strings.Contains(s, "nft -f"):
return "changed\n"
}
return ""
}}
msg, err := syncFirewall(context.Background(), r, firewallState{Enabled: true, Toward: []string{"10.0.0.147", "192.168.7.20"}})
if err != nil || !strings.Contains(msg, "192.168.7.0/24") {
t.Fatalf("%q %v", msg, err)
}
if strings.Contains(strings.Join(r.scripts, "\n"), "route get 10.0.0.147") {
t.Error("asked the route to a machine already let in")
}
}
func firewallRulesFor(nets string) string {
return firewall.Script(firewall.Spec{Ports: []firewall.Port{firewall.MirrorFiles}, Allow: []netip.Prefix{netip.MustParsePrefix(nets)}})
}
+146
View File
@@ -0,0 +1,146 @@
package cli
import (
"context"
"encoding/json"
"log/slog"
"net"
"net/netip"
"net/url"
"slices"
"sync"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// firewallStateKey is where, in the store, the firewall's state is kept
// for the store's members to read.
const firewallStateKey = "firewall"
// firewallPublisher tells the machines that cannot ask a control plane
// what the cluster's settings say of their firewall: the mirror, over its
// API, and the store's members, through the store they hold a copy of.
// Nodes are told over their agents' streams and control planes by their
// own servers. On the leader only, at start, at every change to the
// firewall's settings, and every few minutes for nodes come and gone.
type firewallPublisher struct {
settings func() config.Server
st *store.DB
nodes func() []string
leader func() bool
log *slog.Logger
mu sync.Mutex
mirror *mirror.Client
kick chan struct{}
}
func newFirewallPublisher(settings func() config.Server, st *store.DB, nodes func() []string,
leader func() bool, log *slog.Logger) *firewallPublisher {
return &firewallPublisher{settings: settings, st: st, nodes: nodes, leader: leader, log: log,
kick: make(chan struct{}, 1)}
}
// setMirror has the mirror told too.
func (p *firewallPublisher) setMirror(m *mirror.Client) {
p.mu.Lock()
p.mirror = m
p.mu.Unlock()
p.Kick()
}
// Kick has the state worked out and told again now.
func (p *firewallPublisher) Kick() {
select {
case p.kick <- struct{}{}:
default:
}
}
func (p *firewallPublisher) run(ctx context.Context, every time.Duration) {
t := time.NewTicker(every)
defer t.Stop()
for {
p.once(ctx)
select {
case <-ctx.Done():
return
case <-t.C:
case <-p.kick:
}
}
}
// state is what the settings say now, with the cluster's machines.
func (p *firewallPublisher) state(ctx context.Context) mirror.FirewallState {
cfg := p.settings()
st := mirror.FirewallState{Enabled: cfg.Firewall.Enabled, Networks: slices.Clone(cfg.FirewallNetworks)}
var toward []string
if p.nodes != nil {
toward = append(toward, p.nodes()...)
}
if p.st != nil {
if cps, err := p.st.ControlPlanes(ctx); err == nil {
for _, c := range cps {
toward = append(toward, c.URL)
}
}
}
for _, t := range toward {
if a, ok := hostAddr(t); ok {
st.Toward = append(st.Toward, a)
}
}
slices.Sort(st.Toward)
st.Toward = slices.Compact(st.Toward)
return st
}
// hostAddr is the address in an address, host:port or URL; names are not
// looked up here.
func hostAddr(s string) (string, bool) {
if u, err := url.Parse(s); err == nil && u.Host != "" {
s = u.Hostname()
} else if h, _, err := net.SplitHostPort(s); err == nil {
s = h
}
a, err := netip.ParseAddr(s)
if err != nil {
return "", false
}
return a.Unmap().String(), true
}
func (p *firewallPublisher) once(ctx context.Context) {
if p.leader != nil && !p.leader() {
return
}
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
st := p.state(ctx)
p.mu.Lock()
m := p.mirror
p.mu.Unlock()
if m != nil {
changed, err := m.SetFirewall(ctx, st)
switch {
case err != nil:
p.log.Warn("could not tell the mirror what the settings say of its firewall", "err", err)
case changed:
p.log.Info("the mirror's firewall follows the cluster's settings", "enabled", st.Enabled)
}
}
if p.st != nil {
b, _ := json.Marshal(st)
changed, err := p.st.PutClusterState(ctx, firewallStateKey, string(b))
switch {
case err != nil:
p.log.Warn("could not record the firewall's state for the store's members", "err", err)
case changed:
p.log.Info("the store's members' firewall follows the cluster's settings", "enabled", st.Enabled)
}
}
}
+59
View File
@@ -0,0 +1,59 @@
package cli
import (
"context"
"encoding/json"
"log/slog"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
func TestTheLeaderTellsTheMirrorAndTheStoreWhatTheFirewallSettingsSay(t *testing.T) {
ctx := context.Background()
st, err := store.OpenSQLite(ctx, filepath.Join(t.TempDir(), "s.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
if err := st.PutControlPlane(ctx, store.ControlPlane{URL: "https://10.0.0.150:7443", Name: "ctl02"}); err != nil {
t.Fatal(err)
}
root := t.TempDir()
srv := httptest.NewServer(mirror.NewAPI(mirror.NewManager(root, slog.New(slog.DiscardHandler)), "http://m:8081").Handler())
defer srv.Close()
mc, _ := mirror.NewClient(srv.URL, tlsconf.Config{}, 0)
cfg := config.Server{Firewall: config.FirewallSettings{Enabled: false}, FirewallNetworks: []string{"192.168.5.0/24"}}
leader := false
p := newFirewallPublisher(func() config.Server { return cfg }, st,
func() []string { return []string{"10.0.0.147", "", "10.0.0.147", "node-by-name"} },
func() bool { return leader }, slog.New(slog.DiscardHandler))
p.setMirror(mc)
p.once(ctx)
if _, err := st.ClusterState(ctx, firewallStateKey); err == nil {
t.Error("a standby told the store")
}
leader = true
p.once(ctx)
var got mirror.FirewallState
v, err := st.ClusterState(ctx, firewallStateKey)
if err != nil || json.Unmarshal([]byte(v), &got) != nil {
t.Fatalf("the store holds %q %v", v, err)
}
if got.Enabled || got.Networks[0] != "192.168.5.0/24" || len(got.Toward) != 2 ||
got.Toward[0] != "10.0.0.147" || got.Toward[1] != "10.0.0.150" {
t.Errorf("state %+v", got)
}
b, err := os.ReadFile(filepath.Join(root, mirror.FirewallFile))
if err != nil || string(b) != v {
t.Errorf("the mirror holds %s, the store %s", b, v)
}
}
+9
View File
@@ -8,6 +8,7 @@ import (
"net"
"net/netip"
"os"
"path/filepath"
"runtime"
"strings"
"time"
@@ -65,6 +66,14 @@ Safe to run again: it changes only what differs.`,
}
if !noFirewall {
mirrorFirewall(cmd.Context(), cmd.OutOrStdout(), controlPlanes)
// And after, as the cluster's settings say: the control plane
// tells the mirror, which writes it beside its models.
state := filepath.Join(*dir, mirror.FirewallFile)
if err := installFirewallSync(cmd.Context(), firewall.Local{}, state); err != nil {
fmt.Fprintf(cmd.OutOrStdout(), " !! firewall: %v; the cluster's settings will not change it here\n", err)
} else {
fmt.Fprintln(cmd.OutOrStdout(), " ok firewall: follows the cluster's settings ("+firewallSyncPathUnit+")")
}
}
return nil
},
+22
View File
@@ -233,6 +233,9 @@ func newServerCmd() *cobra.Command {
var agentHandler http.Handler
var agentPath string
var admin *control.Controller
// What the settings say of the firewall, for the mirror and the
// store's members; nil without a control plane.
var fwPub *firewallPublisher
updateMetrics := &modelserve.UpdateCollector{}
// The store's cluster, when it is one; read by the metrics and
// by `ctl status`.
@@ -276,6 +279,19 @@ func newServerCmd() *cobra.Command {
ctl.Reconciler().SetNotifier(notifier)
admin = ctl
agentPath, agentHandler = svc.Handler()
fwPub = newFirewallPublisher(func() config.Server { return *current.Load() }, rt.Store(),
func() []string {
var out []string
for _, n := range ctl.Inventory().Nodes() {
out = append(out, n.MgmtAddr)
}
return out
},
func() bool { return ctl.Leadership() == nil || ctl.Leadership().IsLeader() }, log)
go fwPub.run(ctx, 5*time.Minute)
if l := ctl.Leadership(); l != nil {
l.OnLead(fwPub.Kick)
}
// Before AdminHandler, which is where the registry is built —
// registering after it silently publishes nothing, which is
@@ -519,6 +535,9 @@ func newServerCmd() *cobra.Command {
return fmt.Errorf("the model mirror: %w", err)
}
rt.SetMirror(mc)
if fwPub != nil {
fwPub.setMirror(mc)
}
// Adapters trained on the nodes are kept there too; sending one
// takes longer than the mirror's usual answers.
if adminAPI != nil {
@@ -743,6 +762,9 @@ func newServerCmd() *cobra.Command {
if agentSvc != nil {
agentSvc.BroadcastFirewall()
}
if fwPub != nil {
fwPub.Kick()
}
}
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "agents") }) && agentSvc != nil {
// A moment apart per node, so off this goroutine.
+1 -1
View File
@@ -160,7 +160,7 @@ func (sa *serverAdd) firewall(ctx context.Context) error {
return err
}
// What applies the cluster's firewall setting there when it changes.
if err := installFirewallSync(ctx, sa.remote); err != nil {
if err := installFirewallSync(ctx, sa.remote, firewallStateFile); err != nil {
fmt.Fprintf(sa.out, " !! firewall: %v\n", err)
}
if !cfg.Firewall.Enabled {
+6 -1
View File
@@ -93,6 +93,11 @@ func (in *serverInit) storeFirewall(ctx context.Context, members []pgha.Node, se
} else {
in.kept("firewall on %s", m.Name)
}
// And after, as the cluster's settings say, read from the
// member's own copy of the store.
if err := firewall.InstallFollower(ctx, m.Remote); err != nil {
fmt.Fprintf(in.out, " !! firewall on %s: %v; the cluster's settings will not change it\n", m.Name, err)
}
continue
}
}
@@ -135,5 +140,5 @@ func loopbackListen(listen string) bool {
// ensureFirewallSync installs the units that apply the cluster's firewall
// setting on this control plane.
func (in *serverInit) ensureFirewallSync(ctx context.Context) error {
return installFirewallSync(ctx, firewall.Local{})
return installFirewallSync(ctx, firewall.Local{}, firewallStateFile)
}
+154
View File
@@ -0,0 +1,154 @@
package firewall
import (
"context"
"fmt"
"strings"
)
// A store member that is nothing else runs no llmgw process: Patroni, etcd
// and PostgreSQL only. What the cluster's settings say of its firewall it
// reads from its own copy of the store, where the leading control plane
// keeps it (the cluster_state table, key firewall), once a minute, by a
// root timer running FollowScript. It turns llmgw's table off and on, and
// lets in the networks the settings add; what `server init` let in stays.
// FollowScript is the script, as python3 runs it. LLMGW_FW_STATE and
// LLMGW_FW_RULES replace the store and the rules file, for its tests.
const FollowScript = `#!/usr/bin/python3
# Written by llmgw: llmgw's firewall on this store member, as the cluster's
# settings say. Run by llmgw-firewall-follow.timer.
import ipaddress, json, os, subprocess, sys, tempfile
RULES = os.environ.get("LLMGW_FW_RULES", "` + RulesFile + `")
def run(*a):
return subprocess.run(a, capture_output=True, text=True)
raw = os.environ.get("LLMGW_FW_STATE")
if raw is None:
r = run("runuser", "-u", "postgres", "--", "psql", "-d", "llmgw", "-tAqc",
"SELECT value FROM cluster_state WHERE key = 'firewall'")
# No table yet, or nothing told: a control plane older than this, or
# none leading. Left as it is.
if r.returncode != 0:
sys.exit(0)
raw = r.stdout
if not raw.strip():
sys.exit(0)
st = json.loads(raw)
present = run("nft", "list", "table", "inet", "` + Table + `").returncode == 0
if not st.get("enabled", True):
if present:
run("systemctl", "disable", "--now", "` + Unit + `")
run("nft", "delete", "table", "inet", "` + Table + `")
print("llmgw's firewall is off here")
sys.exit(0)
if not os.path.exists(RULES):
sys.exit(0)
lines = open(RULES).read().split("\n")
want = []
for n in st.get("networks") or []:
try:
want.append(ipaddress.ip_network(n, strict=False))
except ValueError:
pass
def saddr(l):
s = l.strip()
return s.startswith("ip saddr {") or s.startswith("ip6 saddr {")
changed = False
for fam, kw in ((4, "ip saddr {"), (6, "ip6 saddr {")):
add = [str(n) for n in want if n.version == fam]
if not add:
continue
same = [i for i, l in enumerate(lines) if l.strip().startswith(kw)]
if same:
i = same[0]
l = lines[i]
a, b = l.index("{") + 1, l.index("}")
have = [x.strip() for x in l[a:b].split(",") if x.strip()]
new = [x for x in add if x not in have]
if new:
lines[i] = l[:a] + " " + ", ".join(have + new) + " " + l[b:]
changed = True
continue
other = [i for i, l in enumerate(lines) if saddr(l)]
if not other:
continue
l = lines[other[0]]
indent = l[:len(l) - len(l.lstrip())]
lines.insert(other[0] + 1, indent + kw + " " + ", ".join(add) + " }" + l[l.index("}") + 1:])
changed = True
if changed:
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(RULES))
with os.fdopen(fd, "w") as f:
f.write("\n".join(lines))
if run("nft", "-c", "-f", tmp).returncode != 0:
os.unlink(tmp)
print("the settings' networks make a ruleset nft refuses; left as it was", file=sys.stderr)
sys.exit(1)
os.chmod(tmp, 0o644)
os.replace(tmp, RULES)
if changed or not present:
run("systemctl", "enable", "` + Unit + `")
if run("nft", "-f", RULES).returncode != 0:
sys.exit(1)
print("llmgw's firewall is on here")
`
// FollowerPath is where FollowScript is installed.
const FollowerPath = "/usr/local/lib/llmgw/firewall-follow"
// FollowerTimer is the timer that runs it.
const FollowerTimer = "llmgw-firewall-follow.timer"
const followService = `[Unit]
Description=llmgw firewall: follow the cluster's settings on this store member
After=postgresql.service patroni.service
[Service]
Type=oneshot
ExecStart=/usr/bin/python3 ` + FollowerPath + `
`
const followTimer = `[Unit]
Description=llmgw firewall: follow the cluster's settings on this store member
[Timer]
OnBootSec=1min
OnUnitActiveSec=1min
[Install]
WantedBy=timers.target
`
// InstallFollower puts FollowScript and its timer on a store member.
func InstallFollower(ctx context.Context, r Runner) error {
body := FollowScript + "\x00" + followService + "\x00" + followTimer
out, err := r.Run(ctx, `set -e
t=$(mktemp -d); trap 'rm -rf "$t"' EXIT
cat > "$t/all"
python3 - "$t" <<'PY'
import sys
d = sys.argv[1]
s, svc, tim = open(d + "/all", "rb").read().split(b"\x00")
open(d + "/script", "wb").write(s)
open(d + "/service", "wb").write(svc)
open(d + "/timer", "wb").write(tim)
PY
sudo install -d -m 755 /usr/local/lib/llmgw
sudo install -m 755 "$t/script" `+FollowerPath+`
sudo install -m 644 "$t/service" /etc/systemd/system/llmgw-firewall-follow.service
sudo install -m 644 "$t/timer" /etc/systemd/system/`+FollowerTimer+`
sudo systemctl daemon-reload
sudo systemctl enable --now `+FollowerTimer+` >/dev/null 2>&1`, []byte(body))
if err != nil {
return fmt.Errorf("installing %s: %v: %s", FollowerTimer, err, strings.TrimSpace(string(out)))
}
return nil
}
+101
View File
@@ -0,0 +1,101 @@
package firewall
import (
"net/netip"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
)
// follow runs FollowScript against a rules file and a state, with nft and
// systemctl replaced by scripts that log what they were asked and answer as
// a machine whose table is loaded or not.
func follow(t *testing.T, rules, state string, loaded bool) (string, string, string) {
t.Helper()
if _, err := exec.LookPath("python3"); err != nil {
t.Skip("no python3")
}
dir := t.TempDir()
bin := filepath.Join(dir, "bin")
_ = os.Mkdir(bin, 0o755)
log := filepath.Join(dir, "log")
list := "exit 1"
if loaded {
list = "exit 0"
}
for name, body := range map[string]string{
"nft": `echo "nft $*" >> ` + log + "\n" + `[ "$1" = list ] && ` + list + "\nexit 0\n",
"systemctl": `echo "systemctl $*" >> ` + log + "\nexit 0\n",
} {
if err := os.WriteFile(filepath.Join(bin, name), []byte("#!/bin/sh\n"+body), 0o755); err != nil {
t.Fatal(err)
}
}
rf := filepath.Join(dir, "firewall.nft")
if rules != "" {
_ = os.WriteFile(rf, []byte(rules), 0o644)
}
script := filepath.Join(dir, "follow")
_ = os.WriteFile(script, []byte(FollowScript), 0o755)
cmd := exec.Command("python3", script)
cmd.Env = append(os.Environ(), "PATH="+bin+":"+os.Getenv("PATH"), "LLMGW_FW_STATE="+state, "LLMGW_FW_RULES="+rf)
out, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("%v: %s", err, out)
}
got, _ := os.ReadFile(rf)
calls, _ := os.ReadFile(log)
return string(out), string(got), string(calls)
}
var memberRules = Script(Spec{Ports: []Port{Postgres, EtcdClient}, Allow: mustPrefixes("10.0.0.0/24")})
func TestAStoreMemberFollowsTheSettings(t *testing.T) {
// Off: the table goes, the rules stay for later.
out, rules, calls := follow(t, memberRules, `{"enabled":false}`, true)
if !strings.Contains(out, "off") || !strings.Contains(calls, "nft delete table inet llmgw") ||
!strings.Contains(calls, "systemctl disable --now "+Unit) || rules != memberRules {
t.Errorf("off: %q %q", out, calls)
}
// On again: loaded from the rules kept.
out, _, calls = follow(t, memberRules, `{"enabled":true}`, false)
if !strings.Contains(out, "on") || !strings.Contains(calls, "nft -f") {
t.Errorf("on: %q %q", out, calls)
}
// On and loaded, nothing new: nothing done.
out, _, calls = follow(t, memberRules, `{"enabled":true,"networks":["10.0.0.0/24"]}`, true)
if out != "" || strings.Contains(calls, "nft -f") {
t.Errorf("unchanged: %q %q", out, calls)
}
// Networks the settings add, of both families, are let in beside what
// init let in; the result reads back as the same ports.
_, rules, calls = follow(t, memberRules, `{"enabled":true,"networks":["192.168.5.0/24","fd00::/64","nonsense"]}`, true)
if !strings.Contains(calls, "nft -c -f") || !strings.Contains(calls, "nft -f") {
t.Errorf("not checked and loaded: %q", calls)
}
back := parse(rules)
var nets []string
for _, p := range back.Allow {
nets = append(nets, p.String())
}
if strings.Join(nets, ",") != "10.0.0.0/24,192.168.5.0/24,fd00::/64" || len(back.Ports) != 2 {
t.Errorf("rules now:\n%s", rules)
}
// Nothing told yet, or no rules: left alone.
if out, _, calls := follow(t, memberRules, "", true); out != "" || calls != "" {
t.Errorf("nothing told: %q %q", out, calls)
}
if out, _, calls := follow(t, "", `{"enabled":true}`, false); out != "" || strings.Contains(calls, "nft -f") {
t.Errorf("no rules: %q %q", out, calls)
}
}
func mustPrefixes(ss ...string) []netip.Prefix {
var out []netip.Prefix
for _, s := range ss {
out = append(out, netip.MustParsePrefix(s))
}
return out
}
+1
View File
@@ -40,6 +40,7 @@ func (a *API) Handler() http.Handler {
mux.HandleFunc("DELETE /v1/models/{model}/{version}", a.remove)
mux.HandleFunc("PUT /v1/models/{model}/{version}/{file}", a.upload)
mux.HandleFunc("GET /v1/health", a.health)
mux.HandleFunc("PUT /v1/firewall", a.putFirewall)
return mux
}
+65
View File
@@ -0,0 +1,65 @@
package mirror
import (
"bytes"
"context"
"encoding/json"
"net/http"
"os"
"path/filepath"
)
// The mirror's machine keeps llmgw's firewall as the cluster's settings
// say, as every other machine does. The mirror runs unprivileged and cannot
// change the ruleset: the control plane tells it what the settings say, it
// writes that beside its models, and a root path unit on the machine
// applies it with `llmgw firewall sync`.
// FirewallState is what the cluster's settings say of a machine's firewall.
type FirewallState struct {
Enabled bool `json:"enabled"`
// Networks are let in besides those the machine reaches others on.
Networks []string `json:"networks,omitempty"`
// Toward are the cluster's machines, by address: each one's network is
// let in, as the machine's route to it says.
Toward []string `json:"toward,omitempty"`
}
// FirewallFile is where, in the mirror's directory, the state is written.
// Hidden, so it is never taken for a model.
const FirewallFile = ".firewall.json"
func (a *API) putFirewall(w http.ResponseWriter, r *http.Request) {
var st FirewallState
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&st); err != nil {
fail(w, http.StatusBadRequest, err)
return
}
b, _ := json.Marshal(st)
p := filepath.Join(a.m.root, FirewallFile)
changed := false
if old, err := os.ReadFile(p); err != nil || !bytes.Equal(old, b) {
// Renamed into place: the path unit sees one change, of a whole file.
tmp := p + ".new"
if err := os.WriteFile(tmp, b, 0o644); err != nil {
fail(w, http.StatusInternalServerError, err)
return
}
if err := os.Rename(tmp, p); err != nil {
fail(w, http.StatusInternalServerError, err)
return
}
changed = true
}
writeJSON(w, http.StatusOK, map[string]bool{"changed": changed})
}
// SetFirewall tells the mirror what the cluster's settings say of its
// machine's firewall; it answers whether that changed anything.
func (c *Client) SetFirewall(ctx context.Context, st FirewallState) (bool, error) {
var out struct {
Changed bool `json:"changed"`
}
err := c.do(ctx, http.MethodPut, "/v1/firewall", st, &out)
return out.Changed, err
}
+38
View File
@@ -0,0 +1,38 @@
package mirror
import (
"context"
"encoding/json"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
func TestTheMirrorKeepsWhatTheSettingsSayOfItsFirewall(t *testing.T) {
root := t.TempDir()
srv := httptest.NewServer(NewAPI(NewManager(root, discardLog()), "http://m:8081").Handler())
defer srv.Close()
c, err := NewClient(srv.URL, tlsconf.Config{}, 0)
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
st := FirewallState{Enabled: true, Networks: []string{"192.168.5.0/24"}, Toward: []string{"10.0.0.147"}}
if changed, err := c.SetFirewall(ctx, st); err != nil || !changed {
t.Fatalf("first: %v %v", changed, err)
}
if changed, err := c.SetFirewall(ctx, st); err != nil || changed {
t.Errorf("the same again: %v %v", changed, err)
}
var back FirewallState
b, _ := os.ReadFile(filepath.Join(root, FirewallFile))
if err := json.Unmarshal(b, &back); err != nil || !back.Enabled || back.Toward[0] != "10.0.0.147" {
t.Errorf("kept %s", b)
}
if all, err := c.List(ctx); err != nil || len(all.Models) != 0 {
t.Errorf("the file is taken for a model: %+v %v", all, err)
}
}
+30
View File
@@ -0,0 +1,30 @@
package store
import (
"context"
"database/sql"
"errors"
"time"
)
// PutClusterState keeps a value under a key, for machines that read the
// store rather than a control plane; it reports whether it changed.
func (s *DB) PutClusterState(ctx context.Context, key, value string) (bool, error) {
if old, err := s.ClusterState(ctx, key); err == nil && old == value {
return false, nil
}
_, err := s.db.ExecContext(ctx, `INSERT INTO cluster_state (key, value, updated_at) VALUES (?, ?, ?)
ON CONFLICT (key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`,
key, value, millis(time.Now()))
return err == nil, err
}
// ClusterState reads a value; ErrNotFound when there is none.
func (s *DB) ClusterState(ctx context.Context, key string) (string, error) {
var v string
err := s.db.QueryRowContext(ctx, `SELECT value FROM cluster_state WHERE key = ?`, key).Scan(&v)
if errors.Is(err, sql.ErrNoRows) {
return "", ErrNotFound
}
return v, err
}
+2 -1
View File
@@ -86,7 +86,8 @@ var copyOrder = []string{
// the exact moment somebody is watching a migration and wondering why nothing
// is being placed.
var notCopied = map[string]string{
"leadership": "a lease on a running process, meaningless in a new store",
"leadership": "a lease on a running process, meaningless in a new store",
"cluster_state": "worked out again by the leader from the settings and the machines, within minutes",
}
// CopyResult is how many rows each table received.
@@ -0,0 +1,7 @@
-- What a control plane works out for machines that cannot ask it: the
-- store's members read it from their own copy of the store.
CREATE TABLE cluster_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at BIGINT NOT NULL
);
@@ -0,0 +1,7 @@
-- What a control plane works out for machines that cannot ask it: the
-- store's members read it from their own copy of the store.
CREATE TABLE cluster_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at INTEGER NOT NULL
);