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(net/ippolicy): publish shared cache policy

Browse Source
This commit is contained in:
AJ ONeal 2026-09-09 22:36:04 -06:00
parent 48a4f2a145
commit 27ba97de6d
No known key found for this signature in database
10 changed files with 571 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
net/ippolicy/README.md
View File

@ -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{
whitelist, err := iplist.NewIPList(ctx, iplist.IPListConfig{
Source: "allowed.csv",
CacheDir: "/var/cache/app/iplist",
// HTTPClient: nil, // uses golib's internal client
})
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)
},
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.

41
net/ippolicy/domains.go
View File

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

9
net/ippolicy/go.mod
View File

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

18
net/ippolicy/go.sum
View File

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

205
net/ippolicy/ipprefix.go
View File

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

123
net/ippolicy/ipprefix_concurrency_test.go Normal file
View File

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

4
net/ippolicy/ipprefix_test.go
View File

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

185
net/ippolicy/policy.go
View File

@ -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
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.git != nil {
_ = p.git.Close()
if p.blacklistExtra != nil {
if err := p.blacklistExtra.Stop(); err != nil {
errs = append(errs, err)
}
return nil
}
func (p *Policy) emit(kind RefreshKind, err error) {
if p.onRefresh == nil {
return
if p.blacklist != nil {
if err := p.blacklist.Stop(); err != nil {
errs = append(errs, err)
}
p.onRefresh(RefreshEvent{Kind: kind, Err: err})
}
return errors.Join(errs...)
}

82
net/ippolicy/policy_start_test.go Normal file
View File

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

32
net/ippolicy/policy_test.go
View File

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