diff --git a/DESIGN.md b/DESIGN.md index 3253925..cf30d34 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -430,6 +430,10 @@ report the generation applied, giving a fleet-wide "converged / N behind" view. source/dest disables its rule loudly, never opens it. - **Adds fail closed, the control plane fails open.** Partial rollout blocks new flows until every hop converges; a dead API leaves the last-good posture running. +- **A generation that severs the API is reverted.** After applying, the agent + reports `applied` over a fresh connection; if that fails at the transport level + it restores the snapshot, reports `reverted`, and skips that generation (persisted + in `/var/lib/tomswall/reverted.json`) until a newer one arrives. --- diff --git a/cmd/tomswall/agent.go b/cmd/tomswall/agent.go index 6f91379..3cdc2b6 100644 --- a/cmd/tomswall/agent.go +++ b/cmd/tomswall/agent.go @@ -29,7 +29,8 @@ func agentCmd() *cobra.Command { Long: `Agent runs the control-plane pull loop: it fetches this device's compiled config from tomswallapi, differentially applies it, and reports the applied generation back. It caches the last known-good config and, if the control plane -is unreachable, keeps applying that cache — it never fails closed. +is unreachable, keeps applying that cache — it never fails closed. A new +generation that cuts the agent off from the API is reverted and reported as such. The agent token defaults to the TOMSWALL_AGENT_TOKEN environment variable, and the device name defaults to the system hostname.`, diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 868c4e0..89c0870 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -2,8 +2,13 @@ package agent import ( "context" + "encoding/json" + "errors" "fmt" "log/slog" + "net/url" + "os" + "path/filepath" "time" "git.unkin.net/unkin/tomswall/internal/config" @@ -12,9 +17,10 @@ import ( ) // Applier applies a translated config to the firewall. Abstracted so the run -// loop is testable without touching the kernel. +// loop is testable without touching the kernel. restore returns the firewall to +// its state before the call; it is nil when nothing changed. type Applier interface { - Apply(ctx context.Context, cfg *config.Config) error + Apply(ctx context.Context, cfg *config.Config) (restore func() error, err error) } // Agent runs the pull-apply-report loop for one device. @@ -62,16 +68,26 @@ func (a *Agent) RunOnce(ctx context.Context) error { return fmt.Errorf("control plane unreachable and no cached config: %w", err) } // Re-apply last known-good; do not report a generation we didn't fetch. - return a.applyConfig(ctx, cached, false) + return a.applyConfig(ctx, cached, nil) } - if err := a.Cache.Write(raw); err != nil { - slog.Warn("agent: caching config failed", "err", err) + rv, err := a.readReverted() + if err != nil { + return err } - return a.applyConfig(ctx, rc, true) + if rv != nil { + a.reportReverted(ctx, rv) + if rc.Generation <= rv.Generation { + slog.Info("agent: generation was reverted, waiting for a newer one", "generation", rc.Generation) + return nil + } + } + return a.applyConfig(ctx, rc, raw) } -func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, report bool) error { +// applyConfig applies rc. A fetched config (raw != nil) is verified by reaching +// the API through the new ruleset and reverted if that fails; only then is it cached. +func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, raw []byte) error { resolver := a.Resolver if resolver == nil { resolver = NewResolver(rc.Resolver) @@ -82,46 +98,177 @@ func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, report bool if err != nil { return fmt.Errorf("translate: %w", err) } - if err := a.Applier.Apply(ctx, cfg); err != nil { + + unlock, err := tryapply.Acquire() + if errors.Is(err, tryapply.ErrPending) { + slog.Warn("agent: a 'tomswall try' is pending, skipping cycle") + return nil + } + if err != nil { + return err + } + defer unlock() + + restore, err := a.Applier.Apply(ctx, cfg) + if err != nil { + if restore != nil { + if rerr := restore(); rerr != nil { + err = fmt.Errorf("%w; restoring previous ruleset: %v", err, rerr) + } + } + if raw != nil { + if rerr := a.Client.ReportStatus(ctx, Status{Status: StatusFailed, Generation: rc.Generation, Error: err.Error()}); rerr != nil { + slog.Warn("agent: reporting status failed", "err", rerr) + } + } return fmt.Errorf("apply: %w", err) } + if raw == nil { + slog.Info("agent: applied cached config", "generation", rc.Generation, "rules", len(cfg.Rules)) + return nil + } + + if err := a.confirm(ctx, rc.Generation); err != nil { + if ctx.Err() != nil || restore == nil { + return err + } + if rerr := restore(); rerr != nil { + return fmt.Errorf("%w; restoring previous ruleset: %v", err, rerr) + } + rv := &reverted{Generation: rc.Generation, Error: err.Error()} + if werr := a.writeReverted(rv); werr != nil { + slog.Error("agent: persisting reverted generation failed", "err", werr) + } + a.reportReverted(ctx, rv) + return fmt.Errorf("generation %d reverted: %w", rc.Generation, err) + } slog.Info("agent: applied config", "generation", rc.Generation, "rules", len(cfg.Rules)) - if report { - if err := a.Client.ReportStatus(ctx, rc.Generation); err != nil { - slog.Warn("agent: reporting status failed", "err", err) - } - // Report the FIB so the control plane can scope router enforcement. - if fib := CollectFIB(ctx); len(fib) > 0 { - if err := a.Client.ReportRoutes(ctx, fib); err != nil { - slog.Warn("agent: reporting routes failed", "err", err) - } + if err := a.Cache.Write(raw); err != nil { + slog.Warn("agent: caching config failed", "err", err) + } + if err := os.Remove(a.revertedPath()); err != nil && !os.IsNotExist(err) { + slog.Warn("agent: clearing reverted generation failed", "err", err) + } + // Report the FIB so the control plane can scope router enforcement. + if fib := CollectFIB(ctx); len(fib) > 0 { + if err := a.Client.ReportRoutes(ctx, fib); err != nil { + slog.Warn("agent: reporting routes failed", "err", err) } } return nil } -// EngineApplier applies via the real nftables differential engine. -type EngineApplier struct{} +var ( + verifyAttempts = 3 + verifyDelay = 2 * time.Second + verifyTimeout = 5 * time.Second +) -// Apply computes and applies the differential change set for cfg. It refuses -// while a 'tomswall try' awaits confirmation. -func (EngineApplier) Apply(_ context.Context, cfg *config.Config) error { - unlock, err := tryapply.Acquire() +// errUnreachable means the API could not be reached through the new ruleset. +var errUnreachable = errors.New("control plane unreachable after apply") + +// confirm reports generation as applied over a fresh connection, which proves +// the API is reachable through the new ruleset. Any HTTP response counts as +// reachable; only repeated transport failures return errUnreachable. +func (a *Agent) confirm(ctx context.Context, generation int64) error { + var err error + for i := 0; i < verifyAttempts; i++ { + if i > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(verifyDelay): + } + } + actx, cancel := context.WithTimeout(ctx, verifyTimeout) + err = a.Client.ReportStatus(actx, Status{Status: StatusApplied, Generation: generation}) + cancel() + if ctx.Err() != nil { + return ctx.Err() + } + var uerr *url.Error + if !errors.As(err, &uerr) { + if err != nil { + slog.Warn("agent: reporting status failed", "err", err) + } + return nil + } + } + return fmt.Errorf("%w: %v", errUnreachable, err) +} + +// reverted is a generation rolled back for severing the API; persisted so it +// is not re-applied after a restart until a newer generation arrives. +type reverted struct { + Generation int64 `json:"generation"` + Error string `json:"error,omitempty"` + Reported bool `json:"reported"` +} + +func (a *Agent) revertedPath() string { + return filepath.Join(filepath.Dir(a.Cache.Path), "reverted.json") +} + +func (a *Agent) readReverted() (*reverted, error) { + b, err := os.ReadFile(a.revertedPath()) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + var rv reverted + if err := json.Unmarshal(b, &rv); err != nil { + return nil, fmt.Errorf("parsing %s: %w", a.revertedPath(), err) + } + return &rv, nil +} + +func (a *Agent) writeReverted(rv *reverted) error { + b, err := json.Marshal(rv) if err != nil { return err } - defer unlock() + return Cache{Path: a.revertedPath()}.Write(b) +} + +// reportReverted reports rv until the control plane accepts it. +func (a *Agent) reportReverted(ctx context.Context, rv *reverted) { + if rv.Reported { + return + } + if err := a.Client.ReportStatus(ctx, Status{Status: StatusReverted, Generation: rv.Generation, Error: rv.Error}); err != nil { + slog.Warn("agent: reporting reverted generation failed, retrying next cycle", "generation", rv.Generation, "err", err) + return + } + rv.Reported = true + if err := a.writeReverted(rv); err != nil { + slog.Warn("agent: persisting reverted generation failed", "err", err) + } +} + +// EngineApplier applies via the real nftables differential engine. +type EngineApplier struct{} + +// Apply computes and applies the differential change set for cfg, snapshotting +// the live table first so the caller can restore it. The caller holds the try lock. +func (EngineApplier) Apply(_ context.Context, cfg *config.Config) (func() error, error) { engine, err := nftables.NewEngine(cfg) if err != nil { - return fmt.Errorf("initializing nftables: %w", err) + return nil, fmt.Errorf("initializing nftables: %w", err) } changes, err := engine.Plan() if err != nil { - return fmt.Errorf("computing changes: %w", err) + return nil, fmt.Errorf("computing changes: %w", err) } if changes.Empty() { - return nil + return nil, nil } - return engine.Apply(changes) + snap, err := engine.Snapshot() + if err != nil { + return nil, fmt.Errorf("snapshotting ruleset: %w", err) + } + restore := func() error { return engine.Restore(snap) } + return restore, engine.Apply(changes) } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 3019815..1cafc57 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -132,10 +132,10 @@ type fakeApplier struct { lastGen int } -func (f *fakeApplier) Apply(_ context.Context, cfg *config.Config) error { +func (f *fakeApplier) Apply(_ context.Context, cfg *config.Config) (func() error, error) { atomic.AddInt32(&f.count, 1) f.lastGen = len(cfg.Rules) - return nil + return nil, nil } const renderedYAML = `generation: 7 diff --git a/internal/agent/client.go b/internal/agent/client.go index d385011..0bec573 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -26,7 +26,9 @@ func NewClient(baseURL, device, token string) *Client { BaseURL: baseURL, Device: device, Token: token, - HTTP: &http.Client{Timeout: 30 * time.Second}, + // No keep-alives: every request, the post-apply check included, opens a + // fresh connection that must pass the current ruleset. + HTTP: &http.Client{Timeout: 30 * time.Second, Transport: noKeepAlive()}, } } @@ -94,10 +96,24 @@ func (c *Client) ReportRoutes(ctx context.Context, prefixes []string) error { return nil } -// ReportStatus tells the control plane which generation this device has applied. -func (c *Client) ReportStatus(ctx context.Context, generation int64) error { +// Status values reported to POST /api/v1/devices/{name}/status. +const ( + StatusApplied = "applied" + StatusReverted = "reverted" + StatusFailed = "failed" +) + +// Status is the outcome of applying one generation. +type Status struct { + Status string `json:"status"` + Generation int64 `json:"generation"` + Error string `json:"error,omitempty"` +} + +// ReportStatus tells the control plane the outcome of applying a generation. +func (c *Client) ReportStatus(ctx context.Context, st Status) error { url := fmt.Sprintf("%s/api/v1/devices/%s/status", c.BaseURL, c.Device) - payload, _ := json.Marshal(map[string]int64{"generation": generation}) + payload, _ := json.Marshal(st) req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload)) if err != nil { return err @@ -116,3 +132,9 @@ func (c *Client) ReportStatus(ctx context.Context, generation int64) error { } return nil } + +func noKeepAlive() http.RoundTripper { + t := http.DefaultTransport.(*http.Transport).Clone() + t.DisableKeepAlives = true + return t +} diff --git a/internal/agent/safeapply_test.go b/internal/agent/safeapply_test.go new file mode 100644 index 0000000..bb38054 --- /dev/null +++ b/internal/agent/safeapply_test.go @@ -0,0 +1,249 @@ +package agent + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.unkin.net/unkin/tomswall/internal/config" + "git.unkin.net/unkin/tomswall/internal/tryapply" +) + +func TestMain(m *testing.M) { + dir, err := os.MkdirTemp("", "tomswall-agent-test") + if err != nil { + panic(err) + } + tryapply.Dir = dir + verifyDelay = time.Millisecond + verifyTimeout = time.Second + code := m.Run() + os.RemoveAll(dir) + os.Exit(code) +} + +// fakeAPI serves a config generation and records status reports; while cut it +// drops connections to the status endpoint, as a severing ruleset would. +type fakeAPI struct { + *httptest.Server + gen atomic.Int64 + cut atomic.Bool + code atomic.Int32 + mu sync.Mutex + reports []Status +} + +func newFakeAPI(t *testing.T, gen int64) *fakeAPI { + f := &fakeAPI{} + f.gen.Store(gen) + f.code.Store(http.StatusNoContent) + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/devices/fw-a/config": + _, _ = w.Write([]byte(strings.Replace(renderedYAML, "generation: 7", "generation: "+itoa(f.gen.Load()), 1))) + case "/api/v1/devices/fw-a/status": + if f.cut.Load() { + conn, _, _ := w.(http.Hijacker).Hijack() + conn.Close() + return + } + var st Status + _ = json.NewDecoder(r.Body).Decode(&st) + f.mu.Lock() + f.reports = append(f.reports, st) + f.mu.Unlock() + w.WriteHeader(int(f.code.Load())) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(f.Close) + return f +} + +func itoa(n int64) string { b, _ := json.Marshal(n); return string(b) } + +func (f *fakeAPI) last() Status { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.reports) == 0 { + return Status{} + } + return f.reports[len(f.reports)-1] +} + +// fakeEngine always changes the ruleset; onApply simulates its effect. +type fakeEngine struct { + applies, restores int + err error + onApply func() + onRestore func() +} + +func (f *fakeEngine) Apply(context.Context, *config.Config) (func() error, error) { + f.applies++ + if f.onApply != nil { + f.onApply() + } + return func() error { + f.restores++ + if f.onRestore != nil { + f.onRestore() + } + return nil + }, f.err +} + +func newAgent(t *testing.T, api *fakeAPI, eng *fakeEngine) *Agent { + return &Agent{ + Client: NewClient(api.URL, "fw-a", "tok"), + Cache: Cache{Path: filepath.Join(t.TempDir(), "rendered.yaml")}, + Applier: eng, + } +} + +func cachedGen(t *testing.T, a *Agent) int64 { + rc, err := a.Cache.Read() + if err != nil { + t.Fatal(err) + } + if rc == nil { + return 0 + } + return rc.Generation +} + +func TestSafeApplyReachableApplies(t *testing.T) { + api := newFakeAPI(t, 7) + eng := &fakeEngine{} + a := newAgent(t, api, eng) + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if eng.restores != 0 || api.last() != (Status{Status: StatusApplied, Generation: 7}) || cachedGen(t, a) != 7 { + t.Fatalf("restores=%d last=%+v cache=%d", eng.restores, api.last(), cachedGen(t, a)) + } +} + +func TestSafeApplyUnreachableRevertsAndReports(t *testing.T) { + api := newFakeAPI(t, 7) + eng := &fakeEngine{onApply: func() { api.cut.Store(true) }, onRestore: func() { api.cut.Store(false) }} + a := newAgent(t, api, eng) + if err := a.RunOnce(context.Background()); !errors.Is(err, errUnreachable) { + t.Fatalf("want errUnreachable, got %v", err) + } + if eng.restores != 1 || cachedGen(t, a) != 0 { + t.Fatalf("restores=%d cache=%d", eng.restores, cachedGen(t, a)) + } + if st := api.last(); st.Status != StatusReverted || st.Generation != 7 || st.Error == "" { + t.Fatalf("last report %+v", st) + } + rv, _ := a.readReverted() + if rv == nil || rv.Generation != 7 || !rv.Reported { + t.Fatalf("persisted %+v", rv) + } +} + +func TestSafeApplyRevertReportedOnceReachable(t *testing.T) { + api := newFakeAPI(t, 7) + eng := &fakeEngine{onApply: func() { api.cut.Store(true) }} + a := newAgent(t, api, eng) + _ = a.RunOnce(context.Background()) + if eng.restores != 1 || api.last().Status != "" { + t.Fatalf("restores=%d last=%+v", eng.restores, api.last()) + } + api.cut.Store(false) + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if eng.applies != 1 || api.last() != (Status{Status: StatusReverted, Generation: 7, Error: api.last().Error}) { + t.Fatalf("applies=%d last=%+v", eng.applies, api.last()) + } +} + +func TestSafeApplyShutdownDoesNotRevert(t *testing.T) { + api := newFakeAPI(t, 7) + ctx, cancel := context.WithCancel(context.Background()) + eng := &fakeEngine{onApply: func() { api.cut.Store(true); cancel() }} + a := newAgent(t, api, eng) + if err := a.RunOnce(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("want context.Canceled, got %v", err) + } + if rv, _ := a.readReverted(); eng.restores != 0 || rv != nil || cachedGen(t, a) != 0 { + t.Fatalf("restores=%d reverted=%+v cache=%d", eng.restores, rv, cachedGen(t, a)) + } +} + +func TestSafeApplyHTTPErrorDoesNotRevert(t *testing.T) { + api := newFakeAPI(t, 7) + api.code.Store(http.StatusInternalServerError) + eng := &fakeEngine{} + a := newAgent(t, api, eng) + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if eng.restores != 0 || cachedGen(t, a) != 7 { + t.Fatalf("restores=%d cache=%d", eng.restores, cachedGen(t, a)) + } +} + +func TestSafeApplyApplyErrorRestoresAndReportsFailed(t *testing.T) { + api := newFakeAPI(t, 7) + eng := &fakeEngine{err: errors.New("netlink: boom")} + a := newAgent(t, api, eng) + if err := a.RunOnce(context.Background()); err == nil { + t.Fatal("want error") + } + if st := api.last(); eng.restores != 1 || st.Status != StatusFailed || !strings.Contains(st.Error, "boom") { + t.Fatalf("restores=%d last=%+v", eng.restores, st) + } +} + +func TestSafeApplyRevertedGenerationSkippedAfterRestart(t *testing.T) { + api := newFakeAPI(t, 7) + eng := &fakeEngine{} + a := newAgent(t, api, eng) + if err := a.writeReverted(&reverted{Generation: 7, Reported: true}); err != nil { + t.Fatal(err) + } + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if eng.applies != 0 { + t.Fatalf("reverted generation re-applied") + } + + api.gen.Store(8) + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if rv, _ := a.readReverted(); eng.applies != 1 || api.last().Generation != 8 || rv != nil { + t.Fatalf("applies=%d last=%+v reverted=%+v", eng.applies, api.last(), rv) + } +} + +func TestSafeApplySkipsWhileTryPending(t *testing.T) { + marker := filepath.Join(tryapply.Dir, "try-snapshot.json") + if err := os.WriteFile(marker, []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + defer os.Remove(marker) + api := newFakeAPI(t, 7) + eng := &fakeEngine{} + a := newAgent(t, api, eng) + if err := a.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + if eng.applies != 0 || api.last().Status != "" { + t.Fatalf("applies=%d last=%+v", eng.applies, api.last()) + } +}