prx/internal/helper/helper.go

// 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
}