From 61d068c95c38c195543e7707a69f3804a8c85134 Mon Sep 17 00:00:00 2001 From: AJ ONeal Date: Thu, 10 Sep 2026 02:09:17 -0600 Subject: [PATCH] refactor(ippolicy): separate policy loading from evaluation --- net/ippolicy/README.md | 13 +- net/ippolicy/policy.go | 284 ++++++++++++++++++------------ net/ippolicy/policy_start_test.go | 61 ++----- net/ippolicy/policy_test.go | 23 ++- 4 files changed, 209 insertions(+), 172 deletions(-) diff --git a/net/ippolicy/README.md b/net/ippolicy/README.md index 6a84a57..26c93c6 100644 --- a/net/ippolicy/README.md +++ b/net/ippolicy/README.md @@ -34,11 +34,12 @@ policy := ippolicy.New(ctx, ippolicy.Config{ }) defer policy.Stop() -if err := policy.Load(ctx, true); err != nil { - return err +evaluator, err := policy.Load(ctx, false) +if err != nil { + slog.Warn("ip policy refresh failed", "err", err) } -switch policy.Evaluate(addr) { +switch evaluator.Evaluate(addr) { case ippolicy.Blacklisted: // reject case ippolicy.Whitelisted, ippolicy.Unlisted: @@ -46,8 +47,10 @@ case ippolicy.Whitelisted, ippolicy.Unlisted: } ``` -Use `Load(ctx, false)` when a current snapshot may be used while refresh work -runs. Use `Start`/`Stop` only when periodic background checks are wanted. +`Policy` owns source loading and refresh coordination. `Evaluator` is the +immutable, read-only decision snapshot. Use `Load(ctx, false)` when the current +snapshot may be used while refresh work runs; it returns the last-good evaluator +when refresh fails. `Stop` shuts down source and DNS refresh work. Every source keeps its last-good snapshot after a failed update. A nil whitelist produces an `Unlisted` policy and intentionally ignores diff --git a/net/ippolicy/policy.go b/net/ippolicy/policy.go index 54a10a6..bf4ec51 100644 --- a/net/ippolicy/policy.go +++ b/net/ippolicy/policy.go @@ -1,5 +1,4 @@ -// Package ippolicy composes independently refreshed whitelist and blacklist -// sources into one IP decision policy. +// Package ippolicy manages IP policy sources and evaluates client addresses. package ippolicy import ( @@ -8,15 +7,19 @@ import ( "fmt" "net/netip" "sync" + "sync/atomic" "time" + "github.com/therootcompany/golib/net/ipcohort" "github.com/therootcompany/golib/net/iplist" + "github.com/therootcompany/golib/sync/cacheable" ) type Config struct { - Whitelist *iplist.IPList - Blacklist *IPPrefixSet - BlacklistExtra *iplist.IPList + Whitelist *iplist.IPList + Blacklist *IPPrefixSet + BlacklistExtra *iplist.IPList + RefreshInterval time.Duration } // Decision is the mutually exclusive result of evaluating an address. @@ -41,6 +44,29 @@ func (d Decision) String() string { } } +// Evaluator is an immutable policy snapshot. It is safe to use concurrently. +type Evaluator struct { + whitelist *ipcohort.Cohort + blacklist *ipcohort.Cohort + blacklistExtra *ipcohort.Cohort +} + +// Evaluate returns the decision for addr. A whitelist match always wins. +func (e *Evaluator) Evaluate(addr netip.Addr) Decision { + if e == nil { + return Unlisted + } + if e.whitelist != nil && e.whitelist.ContainsAddr(addr) { + return Whitelisted + } + if (e.blacklistExtra != nil && e.blacklistExtra.ContainsAddr(addr)) || + (e.blacklist != nil && e.blacklist.ContainsAddr(addr)) { + return Blacklisted + } + return Unlisted +} + +// Policy owns policy source loading and publishes immutable Evaluator snapshots. type Policy struct { whitelist *DomainSet blacklistExtra *DomainSet @@ -48,30 +74,33 @@ type Policy struct { whitelistSource *iplist.IPList blacklistExtraSource *iplist.IPList - start sync.Once - cancel context.CancelFunc - loadMu sync.Mutex + current atomic.Pointer[Evaluator] + loadedAt atomic.Pointer[time.Time] + lastErr atomic.Pointer[error] + refresh cacheable.Refresh + loadMu sync.Mutex + interval time.Duration } -// New creates a Policy from the given sources. The caller is responsible for -// constructing and later closing the *iplist.IPList values; New does not take -// ownership of them. -// -// A nil Whitelist is treated as an empty whitelist: every address evaluates to -// Unlisted. This also intentionally drops the blacklists, because a policy -// without a whitelist provides no meaningful allow-list semantics — blocking -// would reject everything not on the blacklist rather than only what is on it. -// The caller receives a RefreshFallback event so it can log or alert. +var ( + _ cacheable.Cacheable[Evaluator] = (*Policy)(nil) + _ cacheable.Mutable[Evaluator] = (*Policy)(nil) + _ cacheable.Inspectable = (*Policy)(nil) +) + +// New creates a Policy from the given sources. The caller owns the sources and +// must stop them separately or through Policy.Stop. func New(ctx context.Context, config Config) *Policy { - p := &Policy{} + p := &Policy{interval: config.RefreshInterval} + if p.interval <= 0 { + p.interval = DefaultDomainSetRefreshInterval + } if config.Whitelist == nil { - p.whitelist = EmptyDomainSet() return p } p.whitelistSource = config.Whitelist p.whitelist = NewDomainSetFromEntries(ctx, sourceEntries(config.Whitelist), DefaultDomainSetRefreshInterval) - if config.BlacklistExtra != nil { p.blacklistExtraSource = config.BlacklistExtra p.blacklistExtra = NewDomainSetFromEntries(ctx, sourceEntries(config.BlacklistExtra), DefaultDomainSetRefreshInterval) @@ -82,40 +111,147 @@ func New(ctx context.Context, config Config) *Policy { return p } -// Load refreshes policy inputs and updates the derived domain sets. It reads -// the current snapshot of each source IP list and pushes the entries into the -// derived DomainSets (re-resolving DNS for any hostname entries). With -// wait=false it returns after issuing the refreshes; with wait=true it waits -// for each source refresh to complete first. -// -// Sources that refresh in the background (via their own Start ticker) do not -// automatically update the derived sets; call Start on the Policy to run a -// periodic Load(false), or call Load explicitly after a source refresh. -func (p *Policy) Load(ctx context.Context, wait bool) error { +// Current returns the last published evaluator without blocking. +func (p *Policy) Current() *Evaluator { return p.current.Load() } + +// Load refreshes policy sources and returns the last-good evaluator. With +// wait=false, a refresh runs in the background when the snapshot is stale. +// Refresh errors are returned while the previous evaluator is retained. +func (p *Policy) Load(ctx context.Context, wait bool) (*Evaluator, error) { + if _, err := p.Revalidate(ctx); err != nil { + return p.Current(), err + } + if p.loadedAt.Load() == nil || wait { + if err := p.refresh.Wait(ctx); err != nil { + return p.Current(), err + } + } + return p.Current(), nil +} + +func (p *Policy) Due(ctx context.Context) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err + } + status := p.Status() + return !status.HasValue || status.StaleAt.IsZero() || !time.Now().Before(status.StaleAt), nil +} + +func (p *Policy) Revalidate(ctx context.Context) (bool, error) { + due, err := p.Due(ctx) + if err != nil || !due || !p.refresh.Begin() { + return false, err + } + go func() { p.refresh.Finish(p.update(ctx)) }() + return true, nil +} + +func (p *Policy) update(ctx context.Context) error { p.loadMu.Lock() defer p.loadMu.Unlock() if p.whitelistSource == nil { - return nil + return p.publish() } - if _, err := p.whitelistSource.Load(ctx, wait); err != nil { - return err + if _, err := p.whitelistSource.Load(ctx, true); err != nil { + return p.recordError(err) } p.whitelist.ReplaceEntries(sourceEntries(p.whitelistSource)) if p.blacklistExtraSource != nil { - if _, err := p.blacklistExtraSource.Load(ctx, wait); err != nil { - return err + if _, err := p.blacklistExtraSource.Load(ctx, true); err != nil { + return p.recordError(err) } p.blacklistExtra.ReplaceEntries(sourceEntries(p.blacklistExtraSource)) } if p.blacklist != nil { - if _, err := p.blacklist.Load(ctx, wait); err != nil { - return err + if _, err := p.blacklist.Load(ctx, true); err != nil { + return p.recordError(err) } } + return p.publish() +} + +func (p *Policy) publish() error { + e := &Evaluator{} + if p.whitelist != nil { + e.whitelist = p.whitelist.Current() + } + if p.blacklistExtra != nil { + e.blacklistExtra = p.blacklistExtra.Current() + } + if p.blacklist != nil { + e.blacklist = p.blacklist.Current() + } + p.current.Store(e) + now := time.Now() + p.loadedAt.Store(&now) + p.lastErr.Store(nil) return nil } +func (p *Policy) recordError(err error) error { + errCopy := err + p.lastErr.Store(&errCopy) + return err +} + +func (p *Policy) Set(e *Evaluator) error { + if e == nil { + return p.Clear() + } + p.current.Store(e) + now := time.Now() + p.loadedAt.Store(&now) + return nil +} + +func (p *Policy) Clear() error { + p.current.Store(nil) + p.loadedAt.Store(nil) + return nil +} + +func (p *Policy) Status() cacheable.Status { + loadedAt := time.Time{} + if t := p.loadedAt.Load(); t != nil { + loadedAt = *t + } + var lastErr error + if e := p.lastErr.Load(); e != nil { + lastErr = *e + } + status := cacheable.Status{LoadedAt: loadedAt, HasValue: p.Current() != nil, Refreshing: p.refresh.Running(), LastError: lastErr} + if !loadedAt.IsZero() { + status.StaleAt = loadedAt.Add(p.interval) + } + return status +} + +// Stop stops policy source and DNS refresh work. +func (p *Policy) Stop() error { + if p == nil { + return nil + } + var errs []error + if p.whitelist != nil { + errs = appendStopError(errs, p.whitelist.Stop()) + } + if p.blacklistExtra != nil { + errs = appendStopError(errs, p.blacklistExtra.Stop()) + } + if p.blacklist != nil { + errs = appendStopError(errs, p.blacklist.Stop()) + } + return errors.Join(errs...) +} + +func appendStopError(errs []error, err error) []error { + if err != nil { + return append(errs, err) + } + return errs +} + func sourceEntries(source *iplist.IPList) []string { entries := source.Current() if entries == nil { @@ -123,77 +259,3 @@ func sourceEntries(source *iplist.IPList) []string { } return *entries } - -func (p *Policy) Evaluate(addr netip.Addr) Decision { - if p.whitelist == nil { - return Unlisted - } - if p.whitelist.Contains(addr) { - return Whitelisted - } - if (p.blacklistExtra != nil && p.blacklistExtra.Contains(addr)) || - (p.blacklist != nil && p.blacklist.Contains(addr)) { - return Blacklisted - } - return Unlisted -} - -// Start starts at most one background loop that periodically re-reads the -// source IP lists and pushes their current entries into the derived domain -// sets (re-resolving DNS each time). This closes the gap where a source -// refreshes via its own ticker but the derived DomainSet is never updated -// because nobody calls Load. The interval defaults to -// DefaultDomainSetRefreshInterval when zero. -// -// Start does not start the DomainSets' own DNS refresh loops; the Policy loop -// drives both source re-reads and DNS re-resolution via Load+ReplaceEntries. -func (p *Policy) Start(ctx context.Context, interval time.Duration) { - p.start.Do(func() { - if interval <= 0 { - interval = DefaultDomainSetRefreshInterval - } - loopCtx, cancel := context.WithCancel(ctx) - p.cancel = cancel - go func() { - ticker := time.NewTicker(interval) - defer ticker.Stop() - for { - select { - case <-loopCtx.Done(): - return - case <-ticker.C: - // Re-read sources and re-resolve derived sets without blocking - // the hot path. Errors are non-fatal: last-good snapshots are - // retained by the sources and DomainSets. - _ = p.Load(loopCtx, false) - } - } - }() - }) -} - -func (p *Policy) Stop() error { - if p == nil { - return nil - } - if p.cancel != nil { - p.cancel() - } - var errs []error - if p.whitelist != nil { - if err := p.whitelist.Stop(); err != nil { - errs = append(errs, err) - } - } - if p.blacklistExtra != nil { - if err := p.blacklistExtra.Stop(); err != nil { - errs = append(errs, err) - } - } - if p.blacklist != nil { - if err := p.blacklist.Stop(); err != nil { - errs = append(errs, err) - } - } - return errors.Join(errs...) -} diff --git a/net/ippolicy/policy_start_test.go b/net/ippolicy/policy_start_test.go index 7fd21c8..f396647 100644 --- a/net/ippolicy/policy_start_test.go +++ b/net/ippolicy/policy_start_test.go @@ -10,73 +10,36 @@ import ( "github.com/therootcompany/golib/net/iplist" ) -// TestPolicyStartPushesSourceUpdatesIntoDerivedSets confirms that when a -// source IP list changes and the Policy's background loop runs, the derived -// DomainSet is updated without an explicit Policy.Load call. This is the gap -// flagged in the handoff: sources refresh on their own tickers but derived -// sets never saw the new entries. -func TestPolicyStartPushesSourceUpdatesIntoDerivedSets(t *testing.T) { +func TestPolicyLoadPublishesUpdatedSnapshot(t *testing.T) { path := filepath.Join(t.TempDir(), "allowed.tsv") if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { t.Fatal(err) } - // Short refresh interval so the source re-reads the file promptly. whitelist, err := iplist.NewIPList(t.Context(), iplist.IPListConfig{ - Source: path, - CacheDir: t.TempDir(), - RefreshInterval: 20 * time.Millisecond, + Source: path, CacheDir: t.TempDir(), RefreshInterval: time.Nanosecond, }) if err != nil { t.Fatal(err) } - - p := New(t.Context(), Config{Whitelist: whitelist}) + p := New(t.Context(), Config{Whitelist: whitelist, RefreshInterval: time.Nanosecond}) defer p.Stop() - if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { - t.Fatal("initial whitelist did not match 192.0.2.1") + evaluator, err := p.Load(t.Context(), true) + if err != nil { + t.Fatal(err) + } + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { + t.Fatal("initial whitelist did not match") } - - // Start the source ticker (re-reads file) and the Policy loop (pushes - // entries into the derived set). - whitelist.Start(t.Context(), 20*time.Millisecond) - p.Start(t.Context(), 20*time.Millisecond) - - // Change the source: drop 192.0.2.1, add 192.0.2.2. if err := os.WriteFile(path, []byte("192.0.2.2\n"), 0o600); err != nil { t.Fatal(err) } - // Within a few intervals the derived set must reflect the new entry. - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - if p.Evaluate(netip.MustParseAddr("192.0.2.2")) == Whitelisted { - break - } - time.Sleep(20 * time.Millisecond) - } - if p.Evaluate(netip.MustParseAddr("192.0.2.2")) != Whitelisted { - t.Fatal("derived set was not updated by Policy.Start loop (192.0.2.2 missing)") - } -} - -// TestPolicyStartIsIdempotent confirms Start can be called multiple times -// safely (only one loop starts). -func TestPolicyStartIsIdempotent(t *testing.T) { - path := filepath.Join(t.TempDir(), "allowed.tsv") - if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { - t.Fatal(err) - } - whitelist, err := iplist.NewIPList(t.Context(), iplist.IPListConfig{ - Source: path, - CacheDir: t.TempDir(), - }) + evaluator, err = p.Load(t.Context(), true) if err != nil { t.Fatal(err) } - p := New(t.Context(), Config{Whitelist: whitelist}) - defer p.Stop() - for range 3 { - p.Start(t.Context(), time.Hour) + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.2")) != Whitelisted { + t.Fatal("updated whitelist did not match") } } diff --git a/net/ippolicy/policy_test.go b/net/ippolicy/policy_test.go index 3de0f97..0edde3c 100644 --- a/net/ippolicy/policy_test.go +++ b/net/ippolicy/policy_test.go @@ -36,19 +36,24 @@ func TestPolicyKeepsSnapshotAfterFailedRefresh(t *testing.T) { if err != nil { t.Fatal(err) } - p := New(t.Context(), Config{Whitelist: whitelist}) + p := New(t.Context(), Config{Whitelist: whitelist, RefreshInterval: time.Nanosecond}) defer p.Stop() - if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { + evaluator, err := p.Load(t.Context(), true) + if err != nil { + t.Fatal(err) + } + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { t.Fatal("initial whitelist did not match") } if err := os.WriteFile(path, []byte("invalid entry\n"), 0o600); err != nil { t.Fatal(err) } - if _, err := whitelist.Load(t.Context(), true); err == nil { + evaluator, err = p.Load(t.Context(), true) + if err == nil { t.Fatal("invalid refresh unexpectedly succeeded") } - if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { t.Fatal("failed refresh replaced valid snapshot") } } @@ -71,12 +76,16 @@ func TestPolicyWhitelistPrecedesBlacklistExtra(t *testing.T) { if err != nil { t.Fatal(err) } - p := New(context.Background(), Config{Whitelist: whitelist, BlacklistExtra: blacklistExtra}) + p := New(context.Background(), Config{Whitelist: whitelist, BlacklistExtra: blacklistExtra, RefreshInterval: time.Nanosecond}) defer p.Stop() - if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { + evaluator, err := p.Load(context.Background(), true) + if err != nil { + t.Fatal(err) + } + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { t.Fatal("whitelist did not take precedence") } - if p.Evaluate(netip.MustParseAddr("192.0.2.2")) != Blacklisted { + if evaluator.Evaluate(netip.MustParseAddr("192.0.2.2")) != Blacklisted { t.Fatal("blacklist extra did not block") } }