Merge pull request 'Create the nftables table in the configured address family' (#32) from benvin/v4-only-family into main

Reviewed-on: #32

cmd/tomswall/try.go

@@ -89,10 +89,10 @@ func tryApply(cfg *config.Config, fallback time.Duration) (string, error) {
 		return "", err
 	}
 	if err := engine.Apply(changes); err != nil {
-		if derr := tryapply.Discard(); derr != nil {
-			err = fmt.Errorf("%w (discarding snapshot: %v)", err, derr)
+		if aerr := tryapply.Abort(); aerr != nil {
+			return "", fmt.Errorf("applying changes: %w; %v; the revert timer restores the previous ruleset within %s", err, aerr, fallback)
 		}
-		return "", fmt.Errorf("applying changes: %w", err)
+		return "", fmt.Errorf("applying changes: %w: previous ruleset restored", err)
 	}
 	return id, nil
 }

internal/config/config.go

@@ -59,6 +59,11 @@ type Settings struct {
 
 	// When true, auto-generate CONTINUE policies for sub-zones to their parent zones.
 	ImplicitContinue bool `yaml:"implicit_continue,omitempty"`
+
+	// Verdict for ct state invalid/untracked packets; continue passes them to the rules.
+	// Unset: invalid drops, untracked continues.
+	InvalidDisposition   PolicyAction `yaml:"invalid_disposition,omitempty"`
+	UntrackedDisposition PolicyAction `yaml:"untracked_disposition,omitempty"`
 }
 
 // Load reads a config file in YAML or JSON format (detected by extension).
@@ -115,6 +120,16 @@ func (c *Config) validateSettings() error {
 	if !validAddressFamilies[c.Settings.AddressFamily] {
 		return fmt.Errorf("unknown address_family %q (use inet, ip, or ip6)", c.Settings.AddressFamily)
 	}
+	for name, d := range map[string]PolicyAction{
+		"invalid_disposition":   c.Settings.InvalidDisposition,
+		"untracked_disposition": c.Settings.UntrackedDisposition,
+	} {
+		switch d {
+		case "", PolicyAccept, PolicyDrop, PolicyReject, PolicyContinue:
+		default:
+			return fmt.Errorf("unknown %s %q (use accept, drop, reject, or continue)", name, d)
+		}
+	}
 	return nil
 }
 

internal/config/config_test.go

@@ -1052,3 +1052,20 @@ func TestSplitZoneList(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateDispositions(t *testing.T) {
+	for _, tc := range []struct {
+		invalid, untracked PolicyAction
+		wantErr            string
+	}{
+		{"", "", ""},
+		{PolicyContinue, PolicyDrop, ""},
+		{"bogus", "", `unknown invalid_disposition "bogus"`},
+		{PolicyAccept, "log", `unknown untracked_disposition "log"`},
+	} {
+		c := baseConfig()
+		c.Settings.InvalidDisposition = tc.invalid
+		c.Settings.UntrackedDisposition = tc.untracked
+		checkErr(t, c.Validate(), tc.wantErr)
+	}
+}

internal/nftables/cleanup.go

@@ -28,7 +28,7 @@ func (e *Engine) FindForeignRules() ([]ForeignRule, error) {
 
 	var ourTable *nftables.Table
 	for _, t := range tables {
-		if t.Name == e.cfg.Settings.TableName && t.Family == nftables.TableFamilyINet {
+		if t.Name == e.cfg.Settings.TableName && t.Family == e.family() {
 			ourTable = t
 			break
 		}
@@ -52,7 +52,7 @@ func (e *Engine) FindForeignRules() ([]ForeignRule, error) {
 
 	var foreign []ForeignRule
 
-	chains, err := e.conn.ListChainsOfTableFamily(nftables.TableFamilyINet)
+	chains, err := e.conn.ListChainsOfTableFamily(e.family())
 	if err != nil {
 		return nil, fmt.Errorf("listing chains: %w", err)
 	}

internal/nftables/compiler.go

@@ -70,21 +70,34 @@ func (c *Compiler) Compile() (*FirewallState, error) {
 }
 
 func (c *Compiler) compileConntrackFastPath(state *FirewallState) error {
+	invalid := c.cfg.Settings.InvalidDisposition
+	if invalid == "" {
+		invalid = config.PolicyDrop
+	}
 	for _, chain := range []string{"input", "forward", "output"} {
-		state.Rules[chain] = append(state.Rules[chain],
-			ManagedRule{
+		state.Rules[chain] = append(state.Rules[chain], ManagedRule{
+			Chain: chain,
+			Exprs: append(matchCtState(ctStateEstablished|ctStateRelated),
+				&expr.Verdict{Kind: expr.VerdictAccept}),
+			Tag: "ct:fastpath:" + chain,
+		})
+		for _, d := range []struct {
+			name   string
+			state  uint32
+			action config.PolicyAction
+		}{
+			{"invalid", ctStateInvalid, invalid},
+			{"untracked", ctStateUntracked, c.cfg.Settings.UntrackedDisposition},
+		} {
+			if d.action == "" || d.action == config.PolicyContinue {
+				continue
+			}
+			state.Rules[chain] = append(state.Rules[chain], ManagedRule{
 				Chain: chain,
-				Exprs: append(matchCtState(ctStateEstablished|ctStateRelated),
-					&expr.Verdict{Kind: expr.VerdictAccept}),
-				Tag: "ct:fastpath:" + chain,
-			},
-			ManagedRule{
-				Chain: chain,
-				Exprs: append(matchCtState(ctStateInvalid),
-					&expr.Verdict{Kind: expr.VerdictDrop}),
-				Tag: "ct:invalid:" + chain,
-			},
-		)
+				Exprs: append(matchCtState(d.state), policyVerdict(d.action, c.cfg.Settings.AddressFamily)...),
+				Tag:   "ct:" + d.name + ":" + chain,
+			})
+		}
 	}
 	return nil
 }
@@ -2049,6 +2062,13 @@ func rejectExprs(proto byte, family config.AddressFamily) []expr.Any {
 			Code: 0,
 		}}
 	}
+	// icmpx is inet-only; ip and ip6 tables silently drop on it.
+	switch family {
+	case config.FamilyIP:
+		return []expr.Any{&expr.Reject{Type: unix.NFT_REJECT_ICMP_UNREACH, Code: 3}} // port-unreachable
+	case config.FamilyIP6:
+		return []expr.Any{&expr.Reject{Type: unix.NFT_REJECT_ICMP_UNREACH, Code: 4}} // port-unreachable
+	}
 	return []expr.Any{&expr.Reject{
 		Type: unix.NFT_REJECT_ICMPX_UNREACH,
 		Code: unix.NFT_REJECT_ICMPX_PORT_UNREACH,

internal/nftables/compiler_test.go

@@ -3105,3 +3105,53 @@ func TestSpecCount_CommaAllMatchesExpansion(t *testing.T) {
 		}
 	}
 }
+
+func TestCompile_Dispositions(t *testing.T) {
+	cases := []struct {
+		invalid, untracked config.PolicyAction
+		want               map[string]expr.Any // tag prefix -> verdict expr, nil = absent
+	}{
+		{"", "", map[string]expr.Any{"ct:invalid:": &expr.Verdict{Kind: expr.VerdictDrop}, "ct:untracked:": nil}},
+		{config.PolicyContinue, config.PolicyContinue, map[string]expr.Any{"ct:invalid:": nil, "ct:untracked:": nil}},
+		{config.PolicyReject, config.PolicyAccept, map[string]expr.Any{
+			"ct:invalid:":   rejectExprs(0, config.FamilyINET)[0],
+			"ct:untracked:": &expr.Verdict{Kind: expr.VerdictAccept},
+		}},
+	}
+	for _, tc := range cases {
+		cfg := &config.Config{
+			Settings: config.Settings{
+				TableName:            "test",
+				AddressFamily:        config.FamilyINET,
+				InvalidDisposition:   tc.invalid,
+				UntrackedDisposition: tc.untracked,
+			},
+			Zones:      map[string]config.Zone{"fw": {Type: config.ZoneFirewall}, "net": {Type: config.ZoneIP}},
+			Interfaces: []config.Interface{{Zone: "net", Interface: "eth0"}},
+			Policy:     []config.Policy{{Source: "all", Dest: "all", Action: config.PolicyDrop}},
+			PortGroups: make(map[string]config.PortGroup),
+		}
+		state, err := NewCompiler(cfg).Compile()
+		if err != nil {
+			t.Fatalf("Compile: %v", err)
+		}
+		for _, chain := range []string{"input", "forward", "output"} {
+			for prefix, want := range tc.want {
+				var got *ManagedRule
+				for i, r := range state.Rules[chain] {
+					if r.Tag == prefix+chain {
+						got = &state.Rules[chain][i]
+					}
+				}
+				switch {
+				case want == nil && got != nil:
+					t.Errorf("%q/%q: unexpected %s%s rule", tc.invalid, tc.untracked, prefix, chain)
+				case want != nil && got == nil:
+					t.Errorf("%q/%q: missing %s%s rule", tc.invalid, tc.untracked, prefix, chain)
+				case want != nil && !reflect.DeepEqual(got.Exprs[len(got.Exprs)-1], want):
+					t.Errorf("%q/%q: %s%s verdict = %#v, want %#v", tc.invalid, tc.untracked, prefix, chain, got.Exprs[len(got.Exprs)-1], want)
+				}
+			}
+		}
+	}
+}

internal/nftables/engine.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/google/nftables"
 	"github.com/google/nftables/expr"
+	"github.com/mdlayher/netlink"
+	"golang.org/x/sys/unix"
 
 	"git.unkin.net/unkin/tomswall/internal/config"
 )
@@ -15,16 +17,101 @@ type Engine struct {
 }
 
 func NewEngine(cfg *config.Config) (*Engine, error) {
-	conn, err := nftables.New()
+	conn, err := nftables.New(nftables.WithSockOptions(largeBuffers))
 	if err != nil {
 		return nil, fmt.Errorf("connecting to nftables: %w", err)
 	}
 	return &Engine{cfg: cfg, conn: conn}, nil
 }
 
+var tableFamilies = map[config.AddressFamily]nftables.TableFamily{
+	config.FamilyINET: nftables.TableFamilyINet,
+	config.FamilyIP:   nftables.TableFamilyIPv4,
+	config.FamilyIP6:  nftables.TableFamilyIPv6,
+}
+
+func (e *Engine) family() nftables.TableFamily {
+	if f, ok := tableFamilies[e.cfg.Settings.AddressFamily]; ok {
+		return f
+	}
+	return nftables.TableFamilyINet
+}
+
+func addressFamily(tf nftables.TableFamily) config.AddressFamily {
+	for f, t := range tableFamilies {
+		if t == tf {
+			return f
+		}
+	}
+	return config.FamilyINET
+}
+
+// withFamily is the engine for the same table name in another address family.
+func (e *Engine) withFamily(f config.AddressFamily) *Engine {
+	cfg := *e.cfg
+	cfg.Settings.AddressFamily = f
+	return &Engine{cfg: &cfg, conn: e.conn}
+}
+
+// overlaps reports whether tables of families a and b filter the same traffic:
+// inet covers both ip and ip6, which do not overlap each other.
+func overlaps(a, b nftables.TableFamily) bool {
+	if a == b {
+		return false
+	}
+	return (a == nftables.TableFamilyINet && (b == nftables.TableFamilyIPv4 || b == nftables.TableFamilyIPv6)) ||
+		(b == nftables.TableFamilyINet && (a == nftables.TableFamilyIPv4 || a == nftables.TableFamilyIPv6))
+}
+
+// staleTables are our-named tables left by a different address_family that
+// would still filter the traffic this family now owns.
+func (e *Engine) staleTables() ([]*nftables.Table, error) {
+	tables, err := e.conn.ListTables()
+	if err != nil {
+		return nil, fmt.Errorf("listing tables: %w", err)
+	}
+	var stale []*nftables.Table
+	for _, t := range tables {
+		if t.Name == e.cfg.Settings.TableName && overlaps(e.family(), t.Family) {
+			stale = append(stale, t)
+		}
+	}
+	return stale, nil
+}
+
+// batchBufSize bounds one batch: the kernel rejects a batch larger than the
+// send buffer (EMSGSIZE) and drops ACKs beyond the receive buffer (ENOBUFS)
+// after committing it.
+// ponytail: fixed cap of tens of thousands of rules; size per batch if exceeded.
+const batchBufSize = 64 << 20
+
+// largeBuffers raises both socket buffers, ignoring rmem_max/wmem_max when
+// CAP_NET_ADMIN allows it and falling back to the capped sizes otherwise.
+func largeBuffers(c *netlink.Conn) error {
+	rc, err := c.SyscallConn()
+	if err != nil {
+		return err
+	}
+	var serr error
+	err = rc.Control(func(fd uintptr) {
+		for _, o := range [][2]int{{unix.SO_SNDBUFFORCE, unix.SO_SNDBUF}, {unix.SO_RCVBUFFORCE, unix.SO_RCVBUF}} {
+			if unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, o[0], batchBufSize) == nil {
+				continue
+			}
+			if serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, o[1], batchBufSize); serr != nil {
+				return
+			}
+		}
+	})
+	if err != nil {
+		return err
+	}
+	return serr
+}
+
 func (e *Engine) ensureTable() *nftables.Table {
 	return e.conn.AddTable(&nftables.Table{
-		Family: nftables.TableFamilyINet,
+		Family: e.family(),
 		Name:   e.cfg.Settings.TableName,
 	})
 }
@@ -132,6 +219,13 @@ func (e *Engine) Apply(changes *ChangeSet) error {
 }
 
 func (e *Engine) apply(changes *ChangeSet, policies map[string]nftables.ChainPolicy) error {
+	stale, err := e.staleTables()
+	if err != nil {
+		return err
+	}
+	for _, t := range stale {
+		e.conn.DelTable(t)
+	}
 	table := e.ensureTable()
 	chains := e.ensureChains(table, policies)
 
@@ -173,18 +267,24 @@ func (e *Engine) apply(changes *ChangeSet, policies map[string]nftables.ChainPol
 }
 
 func (e *Engine) Flush() error {
-	tables, err := e.conn.ListTables()
+	tables, err := e.staleTables()
 	if err != nil {
-		return fmt.Errorf("listing tables: %w", err)
+		return err
+	}
+	own, err := e.findTable()
+	if err != nil {
+		return err
+	}
+	if own != nil {
+		tables = append(tables, own)
+	}
+	if len(tables) == 0 {
+		return nil
 	}
-
 	for _, t := range tables {
-		if t.Name == e.cfg.Settings.TableName {
-			e.conn.DelTable(t)
-			return e.conn.Flush()
-		}
+		e.conn.DelTable(t)
 	}
-	return nil
+	return e.conn.Flush()
 }
 
 func (e *Engine) findTable() (*nftables.Table, error) {
@@ -193,7 +293,7 @@ func (e *Engine) findTable() (*nftables.Table, error) {
 		return nil, fmt.Errorf("listing tables: %w", err)
 	}
 	for _, t := range tables {
-		if t.Name == e.cfg.Settings.TableName && t.Family == nftables.TableFamilyINet {
+		if t.Name == e.cfg.Settings.TableName && t.Family == e.family() {
 			return t, nil
 		}
 	}
@@ -222,7 +322,7 @@ func (e *Engine) readCurrentState() (*FirewallState, error) {
 		}
 	}
 
-	chains, err := e.conn.ListChainsOfTableFamily(nftables.TableFamilyINet)
+	chains, err := e.conn.ListChainsOfTableFamily(e.family())
 	if err != nil {
 		return nil, fmt.Errorf("listing chains: %w", err)
 	}
@@ -252,6 +352,7 @@ func (e *Engine) readCurrentState() (*FirewallState, error) {
 // survives the process that took it.
 type Snapshot struct {
 	Table    string                          `json:"table"`
+	Family   config.AddressFamily            `json:"family,omitempty"`
 	Present  bool                            `json:"present"`
 	Policies map[string]nftables.ChainPolicy `json:"policies,omitempty"`
 	Rules    map[string][]SnapshotRule       `json:"rules,omitempty"`
@@ -264,16 +365,30 @@ type SnapshotRule struct {
 	Exprs [][]byte `json:"exprs"`
 }
 
-// Snapshot captures the live tomswall table so Restore can roll back to it.
+// Snapshot captures the live tomswall table so Restore can roll back to it,
+// falling back to the overlapping table of another family that apply replaces.
 func (e *Engine) Snapshot() (*Snapshot, error) {
-	snap := &Snapshot{Table: e.cfg.Settings.TableName}
 	t, err := e.findTable()
-	if err != nil || t == nil {
-		return snap, err
+	if err != nil {
+		return nil, err
+	}
+	if t == nil {
+		stale, err := e.staleTables()
+		if err != nil {
+			return nil, err
+		}
+		// ponytail: captures one stale table; an inet config replacing both ip and ip6 restores only the first.
+		if len(stale) > 0 {
+			return e.withFamily(addressFamily(stale[0].Family)).Snapshot()
+		}
+	}
+	snap := &Snapshot{Table: e.cfg.Settings.TableName, Family: addressFamily(e.family())}
+	if t == nil {
+		return snap, nil
 	}
 	snap.Present = true
 
-	chains, err := e.conn.ListChainsOfTableFamily(nftables.TableFamilyINet)
+	chains, err := e.conn.ListChainsOfTableFamily(e.family())
 	if err != nil {
 		return nil, fmt.Errorf("listing chains: %w", err)
 	}
@@ -288,7 +403,7 @@ func (e *Engine) Snapshot() (*Snapshot, error) {
 	if err != nil {
 		return nil, err
 	}
-	snap.Rules, err = encodeState(state)
+	snap.Rules, err = encodeState(state, byte(e.family()))
 	if err != nil {
 		return nil, err
 	}
@@ -302,10 +417,14 @@ func (e *Engine) Restore(s *Snapshot) error {
 	if s.Table != e.cfg.Settings.TableName {
 		return fmt.Errorf("snapshot is of table %q, engine manages %q", s.Table, e.cfg.Settings.TableName)
 	}
+	// Snapshots predating the family field are of the inet table.
+	if f := addressFamily(tableFamilies[s.Family]); f != addressFamily(e.family()) {
+		return e.withFamily(f).Restore(s)
+	}
 	if !s.Present {
 		return e.Flush()
 	}
-	want, err := decodeState(s.Rules)
+	want, err := decodeState(s.Rules, byte(e.family()))
 	if err != nil {
 		return err
 	}

internal/nftables/engine_test.go

@@ -0,0 +1,59 @@
+package nftables
+
+import (
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/mdlayher/netlink"
+	"golang.org/x/sys/unix"
+)
+
+func TestLargeBuffersRaisesSocketBuffers(t *testing.T) {
+	c, err := netlink.Dial(unix.NETLINK_NETFILTER, nil)
+	if err != nil {
+		t.Skipf("netlink unavailable: %v", err)
+	}
+	defer c.Close()
+	if err := largeBuffers(c); err != nil {
+		t.Fatal(err)
+	}
+	// Without CAP_NET_ADMIN the kernel caps at the sysctl max; it doubles either way.
+	for opt, sysctl := range map[int]string{unix.SO_RCVBUF: "rmem_max", unix.SO_SNDBUF: "wmem_max"} {
+		want := 2 * min(batchBufSize, procInt(t, "/proc/sys/net/core/"+sysctl))
+		if got := sockBuf(t, c, opt); got < want {
+			t.Errorf("%s-bounded buffer = %d, want >= %d", sysctl, got, want)
+		}
+	}
+}
+
+func procInt(t *testing.T, path string) int {
+	t.Helper()
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Skipf("reading %s: %v", path, err)
+	}
+	v, err := strconv.Atoi(strings.TrimSpace(string(b)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return v
+}
+
+func sockBuf(t *testing.T, c *netlink.Conn, opt int) int {
+	t.Helper()
+	rc, err := c.SyscallConn()
+	if err != nil {
+		t.Fatal(err)
+	}
+	var v int
+	var serr error
+	if err := rc.Control(func(fd uintptr) { v, serr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, opt) }); err != nil {
+		t.Fatal(err)
+	}
+	if serr != nil {
+		t.Fatal(serr)
+	}
+	return v
+}

internal/nftables/family_test.go

@@ -0,0 +1,113 @@
+package nftables
+
+import (
+	"testing"
+
+	"github.com/google/nftables"
+	"github.com/google/nftables/expr"
+	"github.com/mdlayher/netlink"
+	"golang.org/x/sys/unix"
+
+	"git.unkin.net/unkin/tomswall/internal/config"
+)
+
+type sentTable struct {
+	msg    int
+	family nftables.TableFamily
+}
+
+// familyEngine fakes a kernel holding tomswall tables of the given families
+// and records table creations/deletions.
+func familyEngine(t *testing.T, af config.AddressFamily, live ...nftables.TableFamily) (*Engine, *[]sentTable) {
+	var sent []sentTable
+	e := testEngine(t, func(req []netlink.Message) ([]netlink.Message, error) {
+		var out []netlink.Message
+		for _, m := range req {
+			switch m.Header.Type {
+			case nftType(unix.NFT_MSG_GETTABLE):
+				for _, f := range live {
+					attrs, _ := netlink.MarshalAttributes([]netlink.Attribute{{Type: unix.NFTA_TABLE_NAME, Data: []byte("tomswall\x00")}})
+					out = append(out, netlink.Message{Header: netlink.Header{Type: nftType(unix.NFT_MSG_NEWTABLE), Sequence: m.Header.Sequence}, Data: append([]byte{byte(f), 0, 0, 0}, attrs...)})
+				}
+			case nftType(unix.NFT_MSG_NEWTABLE):
+				sent = append(sent, sentTable{unix.NFT_MSG_NEWTABLE, nftables.TableFamily(m.Data[0])})
+			case nftType(unix.NFT_MSG_DELTABLE):
+				sent = append(sent, sentTable{unix.NFT_MSG_DELTABLE, nftables.TableFamily(m.Data[0])})
+			}
+		}
+		return out, nil
+	})
+	e.cfg.Settings.AddressFamily = af
+	return e, &sent
+}
+
+func TestApplyIPFamilyReplacesInetTable(t *testing.T) {
+	e, sent := familyEngine(t, config.FamilyIP, nftables.TableFamilyINet, nftables.TableFamilyIPv6)
+	if err := e.Apply(&ChangeSet{}); err != nil {
+		t.Fatal(err)
+	}
+	want := []sentTable{{unix.NFT_MSG_DELTABLE, nftables.TableFamilyINet}, {unix.NFT_MSG_NEWTABLE, nftables.TableFamilyIPv4}}
+	if len(*sent) != len(want) || (*sent)[0] != want[0] || (*sent)[1] != want[1] {
+		t.Errorf("got %+v, want %+v (the ip6 table must survive)", *sent, want)
+	}
+}
+
+func TestApplyInetFamilyReplacesIPTables(t *testing.T) {
+	e, sent := familyEngine(t, config.FamilyINET, nftables.TableFamilyIPv4, nftables.TableFamilyIPv6)
+	if err := e.Apply(&ChangeSet{}); err != nil {
+		t.Fatal(err)
+	}
+	var dels int
+	for _, s := range *sent {
+		if s.msg == unix.NFT_MSG_DELTABLE {
+			dels++
+		}
+	}
+	if dels != 2 {
+		t.Errorf("want ip and ip6 tables deleted, got %+v", *sent)
+	}
+}
+
+func TestFlushIPFamilyKeepsIP6Table(t *testing.T) {
+	e, sent := familyEngine(t, config.FamilyIP, nftables.TableFamilyIPv6, nftables.TableFamilyIPv4)
+	if err := e.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if len(*sent) != 1 || (*sent)[0] != (sentTable{unix.NFT_MSG_DELTABLE, nftables.TableFamilyIPv4}) {
+		t.Errorf("got %+v, want only the ip table deleted", *sent)
+	}
+}
+
+func TestSnapshotFallsBackToReplacedTable(t *testing.T) {
+	e, _ := familyEngine(t, config.FamilyIP, nftables.TableFamilyINet)
+	snap, err := e.Snapshot()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !snap.Present || snap.Family != config.FamilyINET {
+		t.Fatalf("want present inet snapshot, got %+v", snap)
+	}
+
+	// Reverting to the inet snapshot drops the tried ip table.
+	e, sent := familyEngine(t, config.FamilyIP, nftables.TableFamilyIPv4)
+	if err := e.Restore(&Snapshot{Table: "tomswall", Family: config.FamilyINET, Present: true}); err != nil {
+		t.Fatal(err)
+	}
+	want := []sentTable{{unix.NFT_MSG_DELTABLE, nftables.TableFamilyIPv4}, {unix.NFT_MSG_NEWTABLE, nftables.TableFamilyINet}}
+	if len(*sent) != 2 || (*sent)[0] != want[0] || (*sent)[1] != want[1] {
+		t.Errorf("got %+v, want %+v", *sent, want)
+	}
+}
+
+func TestRejectExprsFamily(t *testing.T) {
+	for af, want := range map[config.AddressFamily]expr.Reject{
+		config.FamilyINET: {Type: unix.NFT_REJECT_ICMPX_UNREACH, Code: unix.NFT_REJECT_ICMPX_PORT_UNREACH},
+		config.FamilyIP:   {Type: unix.NFT_REJECT_ICMP_UNREACH, Code: 3},
+		config.FamilyIP6:  {Type: unix.NFT_REJECT_ICMP_UNREACH, Code: 4},
+	} {
+		got := rejectExprs(unix.IPPROTO_UDP, af)[0].(*expr.Reject)
+		if *got != want {
+			t.Errorf("%s: got %+v, want %+v", af, *got, want)
+		}
+	}
+}

internal/nftables/snapshot.go

@@ -4,14 +4,11 @@ import (
 	"encoding/binary"
 	"fmt"
 
-	"github.com/google/nftables"
 	"github.com/google/nftables/expr"
 	"github.com/mdlayher/netlink"
 	"golang.org/x/sys/unix"
 )
 
-const inet = byte(nftables.TableFamilyINet)
-
 // exprByName mirrors the expression types google/nftables can parse back from the kernel.
 var exprByName = map[string]func() expr.Any{
 	"ct":           func() expr.Any { return &expr.Ct{} },
@@ -40,13 +37,13 @@ var exprByName = map[string]func() expr.Any{
 	"notrack":      func() expr.Any { return &expr.Notrack{} },
 }
 
-func encodeState(state *FirewallState) (map[string][]SnapshotRule, error) {
+func encodeState(state *FirewallState, fam byte) (map[string][]SnapshotRule, error) {
 	out := make(map[string][]SnapshotRule, len(state.Rules))
 	for chain, rules := range state.Rules {
 		for _, r := range rules {
 			sr := SnapshotRule{Tag: r.Tag}
 			for _, e := range r.Exprs {
-				b, err := expr.Marshal(inet, e)
+				b, err := expr.Marshal(fam, e)
 				if err != nil {
 					return nil, fmt.Errorf("encoding %s rule %q: %w", chain, r.Tag, err)
 				}
@@ -58,13 +55,13 @@ func encodeState(state *FirewallState) (map[string][]SnapshotRule, error) {
 	return out, nil
 }
 
-func decodeState(rules map[string][]SnapshotRule) (*FirewallState, error) {
+func decodeState(rules map[string][]SnapshotRule, fam byte) (*FirewallState, error) {
 	state := &FirewallState{Rules: make(map[string][]ManagedRule, len(rules))}
 	for chain, rs := range rules {
 		for _, sr := range rs {
 			r := ManagedRule{Chain: chain, Tag: sr.Tag}
 			for _, b := range sr.Exprs {
-				e, err := decodeExpr(b)
+				e, err := decodeExpr(b, fam)
 				if err != nil {
 					return nil, fmt.Errorf("decoding %s rule %q: %w", chain, sr.Tag, err)
 				}
@@ -77,7 +74,7 @@ func decodeState(rules map[string][]SnapshotRule) (*FirewallState, error) {
 }
 
 // decodeExpr reverses expr.Marshal, as google/nftables does when reading rules.
-func decodeExpr(b []byte) (expr.Any, error) {
+func decodeExpr(b []byte, fam byte) (expr.Any, error) {
 	ad, err := netlink.NewAttributeDecoder(b)
 	if err != nil {
 		return nil, err
@@ -104,13 +101,13 @@ func decodeExpr(b []byte) (expr.Any, error) {
 	if name == "notrack" {
 		return e, nil
 	}
-	if err := expr.Unmarshal(inet, data, e); err != nil {
+	if err := expr.Unmarshal(fam, data, e); err != nil {
 		return nil, err
 	}
 	// A verdict is an immediate into the verdict register with no data.
 	if imm, ok := e.(*expr.Immediate); ok && imm.Register == unix.NFT_REG_VERDICT && len(imm.Data) == 0 {
 		v := &expr.Verdict{}
-		if err := expr.Unmarshal(inet, data, v); err != nil {
+		if err := expr.Unmarshal(fam, data, v); err != nil {
 			return nil, err
 		}
 		return v, nil

internal/nftables/snapshot_test.go

@@ -29,7 +29,7 @@ func TestSnapshotRulesRoundTrip(t *testing.T) {
 		"input": {{Chain: "input", Tag: "ssh", Exprs: exprs}, {Chain: "input", Tag: "drop", Exprs: []expr.Any{&expr.Verdict{Kind: expr.VerdictDrop}}}},
 	}}
 
-	rules, err := encodeState(state)
+	rules, err := encodeState(state, byte(nftables.TableFamilyINet))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -45,7 +45,7 @@ func TestSnapshotRulesRoundTrip(t *testing.T) {
 	if snap.Policies["input"] != nftables.ChainPolicyAccept {
 		t.Errorf("policy lost: %v", snap.Policies)
 	}
-	got, err := decodeState(snap.Rules)
+	got, err := decodeState(snap.Rules, byte(nftables.TableFamilyINet))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -79,7 +79,7 @@ func TestSnapshotAndRestoreAbsentTable(t *testing.T) {
 		for _, m := range req {
 			sent = append(sent, m.Header.Type)
 			if m.Header.Type == nftType(unix.NFT_MSG_GETTABLE) && tablePresent {
-				data := []byte{inet, 0, 0, 0}
+				data := []byte{byte(nftables.TableFamilyINet), 0, 0, 0}
 				attrs, _ := netlink.MarshalAttributes([]netlink.Attribute{{Type: unix.NFTA_TABLE_NAME, Data: []byte("tomswall\x00")}})
 				return []netlink.Message{{Header: netlink.Header{Type: nftType(unix.NFT_MSG_NEWTABLE), Sequence: m.Header.Sequence}, Data: append(data, attrs...)}}, nil
 			}
@@ -116,7 +116,7 @@ func TestRestorePresentTable(t *testing.T) {
 		{Tag: "ssh", Exprs: []expr.Any{&expr.Ct{Register: 1, Key: expr.CtKeySTATE}, &expr.Verdict{Kind: expr.VerdictAccept}}},
 		{Tag: "drop", Exprs: []expr.Any{&expr.Verdict{Kind: expr.VerdictDrop}}},
 	} {
-		enc, err := encodeState(&FirewallState{Rules: map[string][]ManagedRule{"input": {r}}})
+		enc, err := encodeState(&FirewallState{Rules: map[string][]ManagedRule{"input": {r}}}, byte(nftables.TableFamilyINet))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -132,7 +132,7 @@ func TestRestorePresentTable(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		return append([]byte{inet, 0, 0, 0}, b...)
+		return append([]byte{byte(nftables.TableFamilyINet), 0, 0, 0}, b...)
 	}
 	handle := make([]byte, 8)
 	binary.BigEndian.PutUint64(handle, 7)

internal/shorewall/convert.go

@@ -99,6 +99,15 @@ func convertDir(dir string, ipv6 bool) (*config.Config, error) {
 	return cfg, nil
 }
 
+// disposition maps a shorewall *_DISPOSITION value; unset means CONTINUE and A_ (audit) variants map to their base action.
+func disposition(v string) config.PolicyAction {
+	v = strings.TrimPrefix(strings.ToLower(v), "a_")
+	if v == "" {
+		return config.PolicyContinue
+	}
+	return config.PolicyAction(v)
+}
+
 func subst(s string, params map[string]string) string {
 	if !strings.Contains(s, "$") {
 		return s
@@ -115,6 +124,8 @@ func convertConf(dir string, cfg *config.Config, params map[string]string, ipv6
 	if err != nil {
 		return err
 	}
+	cfg.Settings.InvalidDisposition = disposition(conf["INVALID_DISPOSITION"])
+	cfg.Settings.UntrackedDisposition = disposition(conf["UNTRACKED_DISPOSITION"])
 	if conf == nil {
 		return nil
 	}
@@ -136,7 +147,6 @@ func convertConf(dir string, cfg *config.Config, params map[string]string, ipv6
 	if v, ok := conf["IMPLICIT_CONTINUE"]; ok {
 		cfg.Settings.ImplicitContinue = v == "Yes"
 	}
-
 	return nil
 }
 

internal/shorewall/convert_test.go

@@ -725,3 +725,34 @@ func TestIsIPv6Dir(t *testing.T) {
 		}
 	})
 }
+
+func TestConvert_Dispositions(t *testing.T) {
+	cases := []struct {
+		conf               string
+		invalid, untracked config.PolicyAction
+	}{
+		{"", config.PolicyContinue, config.PolicyContinue},
+		{"IP_FORWARDING=Yes", config.PolicyContinue, config.PolicyContinue},
+		{"INVALID_DISPOSITION=CONTINUE\nUNTRACKED_DISPOSITION=ACCEPT", config.PolicyContinue, config.PolicyAccept},
+		{"INVALID_DISPOSITION=DROP\nUNTRACKED_DISPOSITION=A_DROP", config.PolicyDrop, config.PolicyDrop},
+		{"INVALID_DISPOSITION=A_REJECT", config.PolicyReject, config.PolicyContinue},
+	}
+	for _, tc := range cases {
+		dir := t.TempDir()
+		writeFile(t, dir, "shorewall.conf", tc.conf)
+		writeFile(t, dir, "zones", "fw firewall\nnet ipv4\n")
+		writeFile(t, dir, "interfaces", "net eth0 -\n")
+		writeFile(t, dir, "policy", "all all DROP\n")
+		cfg, err := Convert(dir)
+		if err != nil {
+			t.Fatalf("Convert(%q): %v", tc.conf, err)
+		}
+		if cfg.Settings.InvalidDisposition != tc.invalid || cfg.Settings.UntrackedDisposition != tc.untracked {
+			t.Errorf("%q: got invalid=%q untracked=%q, want %q/%q", tc.conf,
+				cfg.Settings.InvalidDisposition, cfg.Settings.UntrackedDisposition, tc.invalid, tc.untracked)
+		}
+		if err := cfg.Validate(); err != nil {
+			t.Errorf("%q: Validate: %v", tc.conf, err)
+		}
+	}
+}

internal/tryapply/tryapply.go

@@ -175,10 +175,28 @@ func Revert(id string) (reverted bool, err error) {
 	if err != nil || p == nil || (id != "" && p.ID != id) {
 		return false, err
 	}
-	if err := restore(p.Snapshot); err != nil {
-		return false, fmt.Errorf("restoring snapshot: %w", err)
+	return true, restorePending(p)
+}
+
+// Abort restores the pending snapshot after a failed apply, which may have
+// committed partially. A failed restore keeps the snapshot and timer so the
+// timer still reverts. The caller must hold the lock.
+func Abort() error {
+	p, err := load()
+	if err != nil {
+		return err
 	}
-	return true, Discard()
+	if p == nil {
+		return errors.New("no pending try to abort")
+	}
+	return restorePending(p)
+}
+
+func restorePending(p *pending) error {
+	if err := restore(p.Snapshot); err != nil {
+		return fmt.Errorf("restoring snapshot: %w", err)
+	}
+	return Discard()
 }
 
 func load() (*pending, error) {

internal/tryapply/tryapply_test.go

@@ -209,3 +209,41 @@ func TestRevertStaleIDIgnored(t *testing.T) {
 		t.Errorf("newer try's snapshot removed: %v", err)
 	}
 }
+
+func TestAbortRestoresAndDisarms(t *testing.T) {
+	cmds := setup(t)
+	restored := stubRestore(t, nil)
+	snap := &nftables.Snapshot{Table: "tomswall", Present: true}
+	arm(t, snap)
+
+	if err := Abort(); err != nil {
+		t.Fatal(err)
+	}
+	if len(*restored) != 1 || !reflect.DeepEqual((*restored)[0], snap) {
+		t.Errorf("restored %+v, want the armed snapshot", *restored)
+	}
+	if _, err := os.Stat(snapshotPath()); !os.IsNotExist(err) {
+		t.Error("snapshot not removed")
+	}
+	if last := (*cmds)[len(*cmds)-1]; last != "systemctl stop "+Unit+".timer" {
+		t.Errorf("timer not stopped, last command %q", last)
+	}
+}
+
+func TestAbortFailureKeepsSnapshotAndTimer(t *testing.T) {
+	cmds := setup(t)
+	boom := errors.New("netlink down")
+	stubRestore(t, boom)
+	arm(t, &nftables.Snapshot{Table: "tomswall"})
+	armed := len(*cmds)
+
+	if err := Abort(); !errors.Is(err, boom) {
+		t.Fatalf("Abort error = %v, want %v", err, boom)
+	}
+	if _, err := os.Stat(snapshotPath()); err != nil {
+		t.Fatalf("snapshot gone after failed abort: %v", err)
+	}
+	if len(*cmds) != armed {
+		t.Errorf("revert timer touched after failed abort: %v", (*cmds)[armed:])
+	}
+}

tomswall.example.yaml

@@ -8,6 +8,10 @@ settings:
   log_level: info
   table_name: tomswall
   implicit_continue: false
+  # ct state invalid/untracked verdict: accept, drop, reject, continue (pass to rules)
+  # defaults: invalid drop, untracked continue (migrate defaults both to continue, as shorewall)
+  invalid_disposition: drop
+  untracked_disposition: continue
 
 # Named port groups — reusable port+protocol combos referenced in rules
 portgroups: