llmgw/internal/cli/serverinit_firewall.go

package cli

import (
	"context"
	"fmt"
	"net"
	"net/netip"
	"os"
	"strconv"
	"strings"

	"gitea.mynodes.xyz/ilker/llmgw/internal/config"
	"gitea.mynodes.xyz/ilker/llmgw/internal/firewall"
	"gitea.mynodes.xyz/ilker/llmgw/internal/pgha"
)

// ensureFirewall keeps this control plane's internal ports to the networks
// the cluster is on: the agents' stream, the metrics, the mirror's files,
// the engines and etcd when they are here. The client API is not among
// them; it is meant to be reached.
func (in *serverInit) ensureFirewall(ctx context.Context) error {
	cfg, err := loadEffectiveConfig(in.configPath())
	if err != nil {
		return err
	}
	// What applies the cluster's setting here when it changes: the server
	// writes it, as llmgw; this applies it, as root.
	if err := in.ensureFirewallSync(ctx); err != nil {
		return err
	}
	if !cfg.Firewall.Enabled {
		in.kept("firewall: off, as the cluster's settings say (firewall.enabled)")
		return nil
	}
	ports := controlPlanePorts(cfg)
	if !in.noMirror && in.mirrorURL == "" {
		ports = append(ports, firewall.MirrorFiles)
	}
	if !in.noNode {
		from, to := cfg.EnginePorts()
		ports = append(ports, firewall.Engines(from, to), firewall.NodeExporter)
	}
	if _, err := os.Stat("/etc/default/etcd"); err == nil {
		// The store's third etcd vote.
		ports = append(ports, firewall.EtcdClient, firewall.EtcdPeer)
	}
	self, err := ipv4Of(in.reachableAddr())
	if err != nil {
		return err
	}
	nets, err := firewall.Networks(ctx, pgha.Local{}, []string{self})
	if err != nil {
		return err
	}
	extra, _ := cfg.FirewallAllow()
	nets = append(nets, extra...)
	changed, err := firewall.Install(ctx, pgha.Local{}, firewall.Spec{Ports: ports, Allow: nets})
	if err != nil {
		return err
	}
	shown := make([]string, len(nets))
	for i, n := range nets {
		shown[i] = n.String()
	}
	if changed {
		in.done("firewall: llmgw's internal ports answer %s only (%s)", strings.Join(shown, ", "), firewall.RulesFile)
	} else {
		in.kept("firewall: llmgw's internal ports answer %s only", strings.Join(shown, ", "))
	}
	return nil
}

// storeFirewall keeps each store member's database, Patroni and etcd ports to
// the networks it reaches this control plane and the other members on. A
// member it cannot be set on is said, and left as it was.
func (in *serverInit) storeFirewall(ctx context.Context, members []pgha.Node, self string, extra []netip.Prefix) {
	ports := []firewall.Port{firewall.Postgres, firewall.PatroniAPI, firewall.EtcdClient, firewall.EtcdPeer,
		firewall.NodeExporter}
	for _, m := range members {
		toward := []string{self}
		for _, o := range members {
			if o.IP != m.IP {
				toward = append(toward, o.IP)
			}
		}
		nets, err := firewall.Networks(ctx, m.Remote, toward)
		if err == nil {
			nets = append(nets, extra...)
			var changed bool
			if changed, err = firewall.Install(ctx, m.Remote, firewall.Spec{Ports: ports, Allow: nets}); err == nil {
				if changed {
					in.done("firewall on %s: the store's ports answer the cluster's networks only", m.Name)
				} 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
			}
		}
		fmt.Fprintf(in.out, "  !!    firewall on %s: %v\n", m.Name, err)
	}
}

// controlPlanePorts are a control plane's internal ports, from its
// configuration: the agents' stream, and the metrics when they listen
// beyond this machine.
func controlPlanePorts(cfg config.Server) []firewall.Port {
	var ports []firewall.Port
	if p := portOf(cfg.AgentListen); p > 0 {
		ports = append(ports, firewall.Single(p, "control plane: the agents' stream"))
	}
	if p := portOf(cfg.MetricsListen); p > 0 && !loopbackListen(cfg.MetricsListen) {
		ports = append(ports, firewall.Single(p, "control plane: metrics, for Prometheus"))
	}
	return ports
}

func portOf(listen string) int {
	_, port, err := net.SplitHostPort(listen)
	if err != nil {
		return 0
	}
	p, _ := strconv.Atoi(port)
	return p
}

func loopbackListen(listen string) bool {
	host, _, err := net.SplitHostPort(listen)
	if err != nil {
		return false
	}
	ip := net.ParseIP(host)
	return host == "localhost" || (ip != nil && ip.IsLoopback())
}

// 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{}, firewallStateFile)
}