diff --git a/internal/helper/helper.go b/internal/helper/helper.go new file mode 100644 index 0000000..20e39ce --- /dev/null +++ b/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 +} diff --git a/internal/helper/helper_test.go b/internal/helper/helper_test.go new file mode 100644 index 0000000..6b62579 --- /dev/null +++ b/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) + } +} diff --git a/internal/helper/lock_other.go b/internal/helper/lock_other.go new file mode 100644 index 0000000..b2faffe --- /dev/null +++ b/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 +} diff --git a/internal/helper/lock_unix.go b/internal/helper/lock_unix.go new file mode 100644 index 0000000..e4ad4e9 --- /dev/null +++ b/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 +}