root/golib
1
0
Fork 0
mirror of https://github.com/therootcompany/golib.git synced 2026-11-04 06:55:00 +00:00
Code Issues Projects Releases Wiki Activity

refactor(ippolicy): separate policy loading from evaluation

Browse Source
This commit is contained in:
AJ ONeal 2026-09-10 02:09:17 -06:00
parent db1d4c55a9
commit 61d068c95c
No known key found for this signature in database
4 changed files with 209 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
net/ippolicy/README.md
View File

@ -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

276
net/ippolicy/policy.go
View File

@ -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
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
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...)
}

61
net/ippolicy/policy_start_test.go
View File

@ -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")
}
}

23
net/ippolicy/policy_test.go
View File

@ -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")
}
}