feat(helper): add privileged transactional core

internal/helper/helper.go

@@ -0,0 +1,708 @@
+// SPDX-License-Identifier: MIT
+
+// Package helper implements the privileged core invoked by the unprivileged
+// prx CLI through pkexec.
+//
+// Every plan is re-validated here before any system change is made, and the
+// apply operation is transactional: on any failure the changes made so far are
+// rolled back. Concurrent apply/remove operations are serialised with an
+// advisory file lock.
+package helper
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"prx/internal/config"
+	"prx/internal/engine"
+	"prx/internal/execx"
+	"prx/internal/fsx"
+	"prx/internal/health"
+	"prx/internal/iprule"
+	"prx/internal/netdetect"
+	"prx/internal/nft"
+	"prx/internal/proto"
+	"prx/internal/state"
+)
+
+// healthTimeout bounds the post-apply connectivity probe.
+const healthTimeout = 3 * time.Second
+
+// Execute runs a single privileged request.
+func Execute(ctx context.Context, r execx.Runner, plan proto.Plan) proto.Result {
+	switch plan.Action {
+	case proto.ActionApply:
+		return apply(ctx, r, plan)
+	case proto.ActionRemove:
+		return remove(ctx, r, plan)
+	case proto.ActionInspect:
+		return inspect(ctx, r, plan)
+	case proto.ActionEngine:
+		return engineAction(ctx, r, plan)
+	case proto.ActionInitEngine:
+		return initEngine(plan)
+	default:
+		return proto.Result{Action: plan.Action, Error: fmt.Sprintf("unknown action %q", plan.Action)}
+	}
+}
+
+// rollback records undo actions in the order they must be reversed.
+type rollback struct {
+	fns []func(context.Context) error
+}
+
+func (rb *rollback) add(fn func(context.Context) error) { rb.fns = append(rb.fns, fn) }
+
+// run executes the undo actions in reverse order and returns every failure so
+// the caller can report that rollback was incomplete.
+func (rb *rollback) run(ctx context.Context) []error {
+	var errs []error
+	for i := len(rb.fns) - 1; i >= 0; i-- {
+		if err := rb.fns[i](ctx); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errs
+}
+
+func apply(ctx context.Context, r execx.Runner, plan proto.Plan) (res proto.Result) {
+	res = proto.Result{Action: proto.ActionApply}
+	if err := validatePlan(plan); err != nil {
+		res.Error = err.Error()
+		return res
+	}
+	if err := preflight(r, plan.ManageEngine && !plan.SkipEngine && plan.ServiceName != ""); err != nil {
+		res.Error = err.Error()
+		return res
+	}
+
+	// A dry run makes no changes, so it does not need the state lock.
+	if !plan.DryRun {
+		unlock, err := lockState(plan.StatePath)
+		if err != nil {
+			res.Error = err.Error()
+			return res
+		}
+		defer func() {
+			if uerr := unlock(); uerr != nil {
+				res.Messages = append(res.Messages, "warning: "+uerr.Error())
+			}
+		}()
+	}
+
+	wan, err := netdetect.Detect(ctx, r, plan.WANInterface, plan.WANGateway4, plan.WANGateway6)
+	if err != nil {
+		res.Error = err.Error()
+		return res
+	}
+	if wan.Gateway4 == "" && wan.Gateway6 == "" {
+		res.Error = "no default gateway detected; cannot install policy routing"
+		return res
+	}
+
+	script, err := nft.Script(nft.Params{
+		WANInterface: wan.Interface,
+		Mark:         plan.Mark,
+		Local4:       wan.Local4,
+		Local6:       wan.Local6,
+	})
+	if err != nil {
+		res.Error = err.Error()
+		return res
+	}
+
+	if plan.DryRun {
+		if err := nft.Check(ctx, r, script); err != nil {
+			res.Error = "dry-run validation failed: " + err.Error()
+			return res
+		}
+		res.OK = true
+		res.Messages = append(res.Messages, "would protect replies: "+describeTarget(wan, plan.Mark))
+		return res
+	}
+
+	rb := &rollback{}
+	changed := false
+
+	// fail aborts the transaction: it undoes everything applied so far and
+	// reports any rollback step that itself failed, so nothing is ever lost
+	// silently.
+	fail := func(msg string) proto.Result {
+		res.Error = msg
+		var failed int
+		for _, rbErr := range rb.run(ctx) {
+			failed++
+			res.Messages = append(res.Messages, "rollback warning: "+rbErr.Error())
+		}
+		if failed > 0 {
+			res.Error = fmt.Sprintf("%s (rollback incomplete: %d step(s) failed — inspect with 'prx doctor')", msg, failed)
+		}
+		return res
+	}
+
+	// 1) Protect inbound replies before the engine takes over routing.
+	if err := nft.Apply(ctx, r, script); err != nil {
+		return fail("apply nftables rules: " + err.Error())
+	}
+	rb.add(func(c context.Context) error { return nft.Remove(c, r) })
+	changed = true
+
+	// 2) Policy routing for marked packets. The rollback for a table is
+	// registered before the table is touched so that a failure part-way
+	// through still removes anything that was added.
+	for _, target := range []struct {
+		family   int
+		gateway  string
+		table    int
+		priority int
+	}{
+		{family: 4, gateway: wan.Gateway4, table: plan.TableV4, priority: plan.PriorityV4},
+		{family: 6, gateway: wan.Gateway6, table: plan.TableV6, priority: plan.PriorityV6},
+	} {
+		if target.gateway == "" {
+			continue
+		}
+		family, table := target.family, target.table
+		rb.add(func(c context.Context) error { return removeTable(c, r, family, table) })
+		spec := iprule.Spec{
+			Family:   family,
+			Mark:     plan.Mark,
+			Table:    table,
+			Priority: target.priority,
+			Gateway:  target.gateway,
+			Dev:      wan.Interface,
+		}
+		if err := ensureRouting(ctx, r, spec, &changed); err != nil {
+			return fail(err.Error())
+		}
+	}
+
+	// 3) Destination-based bypass (direct) routing.
+	if err := applyBypass(ctx, r, plan, wan, rb, &changed); err != nil {
+		return fail(err.Error())
+	}
+
+	// 4) Ensure the engine is running.
+	engineWasRunning := false
+	if plan.ManageEngine && !plan.SkipEngine && plan.ServiceName != "" {
+		active, err := engine.IsActive(ctx, r, plan.ServiceName)
+		if err != nil {
+			return fail(err.Error())
+		}
+		engineWasRunning = active
+		if !active {
+			if err := engine.Start(ctx, r, plan.ServiceName); err != nil {
+				return fail(err.Error())
+			}
+			changed = true
+			service := plan.ServiceName
+			rb.add(func(c context.Context) error { return engine.Stop(c, r, service) })
+		}
+	}
+
+	// 5) Health checks.
+	checks, fatalOK := runChecks(ctx, r, plan, wan)
+	res.Checks = checks
+	if !fatalOK {
+		return fail("post-apply health checks failed; changes rolled back")
+	}
+
+	// 6) Record state.
+	st := &state.State{
+		Enabled:          true,
+		WANInterface:     wan.Interface,
+		Gateway4:         wan.Gateway4,
+		Gateway6:         wan.Gateway6,
+		Mark:             plan.Mark,
+		TableV4:          plan.TableV4,
+		TableV6:          plan.TableV6,
+		PriorityV4:       plan.PriorityV4,
+		PriorityV6:       plan.PriorityV6,
+		BypassTableV4:    plan.BypassTableV4,
+		BypassTableV6:    plan.BypassTableV6,
+		BypassPrioV4:     plan.BypassPrioV4,
+		BypassPrioV6:     plan.BypassPrioV6,
+		BypassCIDRs:      plan.BypassCIDRs,
+		EngineManaged:    plan.ManageEngine && !plan.SkipEngine,
+		EngineWasRunning: engineWasRunning,
+		EnabledAt:        time.Now().UTC(),
+	}
+	if err := state.Save(plan.StatePath, st); err != nil {
+		return fail(err.Error())
+	}
+
+	res.OK = true
+	res.Changed = changed
+	res.State = &proto.StateInfo{
+		Enabled:      true,
+		WANInterface: wan.Interface,
+		Gateway4:     wan.Gateway4,
+		Gateway6:     wan.Gateway6,
+		Mark:         plan.Mark,
+	}
+	res.Messages = append(res.Messages, "proxy routing enabled: "+describeTarget(wan, plan.Mark))
+	return res
+}
+
+// describeTarget renders a human readable summary of the applied configuration.
+func describeTarget(wan netdetect.WAN, mark uint32) string {
+	parts := []string{"interface " + wan.Interface}
+	if wan.Gateway4 != "" {
+		parts = append(parts, "gw4 "+wan.Gateway4)
+	}
+	if wan.Gateway6 != "" {
+		parts = append(parts, "gw6 "+wan.Gateway6)
+	}
+	parts = append(parts, fmt.Sprintf("mark 0x%x", mark))
+	return strings.Join(parts, ", ")
+}
+
+func ensureRouting(ctx context.Context, r execx.Runner, spec iprule.Spec, changed *bool) error {
+	ch, err := iprule.EnsureRule(ctx, r, spec)
+	if err != nil {
+		return err
+	}
+	*changed = *changed || ch
+
+	ch, err = iprule.EnsureRoute(ctx, r, spec.Family, spec.Table, spec.Gateway, spec.Dev)
+	if err != nil {
+		return err
+	}
+	*changed = *changed || ch
+	return nil
+}
+
+func applyBypass(ctx context.Context, r execx.Runner, plan proto.Plan, wan netdetect.WAN, rb *rollback, changed *bool) error {
+	registeredV4, registeredV6 := false, false
+	for _, cidr := range plan.BypassCIDRs {
+		ip, _, err := net.ParseCIDR(cidr)
+		if err != nil {
+			return fmt.Errorf("invalid bypass CIDR %q: %w", cidr, err)
+		}
+
+		family, table, prio, gw := 4, plan.BypassTableV4, plan.BypassPrioV4, wan.Gateway4
+		if ip.To4() == nil {
+			family, table, prio, gw = 6, plan.BypassTableV6, plan.BypassPrioV6, wan.Gateway6
+		}
+		if gw == "" {
+			// No gateway for this family on this host; skip silently.
+			continue
+		}
+
+		// Register the table cleanup before touching it.
+		if family == 4 && !registeredV4 {
+			t := table
+			rb.add(func(c context.Context) error { return removeTable(c, r, 4, t) })
+			registeredV4 = true
+		}
+		if family == 6 && !registeredV6 {
+			t := table
+			rb.add(func(c context.Context) error { return removeTable(c, r, 6, t) })
+			registeredV6 = true
+		}
+
+		if ch, err := iprule.EnsureCIDRRoute(ctx, r, family, table, cidr, gw, wan.Interface); err != nil {
+			return err
+		} else {
+			*changed = *changed || ch
+		}
+		spec := iprule.BypassSpec{Family: family, CIDR: cidr, Table: table, Priority: prio, Gateway: gw, Dev: wan.Interface}
+		if ch, err := iprule.EnsureBypassRule(ctx, r, spec); err != nil {
+			return err
+		} else {
+			*changed = *changed || ch
+		}
+	}
+	return nil
+}
+
+func removeTable(ctx context.Context, r execx.Runner, family, table int) error {
+	if _, err := iprule.RemoveRulesByTable(ctx, r, family, table); err != nil {
+		return err
+	}
+	return iprule.FlushTable(ctx, r, family, table)
+}
+
+func remove(ctx context.Context, r execx.Runner, plan proto.Plan) (res proto.Result) {
+	res = proto.Result{Action: proto.ActionRemove}
+	if err := validatePlan(plan); err != nil {
+		res.Error = err.Error()
+		return res
+	}
+	if err := preflight(r, plan.StopEngineOnOff && plan.ManageEngine && plan.ServiceName != ""); err != nil {
+		res.Error = err.Error()
+		return res
+	}
+
+	unlock, err := lockState(plan.StatePath)
+	if err != nil {
+		res.Error = err.Error()
+		return res
+	}
+	defer func() {
+		if uerr := unlock(); uerr != nil {
+			res.Messages = append(res.Messages, "warning: "+uerr.Error())
+		}
+	}()
+
+	var msgs []string
+	for _, family := range []int{4, 6} {
+		table := plan.TableV4
+		bp := plan.BypassTableV4
+		if family == 6 {
+			table = plan.TableV6
+			bp = plan.BypassTableV6
+		}
+		for _, spec := range []struct {
+			label string
+			table int
+		}{
+			{label: fmt.Sprintf("ipv%d policy", family), table: table},
+			{label: fmt.Sprintf("ipv%d bypass", family), table: bp},
+		} {
+			n, err := iprule.RemoveRulesByTable(ctx, r, family, spec.table)
+			if err != nil {
+				res.Error = err.Error()
+				return res
+			}
+			if n > 0 {
+				msgs = append(msgs, fmt.Sprintf("removed %d %s rule(s)", n, spec.label))
+			}
+			if err := iprule.FlushTable(ctx, r, family, spec.table); err != nil {
+				res.Error = err.Error()
+				return res
+			}
+		}
+	}
+
+	if err := nft.Remove(ctx, r); err != nil {
+		res.Error = "remove nftables table: " + err.Error()
+		return res
+	}
+
+	if plan.StopEngineOnOff && plan.ManageEngine && plan.ServiceName != "" {
+		if err := engine.Stop(ctx, r, plan.ServiceName); err != nil {
+			res.Messages = append(res.Messages, "warning: "+err.Error())
+		} else {
+			msgs = append(msgs, "stopped engine "+plan.ServiceName)
+		}
+	}
+
+	if err := state.Remove(plan.StatePath); err != nil {
+		res.Messages = append(res.Messages, "warning: "+err.Error())
+	}
+
+	res.OK = true
+	res.Changed = true
+	res.Messages = append(msgs, "proxy routing disabled")
+	res.State = &proto.StateInfo{Enabled: false}
+	return res
+}
+
+func inspect(ctx context.Context, r execx.Runner, plan proto.Plan) proto.Result {
+	res := proto.Result{Action: proto.ActionInspect, OK: true}
+
+	exists, err := nft.Exists(ctx, r)
+	if err != nil {
+		res.Checks = append(res.Checks, proto.Check{Name: "nft-table", OK: false, Fatal: true, Detail: err.Error()})
+	} else {
+		res.Checks = append(res.Checks, proto.Check{Name: "nft-table", OK: exists, Fatal: true, Detail: nft.Table()})
+	}
+
+	wan, err := netdetect.Detect(ctx, r, plan.WANInterface, plan.WANGateway4, plan.WANGateway6)
+	if err != nil {
+		res.Checks = append(res.Checks, proto.Check{Name: "wan", OK: false, Detail: err.Error(), Fatal: true})
+		return res
+	}
+	res.Checks = append(res.Checks, proto.Check{Name: "wan", OK: true, Detail: describeTarget(wan, plan.Mark)})
+
+	for _, target := range []struct {
+		family   int
+		gateway  string
+		table    int
+		priority int
+	}{
+		{family: 4, gateway: wan.Gateway4, table: plan.TableV4, priority: plan.PriorityV4},
+		{family: 6, gateway: wan.Gateway6, table: plan.TableV6, priority: plan.PriorityV6},
+	} {
+		if target.gateway == "" {
+			// No gateway for this family on this host: nothing to check.
+			continue
+		}
+		spec := iprule.Spec{Family: target.family, Mark: plan.Mark, Table: target.table, Priority: target.priority}
+		ruleOK, err := iprule.RuleExists(ctx, r, spec)
+		res.Checks = append(res.Checks, proto.Check{
+			Name:   fmt.Sprintf("ipv%d-rule", target.family),
+			OK:     ruleOK && err == nil,
+			Fatal:  true,
+			Detail: errDetail(err, fmt.Sprintf("fwmark 0x%x -> table %d prio %d", plan.Mark, target.table, target.priority)),
+		})
+		routeOK, err := iprule.TableHasRoutes(ctx, r, target.family, target.table)
+		res.Checks = append(res.Checks, proto.Check{
+			Name:   fmt.Sprintf("ipv%d-route", target.family),
+			OK:     routeOK && err == nil,
+			Fatal:  true,
+			Detail: errDetail(err, "table "+strconv.Itoa(target.table)),
+		})
+	}
+
+	if plan.ManageEngine && plan.ServiceName != "" {
+		active, err := engine.IsActive(ctx, r, plan.ServiceName)
+		res.Checks = append(res.Checks, proto.Check{Name: "engine", OK: active && err == nil, Warn: true, Detail: errDetail(err, plan.ServiceName)})
+	}
+
+	tuns, tunErr := netdetect.TunInterfaces(ctx, r)
+	tunDetail := strings.Join(tuns, ", ")
+	if tunErr != nil {
+		tunDetail = errDetail(tunErr, tunDetail)
+	}
+	res.Checks = append(res.Checks, proto.Check{
+		Name:   "tun",
+		OK:     tunErr == nil && len(tuns) > 0,
+		Warn:   true,
+		Detail: tunDetail,
+	})
+
+	// A corrupt state file must be reported rather than silently reported as
+	// "disabled".
+	st, loadErr := state.Load(plan.StatePath)
+	if loadErr != nil {
+		res.Checks = append(res.Checks, proto.Check{
+			Name:   "state",
+			OK:     false,
+			Warn:   true,
+			Detail: loadErr.Error(),
+		})
+	}
+	res.State = &proto.StateInfo{
+		Enabled:      st.Enabled,
+		WANInterface: st.WANInterface,
+		Gateway4:     st.Gateway4,
+		Gateway6:     st.Gateway6,
+		Mark:         st.Mark,
+	}
+	return res
+}
+
+func engineAction(ctx context.Context, r execx.Runner, plan proto.Plan) proto.Result {
+	res := proto.Result{Action: proto.ActionEngine}
+	if plan.ServiceName == "" || !config.ValidServiceName(plan.ServiceName) {
+		res.Error = fmt.Sprintf("invalid service name %q", plan.ServiceName)
+		return res
+	}
+	switch plan.EngineOp {
+	case proto.EngineStart:
+		if err := engine.Start(ctx, r, plan.ServiceName); err != nil {
+			res.Error = err.Error()
+			return res
+		}
+	case proto.EngineStop:
+		if err := engine.Stop(ctx, r, plan.ServiceName); err != nil {
+			res.Error = err.Error()
+			return res
+		}
+	case proto.EngineReload:
+		if err := engine.Reload(ctx, r, plan.ServiceName); err != nil {
+			res.Error = err.Error()
+			return res
+		}
+	case proto.EngineStatus:
+		active, err := engine.IsActive(ctx, r, plan.ServiceName)
+		if err != nil {
+			res.Error = err.Error()
+			return res
+		}
+		res.OK = true
+		res.Messages = append(res.Messages, fmt.Sprintf("%s active=%v", plan.ServiceName, active))
+		return res
+	default:
+		res.Error = fmt.Sprintf("unknown engine operation %q", plan.EngineOp)
+		return res
+	}
+	res.OK = true
+	res.Changed = true
+	return res
+}
+
+func initEngine(plan proto.Plan) proto.Result {
+	res := proto.Result{Action: proto.ActionInitEngine}
+	if plan.EngineConfig == "" || !filepath.IsAbs(plan.EngineConfig) {
+		res.Error = "engine config path must be absolute"
+		return res
+	}
+	if len(plan.EngineConfigBody) == 0 {
+		res.Error = "no engine config body supplied"
+		return res
+	}
+	// Reject anything that is not valid JSON before writing it to disk.
+	var probe json.RawMessage
+	if err := json.Unmarshal(plan.EngineConfigBody, &probe); err != nil {
+		res.Error = "engine config is not valid JSON: " + err.Error()
+		return res
+	}
+	if err := fsx.WriteFileAtomic(plan.EngineConfig, plan.EngineConfigBody, 0o644); err != nil {
+		res.Error = err.Error()
+		return res
+	}
+	res.OK = true
+	res.Changed = true
+	res.Messages = append(res.Messages, "wrote engine config to "+plan.EngineConfig)
+	return res
+}
+
+func runChecks(ctx context.Context, r execx.Runner, plan proto.Plan, wan netdetect.WAN) ([]proto.Check, bool) {
+	var checks []proto.Check
+	fatalOK := true
+
+	addFatal := func(name string, ok bool, detail string) {
+		checks = append(checks, proto.Check{Name: name, OK: ok, Detail: detail, Fatal: true})
+		if !ok {
+			fatalOK = false
+		}
+	}
+
+	exists, err := nft.Exists(ctx, r)
+	if err != nil {
+		addFatal("nft-table", false, err.Error())
+	} else {
+		addFatal("nft-table", exists, nft.Table())
+	}
+
+	for _, target := range []struct {
+		family   int
+		gateway  string
+		table    int
+		priority int
+	}{
+		{family: 4, gateway: wan.Gateway4, table: plan.TableV4, priority: plan.PriorityV4},
+		{family: 6, gateway: wan.Gateway6, table: plan.TableV6, priority: plan.PriorityV6},
+	} {
+		if target.gateway == "" {
+			continue
+		}
+		ruleOK, err := iprule.RuleExists(ctx, r, iprule.Spec{Family: target.family, Mark: plan.Mark, Table: target.table, Priority: target.priority})
+		addFatal(fmt.Sprintf("ipv%d-rule", target.family), ruleOK && err == nil, errDetail(err, "fwmark rule present"))
+		routeOK, err := iprule.RouteExists(ctx, r, target.family, target.table, target.gateway, wan.Interface)
+		addFatal(fmt.Sprintf("ipv%d-route", target.family), routeOK && err == nil, errDetail(err, "table "+strconv.Itoa(target.table)))
+	}
+
+	if plan.ManageEngine && !plan.SkipEngine && plan.ServiceName != "" {
+		active, err := engine.IsActive(ctx, r, plan.ServiceName)
+		addFatal("engine-active", active && err == nil, errDetail(err, plan.ServiceName))
+		tuns, tunErr := netdetect.TunInterfaces(ctx, r)
+		detail := strings.Join(tuns, ", ")
+		if tunErr != nil {
+			detail = errDetail(tunErr, detail)
+		}
+		checks = append(checks, proto.Check{Name: "tun", OK: tunErr == nil && len(tuns) > 0, Warn: true, Detail: detail})
+	}
+
+	if plan.HealthProbe != "" {
+		if err := health.DialTCP(ctx, plan.HealthProbe, healthTimeout); err != nil {
+			checks = append(checks, proto.Check{
+				Name:   "connectivity",
+				OK:     false,
+				Warn:   true,
+				Detail: fmt.Sprintf("%v (set health.probe, or \"\" to disable)", err),
+			})
+		} else {
+			checks = append(checks, proto.Check{Name: "connectivity", OK: true, Detail: "tcp " + plan.HealthProbe})
+		}
+	}
+
+	return checks, fatalOK
+}
+
+func errDetail(err error, ok string) string {
+	if err != nil {
+		return err.Error()
+	}
+	return ok
+}
+
+// packageHints names the package that provides each required binary, so the
+// preflight failure is actionable on any distribution.
+var packageHints = map[string]string{
+	"nft":        "install nftables",
+	"ip":         "install iproute2",
+	"systemctl":  "install systemd",
+	"journalctl": "install systemd",
+}
+
+// preflight verifies the external commands required for an operation exist.
+func preflight(r execx.Runner, needSystemctl bool) error {
+	bins := []string{"nft", "ip"}
+	if needSystemctl {
+		bins = append(bins, "systemctl")
+	}
+	for _, bin := range bins {
+		if _, err := r.LookPath(bin); err != nil {
+			return fmt.Errorf("required binary %q not found in PATH; %s to continue", bin, packageHints[bin])
+		}
+	}
+	return nil
+}
+
+// validatePlan re-validates every field of an untrusted plan before use.
+func validatePlan(plan proto.Plan) error {
+	if plan.Mark == 0 {
+		return errors.New("mark must be non-zero")
+	}
+	if plan.WANInterface != "" && !config.ValidInterfaceName(plan.WANInterface) {
+		return fmt.Errorf("invalid wan_interface %q", plan.WANInterface)
+	}
+	if plan.WANGateway4 != "" && net.ParseIP(plan.WANGateway4) == nil {
+		return fmt.Errorf("invalid wan_gateway_v4 %q", plan.WANGateway4)
+	}
+	if plan.WANGateway6 != "" && net.ParseIP(plan.WANGateway6) == nil {
+		return fmt.Errorf("invalid wan_gateway_v6 %q", plan.WANGateway6)
+	}
+	for _, check := range []struct {
+		field string
+		value int
+	}{
+		{"table_v4", plan.TableV4},
+		{"table_v6", plan.TableV6},
+		{"bypass_table_v4", plan.BypassTableV4},
+		{"bypass_table_v6", plan.BypassTableV6},
+	} {
+		if err := config.CheckTableID(check.field, check.value); err != nil {
+			return err
+		}
+	}
+	for _, check := range []struct {
+		field string
+		value int
+	}{
+		{"priority_v4", plan.PriorityV4},
+		{"priority_v6", plan.PriorityV6},
+		{"bypass_priority_v4", plan.BypassPrioV4},
+		{"bypass_priority_v6", plan.BypassPrioV6},
+	} {
+		if err := config.CheckPriority(check.field, check.value); err != nil {
+			return err
+		}
+	}
+	if plan.ManageEngine && plan.ServiceName != "" && !config.ValidServiceName(plan.ServiceName) {
+		return fmt.Errorf("invalid service name %q", plan.ServiceName)
+	}
+	if plan.EngineConfig != "" && !filepath.IsAbs(plan.EngineConfig) {
+		return fmt.Errorf("engine_config %q must be absolute", plan.EngineConfig)
+	}
+	if plan.StatePath == "" || !filepath.IsAbs(plan.StatePath) {
+		return fmt.Errorf("state_path %q must be an absolute path", plan.StatePath)
+	}
+	for _, cidr := range plan.BypassCIDRs {
+		if _, _, err := net.ParseCIDR(cidr); err != nil {
+			return fmt.Errorf("invalid bypass CIDR %q", cidr)
+		}
+	}
+	return nil
+}

internal/helper/helper_test.go

@@ -0,0 +1,298 @@
+// SPDX-License-Identifier: MIT
+
+package helper
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"prx/internal/execx"
+	"prx/internal/proto"
+)
+
+func basePlan(t *testing.T) proto.Plan {
+	t.Helper()
+	return proto.Plan{
+		Action:        proto.ActionApply,
+		Mark:          0x100,
+		TableV4:       100,
+		TableV6:       101,
+		PriorityV4:    100,
+		PriorityV6:    100,
+		BypassTableV4: 102,
+		BypassTableV6: 103,
+		BypassPrioV4:  50,
+		BypassPrioV6:  50,
+		ManageEngine:  false,
+		StatePath:     filepath.Join(t.TempDir(), "state.json"),
+	}
+}
+
+// systemFake emulates a host with a working WAN and optionally active engine.
+func systemFake(active bool) *execx.Fake {
+	return &execx.Fake{Handler: func(name string, args []string, _ []byte) (execx.Result, error) {
+		joined := strings.TrimSpace(name + " " + strings.Join(args, " "))
+		switch {
+		case joined == "ip -j route show default":
+			return execx.Result{Stdout: []byte(`[{"dst":"default","gateway":"192.168.1.1","dev":"wlan0","metric":600}]`)}, nil
+		case joined == "ip -6 -j route show default":
+			return execx.Result{Stdout: []byte(`[]`)}, nil
+		case joined == "ip -j addr show dev wlan0":
+			return execx.Result{Stdout: []byte(`[{"ifname":"wlan0","addr_info":[{"family":"inet","local":"192.168.1.100","prefixlen":24,"scope":"global"}]}]`)}, nil
+		case joined == "ip rule show" || joined == "ip -6 rule show":
+			return execx.Result{Stdout: []byte("100:\tfrom all fwmark 0x100 lookup 100\n")}, nil
+		case strings.HasPrefix(joined, "ip route show table"):
+			return execx.Result{Stdout: []byte("default via 192.168.1.1 dev wlan0\n")}, nil
+		case joined == "ip -j link show type tun":
+			return execx.Result{Stdout: []byte(`[{"ifname":"tun0"}]`)}, nil
+		case strings.HasPrefix(joined, "nft"):
+			return execx.Result{}, nil
+		case strings.HasPrefix(joined, "systemctl is-active"):
+			if active {
+				return execx.Result{Stdout: []byte("active\n")}, nil
+			}
+			return execx.Result{Stdout: []byte("inactive\n"), Code: 3}, &execx.ExitError{Name: "systemctl", Code: 3, Stderr: []byte("inactive")}
+		case strings.HasPrefix(joined, "systemctl start"):
+			return execx.Result{}, nil
+		case strings.HasPrefix(joined, "systemctl stop"):
+			return execx.Result{}, nil
+		}
+		return execx.Result{}, nil
+	}}
+}
+
+func TestValidatePlan(t *testing.T) {
+	cases := map[string]func(*proto.Plan){
+		"zero mark":      func(p *proto.Plan) { p.Mark = 0 },
+		"reserved table": func(p *proto.Plan) { p.TableV4 = 254 },
+		"bad priority":   func(p *proto.Plan) { p.PriorityV4 = 99999 },
+		"bad cidr":       func(p *proto.Plan) { p.BypassCIDRs = []string{"nope"} },
+		"relative state": func(p *proto.Plan) { p.StatePath = "state.json" },
+		"bad iface":      func(p *proto.Plan) { p.WANInterface = "bad iface!" },
+		"bad gateway":    func(p *proto.Plan) { p.WANGateway4 = "not-ip" },
+	}
+	for name, mutate := range cases {
+		p := basePlan(t)
+		mutate(&p)
+		if err := validatePlan(p); err == nil {
+			t.Errorf("case %q: expected error", name)
+		}
+	}
+	if err := validatePlan(basePlan(t)); err != nil {
+		t.Fatalf("valid plan rejected: %v", err)
+	}
+}
+
+func TestApplyDryRunChangesNothing(t *testing.T) {
+	p := basePlan(t)
+	p.DryRun = true
+	res := Execute(context.Background(), systemFake(false), p)
+	if !res.OK {
+		t.Fatalf("dry run failed: %+v", res)
+	}
+	if res.Changed {
+		t.Fatal("dry run must not report changes")
+	}
+}
+
+func TestApplySkipEngineSucceeds(t *testing.T) {
+	p := basePlan(t)
+	p.SkipEngine = true
+	f := systemFake(false)
+	res := Execute(context.Background(), f, p)
+	if !res.OK {
+		t.Fatalf("apply failed: %+v", res)
+	}
+	if _, err := filepath.Glob(p.StatePath); err != nil {
+		t.Fatal(err)
+	}
+	if !hasCall(f.Calls, "nft", "-f", "-") {
+		t.Fatal("expected nft apply call")
+	}
+}
+
+func TestApplyRollsBackOnHealthFailure(t *testing.T) {
+	p := basePlan(t)
+	p.ManageEngine = true
+	p.ServiceName = "sing-box"
+	p.SkipEngine = false
+	f := systemFake(false) // engine never becomes active
+	res := Execute(context.Background(), f, p)
+	if res.OK {
+		t.Fatal("expected apply to fail")
+	}
+	if !strings.Contains(res.Error, "rolled back") {
+		t.Fatalf("expected rollback message, got %q", res.Error)
+	}
+	if !hasCall(f.Calls, "nft", "destroy", "table", "inet", "prx") {
+		t.Fatal("expected rollback to destroy the nft table")
+	}
+}
+
+func TestRemoveIsIdempotent(t *testing.T) {
+	p := basePlan(t)
+	p.Action = proto.ActionRemove
+	p.StatePath = filepath.Join(t.TempDir(), "state.json")
+	f := &execx.Fake{Handler: func(name string, args []string, _ []byte) (execx.Result, error) {
+		if strings.HasPrefix(strings.TrimSpace(name+" "+strings.Join(args, " ")), "nft") {
+			return execx.Result{}, nil
+		}
+		return execx.Result{Stdout: []byte("")}, nil
+	}}
+	for i := 0; i < 2; i++ {
+		res := Execute(context.Background(), f, p)
+		if !res.OK {
+			t.Fatalf("remove iteration %d failed: %+v", i, res)
+		}
+	}
+}
+
+func TestInspectReturnsChecks(t *testing.T) {
+	p := basePlan(t)
+	p.Action = proto.ActionInspect
+	res := Execute(context.Background(), systemFake(true), p)
+	if !res.OK {
+		t.Fatalf("inspect failed: %+v", res)
+	}
+	if len(res.Checks) == 0 {
+		t.Fatal("expected checks")
+	}
+}
+
+func TestInitEngineRejectsBadJSON(t *testing.T) {
+	p := proto.Plan{
+		Action:           proto.ActionInitEngine,
+		EngineConfig:     filepath.Join(t.TempDir(), "config.json"),
+		EngineConfigBody: []byte("not json"),
+	}
+	res := Execute(context.Background(), systemFake(false), p)
+	if res.OK {
+		t.Fatal("expected rejection of invalid JSON")
+	}
+}
+
+func hasCall(calls []execx.Call, name string, args ...string) bool {
+	for _, c := range calls {
+		if c.Name != name || len(c.Args) != len(args) {
+			continue
+		}
+		match := true
+		for i := range args {
+			if c.Args[i] != args[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			return true
+		}
+	}
+	return false
+}
+
+// TestApplyRollsBackPartialRoutingFailure verifies that a failure after the
+// policy rule was added still cleans up the routing table. The rollback for a
+// table must be registered before the table is touched.
+func TestApplyRollsBackPartialRoutingFailure(t *testing.T) {
+	p := basePlan(t)
+	p.SkipEngine = true
+
+	f := &execx.Fake{Handler: func(name string, args []string, _ []byte) (execx.Result, error) {
+		joined := strings.TrimSpace(name + " " + strings.Join(args, " "))
+		switch {
+		case joined == "ip -j route show default":
+			return execx.Result{Stdout: []byte(`[{"dst":"default","gateway":"192.168.1.1","dev":"wlan0","metric":600}]`)}, nil
+		case joined == "ip -6 -j route show default":
+			return execx.Result{Stdout: []byte(`[]`)}, nil
+		case joined == "ip -j addr show dev wlan0":
+			return execx.Result{Stdout: []byte(`[{"ifname":"wlan0","addr_info":[{"family":"inet","local":"192.168.1.100","prefixlen":24,"scope":"global"}]}]`)}, nil
+		case joined == "ip rule show":
+			// No rule present yet, so EnsureRule will add one.
+			return execx.Result{Stdout: []byte("")}, nil
+		case joined == "ip route show table 100":
+			// RouteExists cannot read the table: force EnsureRoute to fail.
+			return execx.Result{Code: 2, Stderr: []byte("RTNETLINK answers: Operation not permitted")}, &execx.ExitError{Name: "ip", Code: 2, Stderr: []byte("RTNETLINK answers: Operation not permitted")}
+		case strings.HasPrefix(joined, "nft"):
+			return execx.Result{}, nil
+		}
+		return execx.Result{}, nil
+	}}
+
+	res := Execute(context.Background(), f, p)
+	if res.OK {
+		t.Fatalf("expected apply to fail: %+v", res)
+	}
+	if !hasCall(f.Calls, "ip", "rule", "add", "fwmark", "0x100", "lookup", "100", "priority", "100") {
+		t.Fatal("expected the policy rule to have been added")
+	}
+	if !hasCall(f.Calls, "ip", "route", "flush", "table", "100") {
+		t.Fatal("partial failure leaked routing table 100 (rollback not registered early enough)")
+	}
+	if !hasCall(f.Calls, "nft", "destroy", "table", "inet", "prx") {
+		t.Fatal("expected rollback to destroy the nft table")
+	}
+	if _, err := os.Stat(p.StatePath); !os.IsNotExist(err) {
+		t.Fatalf("state file should not exist after rollback, stat err = %v", err)
+	}
+}
+
+func TestApplyAcquiresStateLock(t *testing.T) {
+	p := basePlan(t)
+	p.SkipEngine = true
+	if res := Execute(context.Background(), systemFake(false), p); !res.OK {
+		t.Fatalf("apply failed: %+v", res)
+	}
+	lock := filepath.Join(filepath.Dir(p.StatePath), "prx.lock")
+	if _, err := os.Stat(lock); err != nil {
+		t.Fatalf("expected a lock file next to the state: %v", err)
+	}
+}
+
+// TestApplyReportsIncompleteRollback verifies that a rollback step which fails
+// is surfaced to the caller instead of being silently discarded.
+func TestApplyReportsIncompleteRollback(t *testing.T) {
+	p := basePlan(t)
+	p.SkipEngine = true
+
+	f := &execx.Fake{Handler: func(name string, args []string, _ []byte) (execx.Result, error) {
+		joined := strings.TrimSpace(name + " " + strings.Join(args, " "))
+		switch {
+		case joined == "ip -j route show default":
+			return execx.Result{Stdout: []byte(`[{"dst":"default","gateway":"192.168.1.1","dev":"wlan0","metric":600}]`)}, nil
+		case joined == "ip -6 -j route show default":
+			return execx.Result{Stdout: []byte(`[]`)}, nil
+		case joined == "ip -j addr show dev wlan0":
+			return execx.Result{Stdout: []byte(`[{"ifname":"wlan0","addr_info":[{"family":"inet","local":"192.168.1.100","prefixlen":24,"scope":"global"}]}]`)}, nil
+		case joined == "ip rule show":
+			return execx.Result{Stdout: []byte("")}, nil
+		case joined == "ip route show table 100":
+			return execx.Result{Code: 2, Stderr: []byte("RTNETLINK answers: Operation not permitted")}, &execx.ExitError{Name: "ip", Code: 2, Stderr: []byte("RTNETLINK answers: Operation not permitted")}
+		case joined == "ip route flush table 100":
+			// The rollback itself fails.
+			return execx.Result{Code: 2, Stderr: []byte("flush refused")}, &execx.ExitError{Name: "ip", Code: 2, Stderr: []byte("flush refused")}
+		case strings.HasPrefix(joined, "nft"):
+			return execx.Result{}, nil
+		}
+		return execx.Result{}, nil
+	}}
+
+	res := Execute(context.Background(), f, p)
+	if res.OK {
+		t.Fatalf("expected apply to fail: %+v", res)
+	}
+	if !strings.Contains(res.Error, "rollback incomplete") {
+		t.Fatalf("res.Error = %q, want it to flag an incomplete rollback", res.Error)
+	}
+	found := false
+	for _, m := range res.Messages {
+		if strings.Contains(m, "rollback warning") {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("expected a rollback warning message, got %v", res.Messages)
+	}
+}

internal/helper/lock_other.go

@@ -0,0 +1,11 @@
+// SPDX-License-Identifier: MIT
+
+//go:build !unix
+
+package helper
+
+// lockState is a no-op on platforms without advisory file locking. prx only
+// supports Unix-like systems; this exists so the package still builds elsewhere.
+func lockState(_ string) (func() error, error) {
+	return func() error { return nil }, nil
+}

internal/helper/lock_unix.go

@@ -0,0 +1,43 @@
+// SPDX-License-Identifier: MIT
+
+//go:build unix
+
+package helper
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"syscall"
+)
+
+// lockState serialises concurrent apply/remove operations using an advisory
+// file lock stored next to the state file. It returns a release function that
+// must always be called (typically deferred); its error reports a release that
+// failed, which the caller should surface.
+func lockState(statePath string) (func() error, error) {
+	dir := filepath.Dir(statePath)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, fmt.Errorf("create %s: %w", dir, err)
+	}
+	path := filepath.Join(dir, "prx.lock")
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
+	if err != nil {
+		return nil, fmt.Errorf("open lock %s: %w", path, err)
+	}
+	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
+		_ = f.Close()
+		return nil, fmt.Errorf("lock %s: %w", path, err)
+	}
+	return func() error {
+		var errs []error
+		if err := syscall.Flock(int(f.Fd()), syscall.LOCK_UN); err != nil {
+			errs = append(errs, fmt.Errorf("unlock %s: %w", path, err))
+		}
+		if err := f.Close(); err != nil {
+			errs = append(errs, fmt.Errorf("close %s: %w", path, err))
+		}
+		return errors.Join(errs...)
+	}, nil
+}