From 27ba97de6d0a96c506456422d963886373797c9e Mon Sep 17 00:00:00 2001 From: AJ ONeal Date: Wed, 9 Sep 2026 22:36:04 -0600 Subject: [PATCH] refactor(net/ippolicy): publish shared cache policy --- net/ippolicy/README.md | 61 ++---- net/ippolicy/domains.go | 41 ++-- net/ippolicy/go.mod | 9 +- net/ippolicy/go.sum | 18 +- net/ippolicy/ipprefix.go | 223 +++++++++++++++++----- net/ippolicy/ipprefix_concurrency_test.go | 123 ++++++++++++ net/ippolicy/ipprefix_test.go | 4 +- net/ippolicy/policy.go | 191 +++++++++++------- net/ippolicy/policy_start_test.go | 82 ++++++++ net/ippolicy/policy_test.go | 32 +--- 10 files changed, 571 insertions(+), 213 deletions(-) create mode 100644 net/ippolicy/ipprefix_concurrency_test.go create mode 100644 net/ippolicy/policy_start_test.go diff --git a/net/ippolicy/README.md b/net/ippolicy/README.md index 82726de..6a84a57 100644 --- a/net/ippolicy/README.md +++ b/net/ippolicy/README.md @@ -1,36 +1,22 @@ # net/ippolicy -`ippolicy` composes independently refreshed IP sources into a fast policy -check. It uses `net/iplist` for local/HTTP source refresh and provides IP -matching, DNS refresh, and the legacy git blacklist. +`ippolicy` composes whitelist, blacklist-extra, and repository-backed prefix data into a fast policy +check. Sources expose lazy loading and explicit refresh; policy construction +does not start hidden refresh goroutines. ## Quick start -Construct each source with `net/iplist.NewSource`, then pass them to -`ippolicy.New`: - ```go -whitelist, err := iplist.NewSource(ctx, iplist.SourceConfig{ - Source: "allowed.csv", - CacheDir: "/var/cache/app/iplist", - // HTTPClient: nil, // uses golib's internal client +whitelist, err := iplist.NewIPList(ctx, iplist.IPListConfig{ + Source: "allowed.csv", + CacheDir: "/var/cache/app/iplist", }) if err != nil { return err } -defer whitelist.Close() +defer whitelist.Stop() -blacklistExtra, err := iplist.NewSource(ctx, iplist.SourceConfig{ - Source: "blocked-ips-extra.tsv", - CacheDir: "/var/cache/app/iplist", - Optional: true, -}) -if err != nil { - return err -} -defer blacklistExtra.Close() - -gitBlacklist, err := ippolicy.NewPrefixSet( +blacklist, err := ippolicy.NewIPPrefixSet( ctx, "https://github.com/bitwire-it/ipblocklist.git", "/var/cache/app/ipblocklist", @@ -40,17 +26,17 @@ gitBlacklist, err := ippolicy.NewPrefixSet( if err != nil { return err } -defer gitBlacklist.Close() +defer blacklist.Stop() policy := ippolicy.New(ctx, ippolicy.Config{ - Whitelist: whitelist, - Blacklist: gitBlacklist, - BlacklistExtra: blacklistExtra, - OnRefresh: func(e ippolicy.RefreshEvent) { - slog.Info("ippolicy refresh", "kind", e.Kind, "err", e.Err) - }, + Whitelist: whitelist, + Blacklist: blacklist, }) -defer policy.Close() +defer policy.Stop() + +if err := policy.Load(ctx, true); err != nil { + return err +} switch policy.Evaluate(addr) { case ippolicy.Blacklisted: @@ -60,14 +46,9 @@ case ippolicy.Whitelisted, ippolicy.Unlisted: } ``` -## Behavior +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. +Every source keeps its last-good snapshot after a failed update. -Each source refreshes independently and keeps its last valid data on failure. -Whitelist matches take precedence over blacklist matches. The policy does not -log; use `OnRefresh` for application logging and metrics. - -A nil `Whitelist` produces a policy where every address evaluates to -`Unlisted`. This intentionally drops the blacklists as well: without a -whitelist the policy cannot distinguish "allowed" from "unknown", so blocking -would be meaningless. The caller receives a `RefreshFallback` event so it can -log or alert. +A nil whitelist produces an `Unlisted` policy and intentionally ignores +blacklists: without a whitelist there is no meaningful allow-list boundary. diff --git a/net/ippolicy/domains.go b/net/ippolicy/domains.go index 0cc981e..efa6dd2 100644 --- a/net/ippolicy/domains.go +++ b/net/ippolicy/domains.go @@ -2,6 +2,7 @@ package ippolicy import ( "context" + "errors" "net" "net/netip" "sync" @@ -10,6 +11,7 @@ import ( "github.com/therootcompany/golib/net/dnsresolver" "github.com/therootcompany/golib/net/ipcohort" + "github.com/therootcompany/golib/sync/cachable" ) const DefaultDomainSetRefreshInterval = 5 * time.Minute @@ -81,10 +83,6 @@ func NewDomainSet(ctx context.Context, staticPrefixes []string, domains []string log().Info("domain set loaded", "static", commaify(len(staticPrefixes)), "domains", commaify(len(domains))) - if len(domains) > 0 { - ds.startRefreshLoop() - } - return ds } @@ -93,18 +91,23 @@ func (ds *DomainSet) Replace(staticPrefixes, domains []string) { staticPrefixes: append([]string(nil), staticPrefixes...), domains: append([]string(nil), domains...), }) - if len(domains) > 0 { - ds.startRefreshLoop() - } ds.resolveDomains(ds.ctx) ds.rebuildCohort() } -func (ds *DomainSet) startRefreshLoop() { - ds.start.Do(func() { go ds.refreshLoop(ds.interval) }) +var _ cachable.Tickable = (*DomainSet)(nil) + +// Start starts at most one optional background DNS refresh ticker. +func (ds *DomainSet) Start(ctx context.Context, interval time.Duration) { + ds.start.Do(func() { + if interval <= 0 { + interval = ds.interval + } + go ds.refreshLoop(ctx, interval) + }) } -func (ds *DomainSet) Close() error { +func (ds *DomainSet) Stop() error { if ds != nil && ds.cancel != nil { ds.cancel() } @@ -120,8 +123,8 @@ func (ds *DomainSet) Contains(addr netip.Addr) bool { return cohort.ContainsAddr(addr) } -func (ds *DomainSet) refreshLoop(interval time.Duration) { - ds.resolveDomains(ds.ctx) +func (ds *DomainSet) refreshLoop(ctx context.Context, interval time.Duration) { + ds.resolveDomains(ctx) ds.rebuildCohort() ticker := time.NewTicker(interval) @@ -129,10 +132,12 @@ func (ds *DomainSet) refreshLoop(interval time.Duration) { for { select { + case <-ctx.Done(): + return case <-ds.ctx.Done(): return case <-ticker.C: - ds.resolveDomains(ds.ctx) + ds.resolveDomains(ctx) ds.rebuildCohort() } } @@ -151,7 +156,11 @@ func (ds *DomainSet) resolveDomains(ctx context.Context) { for _, domain := range sources.domains { ips, _, err := resolver.LookupIP(ctx, domain) - if err != nil || len(ips) == 0 { + if err != nil { + if errors.Is(err, dnsresolver.ErrNoAddresses) || errors.Is(err, dnsresolver.ErrNameNotFound) { + // A definitive no-such-name result removes old addresses. + continue + } if old, ok := prev[domain]; ok { next[domain] = old log().Warn("resolve failed, keeping prior IPs", "domain", domain, "count", len(old), "err", err) @@ -160,6 +169,10 @@ func (ds *DomainSet) resolveDomains(ctx context.Context) { } continue } + if len(ips) == 0 { + // A successful empty result removes old addresses. + continue + } addrs := make([]string, 0, len(ips)) for _, ip := range ips { diff --git a/net/ippolicy/go.mod b/net/ippolicy/go.mod index 0faacac..c773e5f 100644 --- a/net/ippolicy/go.mod +++ b/net/ippolicy/go.mod @@ -3,17 +3,18 @@ module github.com/therootcompany/golib/net/ippolicy go 1.26.5 require ( - github.com/therootcompany/golib/net/dnsresolver v0.5.0 - github.com/therootcompany/golib/net/gitshallow v0.9.1 + github.com/therootcompany/golib/net/dnsresolver v0.6.0 + github.com/therootcompany/golib/net/gitshallow v0.10.0 github.com/therootcompany/golib/net/ipcohort v0.9.0 - github.com/therootcompany/golib/net/iplist v0.8.5 + github.com/therootcompany/golib/net/iplist v0.9.0 + github.com/therootcompany/golib/sync/cachable v0.1.0 ) require ( github.com/miekg/dns v1.1.69 // indirect github.com/therootcompany/golib/https v0.9.0 // indirect github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 // indirect - github.com/therootcompany/golib/net/httpcache v0.5.1 // indirect + github.com/therootcompany/golib/net/httpcache v0.6.0 // indirect golang.org/x/mod v0.30.0 // indirect golang.org/x/net v0.47.0 // indirect golang.org/x/sync v0.20.0 // indirect diff --git a/net/ippolicy/go.sum b/net/ippolicy/go.sum index 7b365a4..81bfcc4 100644 --- a/net/ippolicy/go.sum +++ b/net/ippolicy/go.sum @@ -6,16 +6,18 @@ github.com/therootcompany/golib/https v0.9.0 h1:D9gh0K6ljkvF805JUbylITn8SHPMo08q github.com/therootcompany/golib/https v0.9.0/go.mod h1:ilExqR/Uh7oB36oy9mwuNhANcFh+zdCfD3uJiZ11YAY= github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0 h1:R5zAPun5p3JPlNfcMD9es4LD9sqh6nWDNTflZBeeUJo= github.com/therootcompany/golib/io/transform/gsheet2csv v1.1.0/go.mod h1:CsntOAhhIMdzSAYuwuRZtfkdoiER9+TfSgYOIDWN71o= -github.com/therootcompany/golib/net/dnsresolver v0.5.0 h1:JTHvRVjc/cCL6SpWFo6btvv0lllW2wTYjRXLE8cOdMw= -github.com/therootcompany/golib/net/dnsresolver v0.5.0/go.mod h1:tOV99lX6ghpWPksxSFAt4CDe5XLefFzIGf2qXsZJap4= -github.com/therootcompany/golib/net/gitshallow v0.9.1 h1:VI/Q4jViKvVuJ+lzZrD7suhSXLfNNhUcd4ZJrkpTP/Q= -github.com/therootcompany/golib/net/gitshallow v0.9.1/go.mod h1:S/3OK7wMOqJJjP79U9Qw/l3jjEyrb/zsjJ8HAPAX/Fc= -github.com/therootcompany/golib/net/httpcache v0.5.1 h1:vQM758tBwPEqCo7ojpSeL3fHqgq29LYmvD2hvDTBfKY= -github.com/therootcompany/golib/net/httpcache v0.5.1/go.mod h1:mKSQJWpGSCDBzWkTCwT7dO1SCVCsFmdkRgKr39fNS1U= +github.com/therootcompany/golib/net/dnsresolver v0.6.0 h1:sy08rzX/2QowofAKnCw1QtTDoatPL6/FXZcb+Wz4GlM= +github.com/therootcompany/golib/net/dnsresolver v0.6.0/go.mod h1:tOV99lX6ghpWPksxSFAt4CDe5XLefFzIGf2qXsZJap4= +github.com/therootcompany/golib/net/gitshallow v0.10.0 h1:cLN9ElTmNWk5HSbaRD3sSFuGdpgy4YXc+LVzjzT0nM4= +github.com/therootcompany/golib/net/gitshallow v0.10.0/go.mod h1:S/3OK7wMOqJJjP79U9Qw/l3jjEyrb/zsjJ8HAPAX/Fc= +github.com/therootcompany/golib/net/httpcache v0.6.0 h1:qFBnSs6rZrtUQ+fweP4bHF2FQd/Syvs7mRN5nO5YtkY= +github.com/therootcompany/golib/net/httpcache v0.6.0/go.mod h1:mKSQJWpGSCDBzWkTCwT7dO1SCVCsFmdkRgKr39fNS1U= github.com/therootcompany/golib/net/ipcohort v0.9.0 h1:j3AaoW2u4XQJz1ya6fixv3pb3KgxIpZV8t0082/kS3E= github.com/therootcompany/golib/net/ipcohort v0.9.0/go.mod h1:Ljcnt2xOryVDGIVV5Kzt5hj2LhW4lBAWo5OeeVYU9ys= -github.com/therootcompany/golib/net/iplist v0.8.5 h1:swn/hcCt6fAG3tYApmxh1OlDak2PjQdczldZWaH1Eus= -github.com/therootcompany/golib/net/iplist v0.8.5/go.mod h1:vK5Na53RcLdtp/IPM/NX9BudyKhXp+8T+elmWBVWzeU= +github.com/therootcompany/golib/net/iplist v0.9.0 h1:rEDrflVvc5pQWt6sptrZVFWPZZ3M7ATUKjfIuFchKKI= +github.com/therootcompany/golib/net/iplist v0.9.0/go.mod h1:8ifRSgp4ptijldClyh688NLom8EZre6E16JZ2AHCEMw= +github.com/therootcompany/golib/sync/cachable v0.1.0 h1:swCXtXB116jhSmla81HaXi/9cPsqFI+Oq/6OcSfdU5Q= +github.com/therootcompany/golib/sync/cachable v0.1.0/go.mod h1:l/NZdGiSbW5CkLlkkRa0yObkqak2sj/jQTXXqXwvCl0= golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= diff --git a/net/ippolicy/ipprefix.go b/net/ippolicy/ipprefix.go index cbd2cb4..dd0c9fe 100644 --- a/net/ippolicy/ipprefix.go +++ b/net/ippolicy/ipprefix.go @@ -5,30 +5,42 @@ import ( "fmt" "net/netip" "os" + "sync" "sync/atomic" "time" "github.com/therootcompany/golib/net/gitshallow" "github.com/therootcompany/golib/net/ipcohort" + "github.com/therootcompany/golib/sync/cachable" ) -const DefaultPrefixSetRefreshInterval = time.Hour + 57*time.Minute + 13*time.Second +const ( + DefaultPrefixSetRefreshInterval = time.Hour + 57*time.Minute + 13*time.Second + DefaultRefreshTimeout = time.Hour + 57*time.Minute + 13*time.Second +) -type PrefixSet struct { - ctx context.Context - cancel context.CancelFunc - repo *gitshallow.Repo - files []string - cohort atomic.Pointer[ipcohort.Cohort] +type IPPrefixSet struct { + ctx context.Context + cancel context.CancelFunc + repo *gitshallow.Repo + files []string + cohort atomic.Pointer[ipcohort.Cohort] + start sync.Once + generation atomic.Uint64 + loadedAt atomic.Pointer[time.Time] + lastErr atomic.Pointer[error] + interval time.Duration + RefreshTimeout time.Duration + refresh cachable.Refresh } -func EmptyPrefixSet() *PrefixSet { - ps := &PrefixSet{} +func EmptyIPPrefixSet() *IPPrefixSet { + ps := &IPPrefixSet{} ps.cohort.Store(&ipcohort.Cohort{}) return ps } -func NewPrefixSet(ctx context.Context, repoURL, dataPath string, files []string, interval time.Duration) (*PrefixSet, error) { +func NewIPPrefixSet(ctx context.Context, repoURL, dataPath string, files []string, interval time.Duration) (*IPPrefixSet, error) { if err := os.MkdirAll(dataPath, 0o755); err != nil { return nil, fmt.Errorf("ippolicy: create data dir: %w", err) } @@ -39,27 +51,162 @@ func NewPrefixSet(ctx context.Context, repoURL, dataPath string, files []string, setCtx, cancel := context.WithCancel(ctx) repo := gitshallow.New(repoURL, dataPath, 0, "") - ps := &PrefixSet{ - ctx: setCtx, - cancel: cancel, - repo: repo, - files: files, + ps := &IPPrefixSet{ + ctx: setCtx, + cancel: cancel, + repo: repo, + files: files, + interval: interval, } ps.cohort.Store(&ipcohort.Cohort{}) - go ps.refreshLoop(interval) - return ps, nil } -func (ps *PrefixSet) Close() error { - if ps != nil && ps.cancel != nil { - ps.cancel() +var ( + _ cachable.Cacheable[ipcohort.Cohort] = (*IPPrefixSet)(nil) + _ cachable.Mutable[ipcohort.Cohort] = (*IPPrefixSet)(nil) + _ cachable.Inspectable = (*IPPrefixSet)(nil) + _ cachable.Tickable = (*IPPrefixSet)(nil) +) + +func (ps *IPPrefixSet) Current() *ipcohort.Cohort { + return ps.cohort.Load() +} + +func (ps *IPPrefixSet) Due(ctx context.Context) (bool, error) { + if err := ctx.Err(); err != nil { + return false, err } + status := ps.Status() + return !status.HasValue || status.StaleAt.IsZero() || !time.Now().Before(status.StaleAt), nil +} + +func (ps *IPPrefixSet) Revalidate(ctx context.Context) (started bool, err error) { + due, err := ps.Due(ctx) + if err != nil || !due { + return false, err + } + if !ps.refresh.Begin() { + return false, nil + } + go func() { + timeout := ps.RefreshTimeout + if timeout <= 0 { + timeout = DefaultRefreshTimeout + } + refreshCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + defer cancel() + ps.refresh.Finish(ps.update(refreshCtx)) + }() + return true, nil +} + +func (ps *IPPrefixSet) update(ctx context.Context) error { + generation := ps.generation.Load() + cohort, err := ps.reload(ctx) + if err != nil { + errCopy := err + ps.lastErr.Store(&errCopy) + return err + } + if ps.generation.Load() != generation { + return nil + } + if cohort != nil { + ps.cohort.Store(cohort) + } + now := time.Now() + ps.loadedAt.Store(&now) + ps.lastErr.Store(nil) return nil } -func (ps *PrefixSet) Contains(addr netip.Addr) bool { +func (ps *IPPrefixSet) Set(cohort *ipcohort.Cohort) error { + ps.generation.Add(1) + ps.cohort.Store(cohort) + if cohort == nil { + ps.loadedAt.Store(nil) + return nil + } + now := time.Now() + ps.loadedAt.Store(&now) + return nil +} + +func (ps *IPPrefixSet) Clear() error { + return ps.Set(nil) +} + +func (ps *IPPrefixSet) Status() cachable.Status { + loadedAt := time.Time{} + if t := ps.loadedAt.Load(); t != nil { + loadedAt = *t + } + var lastErr error + if p := ps.lastErr.Load(); p != nil { + lastErr = *p + } + status := cachable.Status{LoadedAt: loadedAt, HasValue: ps.Current() != nil, Refreshing: ps.refresh.Running(), LastError: lastErr} + if !loadedAt.IsZero() { + status.StaleAt = loadedAt.Add(ps.interval) + } + return status +} + +func (ps *IPPrefixSet) Load(ctx context.Context, wait bool) (*ipcohort.Cohort, error) { + if _, err := ps.Revalidate(ctx); err != nil { + return ps.Current(), err + } + if ps.Current() == nil || wait { + if err := ps.refresh.Wait(ctx); err != nil { + return ps.Current(), err + } + } + return ps.Current(), nil +} + +func (ps *IPPrefixSet) Start(ctx context.Context, interval time.Duration) { + ps.start.Do(func() { + if interval <= 0 { + interval = DefaultPrefixSetRefreshInterval + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ps.ctx.Done(): + return + case <-ticker.C: + if _, err := ps.Revalidate(ps.ctx); err != nil { + log().Warn("prefix set reload failed", "err", err) + } + } + } + }() + }) +} + +func (ps *IPPrefixSet) Stop() error { + if ps == nil { + return nil + } + if ps.cancel != nil { + ps.cancel() + } + timeout := ps.RefreshTimeout + if timeout <= 0 { + timeout = DefaultRefreshTimeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + return ps.refresh.Stop(ctx) +} + +func (ps *IPPrefixSet) Contains(addr netip.Addr) bool { cohort := ps.cohort.Load() if cohort == nil { cohort = &ipcohort.Cohort{} @@ -68,10 +215,10 @@ func (ps *PrefixSet) Contains(addr netip.Addr) bool { return cohort.ContainsAddr(addr) } -func (ps *PrefixSet) reload(ctx context.Context) error { - updated, err := ps.repo.Fetch(ctx) +func (ps *IPPrefixSet) reload(ctx context.Context) (*ipcohort.Cohort, error) { + updated, err := ps.repo.Update(ctx) if err != nil { - return err + return nil, err } paths := make([]string, len(ps.files)) @@ -80,18 +227,16 @@ func (ps *PrefixSet) reload(ctx context.Context) error { } if cachedCohortValid(updated, ps.cohort.Load(), paths) { - return nil + return nil, nil } cohort, err := ipcohort.LoadFiles(paths...) if err != nil { - return fmt.Errorf("load files: %w", err) + return nil, fmt.Errorf("load files: %w", err) } - ps.cohort.Store(cohort) - log().Info("prefix set loaded", "entries", commaify(cohort.Size())) - return nil + return cohort, nil } // cachedCohortValid reports whether the current cohort can be reused @@ -109,23 +254,3 @@ func filesPresent(paths []string) bool { } return true } - -func (ps *PrefixSet) refreshLoop(interval time.Duration) { - if err := ps.reload(ps.ctx); err != nil { - log().Warn("prefix set initial load (will retry)", "err", err) - } - - ticker := time.NewTicker(interval) - defer ticker.Stop() - - for { - select { - case <-ps.ctx.Done(): - return - case <-ticker.C: - if err := ps.reload(ps.ctx); err != nil { - log().Warn("prefix set reload failed", "err", err) - } - } - } -} diff --git a/net/ippolicy/ipprefix_concurrency_test.go b/net/ippolicy/ipprefix_concurrency_test.go new file mode 100644 index 0000000..cbdb7ab --- /dev/null +++ b/net/ippolicy/ipprefix_concurrency_test.go @@ -0,0 +1,123 @@ +package ippolicy + +import ( + "context" + "net/netip" + "testing" + "time" + + "github.com/therootcompany/golib/net/ipcohort" +) + +// TestIPPrefixSet_SetPublishesManually confirms Set publishes immediately and +// bumps the generation so a subsequent stale refresh cannot overwrite it. +func TestIPPrefixSet_SetPublishesManually(t *testing.T) { + ps := EmptyIPPrefixSet() + + cohort, err := ipcohort.Parse([]string{"192.0.2.0/24"}) + if err != nil { + t.Fatal(err) + } + if err := ps.Set(cohort); err != nil { + t.Fatalf("Set: %v", err) + } + if got := ps.Current(); got == nil || got.Size() == 0 { + t.Fatalf("after Set Current = %v", got) + } + if !ps.Contains(netip.MustParseAddr("192.0.2.1")) { + t.Error("Contains should match after Set") + } + if ps.Status().LoadedAt.IsZero() { + t.Error("LoadedAt should be set after Set") + } +} + +// TestIPPrefixSet_ClearNilsCohort confirms Clear removes the published value +// and resets the loaded clock. +func TestIPPrefixSet_ClearNilsCohort(t *testing.T) { + ps := EmptyIPPrefixSet() + cohort, err := ipcohort.Parse([]string{"192.0.2.0/24"}) + if err != nil { + t.Fatal(err) + } + if err := ps.Set(cohort); err != nil { + t.Fatal(err) + } + if err := ps.Clear(); err != nil { + t.Fatalf("Clear: %v", err) + } + if ps.Current() != nil { + t.Error("Current should be nil after Clear") + } + if !ps.Status().LoadedAt.IsZero() { + t.Error("LoadedAt should be zero after Clear") + } +} + +// TestIPPrefixSet_DueWhenEmpty confirms an empty set is due. +func TestIPPrefixSet_DueWhenEmpty(t *testing.T) { + ps := EmptyIPPrefixSet() + due, err := ps.Due(context.Background()) + if err != nil { + t.Fatalf("Due: %v", err) + } + if !due { + t.Error("empty IPPrefixSet should be due") + } +} + +// TestIPPrefixSet_ClearIsIdempotent confirms repeated Clear is safe. +func TestIPPrefixSet_ClearIsIdempotent(t *testing.T) { + ps := EmptyIPPrefixSet() + for range 3 { + if err := ps.Clear(); err != nil { + t.Fatalf("Clear: %v", err) + } + } + if ps.Current() != nil { + t.Error("Current should be nil after repeated Clear") + } +} + +// TestIPPrefixSet_StopOnZeroValue confirms a zero-value set can be stopped +// without panicking (defensive). +func TestIPPrefixSet_StopOnZeroValue(t *testing.T) { + var ps *IPPrefixSet + // nil receiver must not panic. + ps.Stop() +} + +// TestIPPrefixSet_StatusRefreshingFlag confirms Status reports Refreshing +// during an in-flight refresh. We cannot easily block a git refresh, so this +// is a smoke test that Refreshing starts false. +func TestIPPrefixSet_StatusNotRefreshingAtRest(t *testing.T) { + ps := EmptyIPPrefixSet() + if ps.Status().Refreshing { + t.Error("Refreshing should be false at rest") + } +} + +// TestIPPrefixSet_IntervalDrivesStaleAt confirms StaleAt is LoadedAt+interval. +func TestIPPrefixSet_IntervalDrivesStaleAt(t *testing.T) { + ps := EmptyIPPrefixSet() + ps.interval = time.Hour + cohort, err := ipcohort.Parse([]string{"192.0.2.0/24"}) + if err != nil { + t.Fatal(err) + } + if err := ps.Set(cohort); err != nil { + t.Fatal(err) + } + status := ps.Status() + if status.StaleAt.IsZero() { + t.Fatal("StaleAt should be set after Set with a non-zero interval") + } + want := status.LoadedAt.Add(time.Hour) + if !status.StaleAt.Equal(want) { + t.Errorf("StaleAt = %v, want %v", status.StaleAt, want) + } + // Right after Set, with a 1h interval, we should NOT be due. + if due, _ := ps.Due(context.Background()); due { + t.Error("Due should be false immediately after Set with a future StaleAt") + } +} diff --git a/net/ippolicy/ipprefix_test.go b/net/ippolicy/ipprefix_test.go index b62f980..00ea155 100644 --- a/net/ippolicy/ipprefix_test.go +++ b/net/ippolicy/ipprefix_test.go @@ -10,10 +10,10 @@ import ( ) func TestPrefixSetContainsZeroValue(t *testing.T) { - var ps PrefixSet + var ps IPPrefixSet // Zero-value must not panic — Contains self-initialises via CAS. if ps.Contains(netip.MustParseAddr("192.0.2.1")) { - t.Fatal("zero-value PrefixSet must not contain an address") + t.Fatal("zero-value IPPrefixSet must not contain an address") } // After first call the cohort should be non-nil. if ps.cohort.Load() == nil { diff --git a/net/ippolicy/policy.go b/net/ippolicy/policy.go index 6fcc5f3..54a10a6 100644 --- a/net/ippolicy/policy.go +++ b/net/ippolicy/policy.go @@ -7,15 +7,16 @@ import ( "errors" "fmt" "net/netip" + "sync" + "time" "github.com/therootcompany/golib/net/iplist" ) type Config struct { - Whitelist *iplist.Source - Blacklist *PrefixSet - BlacklistExtra *iplist.Source - OnRefresh func(RefreshEvent) + Whitelist *iplist.IPList + Blacklist *IPPrefixSet + BlacklistExtra *iplist.IPList } // Decision is the mutually exclusive result of evaluating an address. @@ -40,41 +41,20 @@ func (d Decision) String() string { } } -type RefreshKind uint8 - -const ( - RefreshSuccess RefreshKind = iota - RefreshFailure - RefreshFallback -) - -func (k RefreshKind) String() string { - switch k { - case RefreshSuccess: - return "success" - case RefreshFailure: - return "failure" - case RefreshFallback: - return "fallback" - default: - return fmt.Sprintf("refresh(%d)", k) - } -} - -type RefreshEvent struct { - Kind RefreshKind - Err error -} - type Policy struct { - onRefresh func(RefreshEvent) - whitelist *DomainSet - extra *DomainSet - git *PrefixSet + whitelist *DomainSet + blacklistExtra *DomainSet + blacklist *IPPrefixSet + whitelistSource *iplist.IPList + blacklistExtraSource *iplist.IPList + + start sync.Once + cancel context.CancelFunc + loadMu sync.Mutex } // New creates a Policy from the given sources. The caller is responsible for -// constructing and later closing the *iplist.Source values; New does not take +// 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 @@ -83,41 +63,67 @@ type Policy struct { // 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. func New(ctx context.Context, config Config) *Policy { - p := &Policy{onRefresh: config.OnRefresh} + p := &Policy{} if config.Whitelist == nil { p.whitelist = EmptyDomainSet() - p.emit(RefreshFallback, errors.New("whitelist source is required")) return p } - p.whitelist = NewDomainSetFromEntries(ctx, config.Whitelist.Entries(), DefaultDomainSetRefreshInterval) - config.Whitelist.Subscribe(func(event iplist.SourceEvent) { - if event.Err != nil { - p.emit(RefreshFailure, event.Err) - return - } - p.whitelist.ReplaceEntries(config.Whitelist.Entries()) - p.emit(RefreshSuccess, nil) - }) + p.whitelistSource = config.Whitelist + p.whitelist = NewDomainSetFromEntries(ctx, sourceEntries(config.Whitelist), DefaultDomainSetRefreshInterval) if config.BlacklistExtra != nil { - p.extra = NewDomainSetFromEntries(ctx, config.BlacklistExtra.Entries(), DefaultDomainSetRefreshInterval) - config.BlacklistExtra.Subscribe(func(event iplist.SourceEvent) { - if event.Err != nil { - p.emit(RefreshFailure, event.Err) - return - } - p.extra.ReplaceEntries(config.BlacklistExtra.Entries()) - p.emit(RefreshSuccess, nil) - }) + p.blacklistExtraSource = config.BlacklistExtra + p.blacklistExtra = NewDomainSetFromEntries(ctx, sourceEntries(config.BlacklistExtra), DefaultDomainSetRefreshInterval) } else { - p.extra = EmptyDomainSet() + p.blacklistExtra = EmptyDomainSet() } - p.git = config.Blacklist - p.emit(RefreshSuccess, nil) + p.blacklist = config.Blacklist 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 { + p.loadMu.Lock() + defer p.loadMu.Unlock() + + if p.whitelistSource == nil { + return nil + } + if _, err := p.whitelistSource.Load(ctx, wait); err != nil { + return err + } + p.whitelist.ReplaceEntries(sourceEntries(p.whitelistSource)) + if p.blacklistExtraSource != nil { + if _, err := p.blacklistExtraSource.Load(ctx, wait); err != nil { + return err + } + p.blacklistExtra.ReplaceEntries(sourceEntries(p.blacklistExtraSource)) + } + if p.blacklist != nil { + if _, err := p.blacklist.Load(ctx, wait); err != nil { + return err + } + } + return nil +} + +func sourceEntries(source *iplist.IPList) []string { + entries := source.Current() + if entries == nil { + return nil + } + return *entries +} + func (p *Policy) Evaluate(addr netip.Addr) Decision { if p.whitelist == nil { return Unlisted @@ -125,32 +131,69 @@ func (p *Policy) Evaluate(addr netip.Addr) Decision { if p.whitelist.Contains(addr) { return Whitelisted } - if (p.extra != nil && p.extra.Contains(addr)) || - (p.git != nil && p.git.Contains(addr)) { + if (p.blacklistExtra != nil && p.blacklistExtra.Contains(addr)) || + (p.blacklist != nil && p.blacklist.Contains(addr)) { return Blacklisted } return Unlisted } -func (p *Policy) Close() error { +// 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 { - _ = p.whitelist.Close() + if err := p.whitelist.Stop(); err != nil { + errs = append(errs, err) + } } - if p.extra != nil { - _ = p.extra.Close() + if p.blacklistExtra != nil { + if err := p.blacklistExtra.Stop(); err != nil { + errs = append(errs, err) + } } - if p.git != nil { - _ = p.git.Close() + if p.blacklist != nil { + if err := p.blacklist.Stop(); err != nil { + errs = append(errs, err) + } } - return nil -} - -func (p *Policy) emit(kind RefreshKind, err error) { - if p.onRefresh == nil { - return - } - p.onRefresh(RefreshEvent{Kind: kind, Err: err}) + return errors.Join(errs...) } diff --git a/net/ippolicy/policy_start_test.go b/net/ippolicy/policy_start_test.go new file mode 100644 index 0000000..7fd21c8 --- /dev/null +++ b/net/ippolicy/policy_start_test.go @@ -0,0 +1,82 @@ +package ippolicy + +import ( + "net/netip" + "os" + "path/filepath" + "testing" + "time" + + "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) { + 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, + }) + if err != nil { + t.Fatal(err) + } + + p := New(t.Context(), Config{Whitelist: whitelist}) + defer p.Stop() + + if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { + t.Fatal("initial whitelist did not match 192.0.2.1") + } + + // 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(), + }) + 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) + } +} diff --git a/net/ippolicy/policy_test.go b/net/ippolicy/policy_test.go index 3598df9..3de0f97 100644 --- a/net/ippolicy/policy_test.go +++ b/net/ippolicy/policy_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/therootcompany/golib/net/iplist" ) @@ -24,19 +25,6 @@ func TestEnumStrings(t *testing.T) { t.Fatalf("Decision(%d).String() = %q, want %q", test.value, got, test.want) } } - for _, test := range []struct { - value RefreshKind - want string - }{ - {RefreshSuccess, "success"}, - {RefreshFailure, "failure"}, - {RefreshFallback, "fallback"}, - {RefreshKind(99), "refresh(99)"}, - } { - if got := test.value.String(); got != test.want { - t.Fatalf("RefreshKind(%d).String() = %q, want %q", test.value, got, test.want) - } - } } func TestPolicyKeepsSnapshotAfterFailedRefresh(t *testing.T) { @@ -44,12 +32,12 @@ func TestPolicyKeepsSnapshotAfterFailedRefresh(t *testing.T) { if err := os.WriteFile(path, []byte("192.0.2.1\n"), 0o600); err != nil { t.Fatal(err) } - whitelist, err := iplist.NewSource(t.Context(), iplist.SourceConfig{Source: path}) + whitelist, err := iplist.NewIPList(t.Context(), iplist.IPListConfig{Source: path, CacheDir: t.TempDir(), RefreshInterval: time.Nanosecond}) if err != nil { t.Fatal(err) } p := New(t.Context(), Config{Whitelist: whitelist}) - defer func() { _ = p.Close() }() + defer p.Stop() if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { t.Fatal("initial whitelist did not match") @@ -57,7 +45,7 @@ func TestPolicyKeepsSnapshotAfterFailedRefresh(t *testing.T) { if err := os.WriteFile(path, []byte("invalid entry\n"), 0o600); err != nil { t.Fatal(err) } - if err := whitelist.Refresh(); err == nil { + if _, err := whitelist.Load(t.Context(), true); err == nil { t.Fatal("invalid refresh unexpectedly succeeded") } if p.Evaluate(netip.MustParseAddr("192.0.2.1")) != Whitelisted { @@ -65,7 +53,7 @@ func TestPolicyKeepsSnapshotAfterFailedRefresh(t *testing.T) { } } -func TestPolicyWhitelistPrecedesExtraBlacklist(t *testing.T) { +func TestPolicyWhitelistPrecedesBlacklistExtra(t *testing.T) { dir := t.TempDir() white := filepath.Join(dir, "white.tsv") black := filepath.Join(dir, "black.tsv") @@ -75,20 +63,20 @@ func TestPolicyWhitelistPrecedesExtraBlacklist(t *testing.T) { if err := os.WriteFile(black, []byte("192.0.2.0/24\n"), 0o600); err != nil { t.Fatal(err) } - whitelist, err := iplist.NewSource(context.Background(), iplist.SourceConfig{Source: white}) + whitelist, err := iplist.NewIPList(context.Background(), iplist.IPListConfig{Source: white, CacheDir: dir}) if err != nil { t.Fatal(err) } - extra, err := iplist.NewSource(context.Background(), iplist.SourceConfig{Source: black}) + blacklistExtra, err := iplist.NewIPList(context.Background(), iplist.IPListConfig{Source: black, CacheDir: dir}) if err != nil { t.Fatal(err) } - p := New(context.Background(), Config{Whitelist: whitelist, BlacklistExtra: extra}) - defer func() { _ = p.Close() }() + p := New(context.Background(), Config{Whitelist: whitelist, BlacklistExtra: blacklistExtra}) + defer p.Stop() if p.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 { - t.Fatal("extra blacklist did not block") + t.Fatal("blacklist extra did not block") } }