diff --git a/BACKLOG.md b/BACKLOG.md index 81191fa..5e0c032 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md index cc46bf9..05e051d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/docs/install.md b/docs/install.md index 2a00268..69a6c34 100644 --- a/docs/install.md +++ b/docs/install.md @@ -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 diff --git a/internal/cli/firewallcmd.go b/internal/cli/firewallcmd.go index 405fae7..cf5d164 100644 --- a/internal/cli/firewallcmd.go +++ b/internal/cli/firewallcmd.go @@ -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 } diff --git a/internal/cli/firewallcmd_test.go b/internal/cli/firewallcmd_test.go index 4a05ffc..d93efec 100644 --- a/internal/cli/firewallcmd_test.go +++ b/internal/cli/firewallcmd_test.go @@ -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)}}) +} diff --git a/internal/cli/firewallpub.go b/internal/cli/firewallpub.go new file mode 100644 index 0000000..f34eeb1 --- /dev/null +++ b/internal/cli/firewallpub.go @@ -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) + } + } +} diff --git a/internal/cli/firewallpub_test.go b/internal/cli/firewallpub_test.go new file mode 100644 index 0000000..3b23065 --- /dev/null +++ b/internal/cli/firewallpub_test.go @@ -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) + } +} diff --git a/internal/cli/mirrorinit.go b/internal/cli/mirrorinit.go index 63614c1..3114c04 100644 --- a/internal/cli/mirrorinit.go +++ b/internal/cli/mirrorinit.go @@ -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 }, diff --git a/internal/cli/server.go b/internal/cli/server.go index 04ad803..9789569 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -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. diff --git a/internal/cli/serveradd.go b/internal/cli/serveradd.go index efb932c..37a8ef4 100644 --- a/internal/cli/serveradd.go +++ b/internal/cli/serveradd.go @@ -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 { diff --git a/internal/cli/serverinit_firewall.go b/internal/cli/serverinit_firewall.go index baa2231..83be84e 100644 --- a/internal/cli/serverinit_firewall.go +++ b/internal/cli/serverinit_firewall.go @@ -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) } diff --git a/internal/firewall/follow.go b/internal/firewall/follow.go new file mode 100644 index 0000000..7e76513 --- /dev/null +++ b/internal/firewall/follow.go @@ -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 +} diff --git a/internal/firewall/follow_test.go b/internal/firewall/follow_test.go new file mode 100644 index 0000000..92f1502 --- /dev/null +++ b/internal/firewall/follow_test.go @@ -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 +} diff --git a/internal/mirror/api.go b/internal/mirror/api.go index 8f9ba2a..dfcebcd 100644 --- a/internal/mirror/api.go +++ b/internal/mirror/api.go @@ -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 } diff --git a/internal/mirror/firewall.go b/internal/mirror/firewall.go new file mode 100644 index 0000000..2911166 --- /dev/null +++ b/internal/mirror/firewall.go @@ -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 +} diff --git a/internal/mirror/firewall_test.go b/internal/mirror/firewall_test.go new file mode 100644 index 0000000..38cd451 --- /dev/null +++ b/internal/mirror/firewall_test.go @@ -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) + } +} diff --git a/internal/store/clusterstate.go b/internal/store/clusterstate.go new file mode 100644 index 0000000..d37048a --- /dev/null +++ b/internal/store/clusterstate.go @@ -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 +} diff --git a/internal/store/copy.go b/internal/store/copy.go index d0eb849..22dc7d4 100644 --- a/internal/store/copy.go +++ b/internal/store/copy.go @@ -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. diff --git a/internal/store/migrations/postgres/0040_cluster_state.sql b/internal/store/migrations/postgres/0040_cluster_state.sql new file mode 100644 index 0000000..8a15e78 --- /dev/null +++ b/internal/store/migrations/postgres/0040_cluster_state.sql @@ -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 +); diff --git a/internal/store/migrations/sqlite/0040_cluster_state.sql b/internal/store/migrations/sqlite/0040_cluster_state.sql new file mode 100644 index 0000000..81d921e --- /dev/null +++ b/internal/store/migrations/sqlite/0040_cluster_state.sql @@ -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 +);