From d52c3ee76ec4ae6c38640a0ba64ea496ba160119 Mon Sep 17 00:00:00 2001 From: benvin Date: Sun, 19 Jul 2026 21:44:30 +1000 Subject: [PATCH] Add central ASN address-group expander Expand asn address groups to concrete prefixes centrally (one iplocate key, consistent fleet-wide) and refresh them on a per-group TTL (default 24h). A background Refresher scans for due groups, unions each group's ASNs to a deduped prefix set, and writes them to a new resolved/resolved_at column (migration 0002). Fail-safe: a lookup error or empty expansion keeps the last-good set, never emptying it. The compiler folds resolved prefixes into the rendered set members; membership churn bumps the generation but never rewrites rules. The iplocate client is endpoint-configurable and response-tolerant, documented as needing endpoint/key confirmation. Unit tests cover TTL parsing, due-checks, and union/dedup/error propagation with a fake expander. --- cmd/tomswallapi/main.go | 15 +++ internal/asnexpand/asnexpand.go | 126 ++++++++++++++++++ internal/asnexpand/asnexpand_test.go | 76 +++++++++++ internal/asnexpand/iplocate.go | 94 +++++++++++++ internal/compiler/compiler.go | 3 +- .../migrations/0002_asn_resolution.sql | 7 + internal/model/model.go | 6 + internal/store/store.go | 32 ++++- 8 files changed, 355 insertions(+), 4 deletions(-) create mode 100644 internal/asnexpand/asnexpand.go create mode 100644 internal/asnexpand/asnexpand_test.go create mode 100644 internal/asnexpand/iplocate.go create mode 100644 internal/database/migrations/0002_asn_resolution.sql diff --git a/cmd/tomswallapi/main.go b/cmd/tomswallapi/main.go index a0ee1fe..ba36218 100644 --- a/cmd/tomswallapi/main.go +++ b/cmd/tomswallapi/main.go @@ -13,9 +13,11 @@ import ( "os/signal" "syscall" + "git.unkin.net/unkin/tomswallapi/internal/asnexpand" "git.unkin.net/unkin/tomswallapi/internal/config" "git.unkin.net/unkin/tomswallapi/internal/database" "git.unkin.net/unkin/tomswallapi/internal/server" + "git.unkin.net/unkin/tomswallapi/internal/store" ) var version = "dev" @@ -52,6 +54,19 @@ func main() { slog.Warn("TOMSWALLAPI_AGENT_TOKEN is not set; agent config endpoint is disabled") } + // Start the central ASN expander when a key is configured. Without one, asn + // address groups simply stay unexpanded (their sets render empty and inert). + if cfg.IPLocateAPIKey != "" { + refresher := &asnexpand.Refresher{ + Store: store.New(db.Pool), + Expander: asnexpand.NewIPLocate(cfg.IPLocateAPIKey), + } + go refresher.Run(ctx) + slog.Info("started ASN expander") + } else { + slog.Warn("TOMSWALLAPI_IPLOCATE_API_KEY is not set; asn address groups will not be expanded") + } + srv := server.New(server.Options{ DB: db, WriteToken: cfg.WriteToken, diff --git a/internal/asnexpand/asnexpand.go b/internal/asnexpand/asnexpand.go new file mode 100644 index 0000000..2cdac63 --- /dev/null +++ b/internal/asnexpand/asnexpand.go @@ -0,0 +1,126 @@ +// Package asnexpand centrally expands ASN address groups into concrete prefixes +// and refreshes them on a per-group TTL. Expansion is centralized (one API key, +// consistent fleet-wide results); devices never call the upstream provider. +package asnexpand + +import ( + "context" + "log/slog" + "time" + + "git.unkin.net/unkin/tomswallapi/internal/model" + "git.unkin.net/unkin/tomswallapi/internal/store" +) + +// DefaultTTL is used when an ASN group specifies no refresh interval. +const DefaultTTL = 24 * time.Hour + +// Expander resolves an ASN (e.g. "13335") to its announced prefixes. +type Expander interface { + Prefixes(ctx context.Context, asn string) ([]string, error) +} + +// ParseTTL parses a group's refresh string (e.g. "24h", "6h"); invalid or empty +// values fall back to DefaultTTL. +func ParseTTL(s string) time.Duration { + if s == "" { + return DefaultTTL + } + d, err := time.ParseDuration(s) + if err != nil || d <= 0 { + return DefaultTTL + } + return d +} + +// due reports whether a group needs re-expansion given its last resolution time. +func due(g model.AddressGroup, now time.Time) bool { + if g.ResolvedAt == nil { + return true + } + return now.Sub(*g.ResolvedAt) >= ParseTTL(g.Refresh) +} + +// Refresher periodically expands ASN groups whose TTL has elapsed and writes the +// resulting prefixes back to the store. +type Refresher struct { + Store *store.Store + Expander Expander + Interval time.Duration // how often to scan for due groups + Now func() time.Time +} + +// Run scans on Interval until ctx is cancelled. One immediate scan runs first. +func (r *Refresher) Run(ctx context.Context) { + if r.Interval <= 0 { + r.Interval = 5 * time.Minute + } + if r.Now == nil { + r.Now = time.Now + } + r.scan(ctx) + t := time.NewTicker(r.Interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + r.scan(ctx) + } + } +} + +// scan expands every due ASN group once. Failures are logged and the prior +// resolution is kept — a provider outage must never empty a set (fail-safe). +func (r *Refresher) scan(ctx context.Context) { + groups, err := r.Store.ListAddressGroups(ctx) + if err != nil { + slog.Error("asnexpand: list groups", "err", err) + return + } + now := r.Now() + for _, g := range groups { + if g.Type != model.GroupASN || !due(g, now) { + continue + } + prefixes, err := r.expand(ctx, g) + if err != nil { + slog.Warn("asnexpand: expansion failed, keeping last-good", + "group", g.Name, "err", err) + continue + } + if len(prefixes) == 0 { + // A genuinely-empty expansion is suspicious; do not clobber last-good. + slog.Warn("asnexpand: empty expansion, keeping last-good", "group", g.Name) + continue + } + if err := r.Store.UpdateResolvedPrefixes(ctx, g.Name, prefixes); err != nil { + slog.Error("asnexpand: store prefixes", "group", g.Name, "err", err) + continue + } + slog.Info("asnexpand: refreshed", "group", g.Name, "prefixes", len(prefixes)) + } +} + +// expand unions the prefixes of every ASN in the group. If any single ASN lookup +// fails, the whole expansion fails so we keep the last-good set rather than +// publishing a partial one. +func (r *Refresher) expand(ctx context.Context, g model.AddressGroup) ([]string, error) { + seen := map[string]struct{}{} + var out []string + for _, asn := range g.Members { + prefixes, err := r.Expander.Prefixes(ctx, asn) + if err != nil { + return nil, err + } + for _, p := range prefixes { + if _, ok := seen[p]; ok { + continue + } + seen[p] = struct{}{} + out = append(out, p) + } + } + return out, nil +} diff --git a/internal/asnexpand/asnexpand_test.go b/internal/asnexpand/asnexpand_test.go new file mode 100644 index 0000000..532937c --- /dev/null +++ b/internal/asnexpand/asnexpand_test.go @@ -0,0 +1,76 @@ +package asnexpand + +import ( + "context" + "errors" + "testing" + "time" + + "git.unkin.net/unkin/tomswallapi/internal/model" +) + +func TestParseTTL(t *testing.T) { + cases := map[string]time.Duration{ + "": DefaultTTL, + "bad": DefaultTTL, + "0s": DefaultTTL, + "-1h": DefaultTTL, + "6h": 6 * time.Hour, + "30m": 30 * time.Minute, + "24h": 24 * time.Hour, + } + for in, want := range cases { + if got := ParseTTL(in); got != want { + t.Errorf("ParseTTL(%q) = %v, want %v", in, got, want) + } + } +} + +func TestDue(t *testing.T) { + now := time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC) + old := now.Add(-25 * time.Hour) + recent := now.Add(-1 * time.Hour) + + if !due(model.AddressGroup{Refresh: "24h"}, now) { + t.Error("never-resolved group should be due") + } + if !due(model.AddressGroup{Refresh: "24h", ResolvedAt: &old}, now) { + t.Error("group past its TTL should be due") + } + if due(model.AddressGroup{Refresh: "24h", ResolvedAt: &recent}, now) { + t.Error("group within its TTL should not be due") + } +} + +type fakeExpander struct { + byASN map[string][]string + err error +} + +func (f fakeExpander) Prefixes(_ context.Context, asn string) ([]string, error) { + if f.err != nil { + return nil, f.err + } + return f.byASN[asn], nil +} + +func TestExpandUnionsAndDedups(t *testing.T) { + r := &Refresher{Expander: fakeExpander{byASN: map[string][]string{ + "13335": {"1.1.1.0/24", "104.16.0.0/13"}, + "209242": {"104.16.0.0/13", "203.0.113.0/24"}, // overlaps with 13335 + }}} + got, err := r.expand(context.Background(), model.AddressGroup{Members: []string{"13335", "209242"}}) + if err != nil { + t.Fatalf("expand: %v", err) + } + if len(got) != 3 { + t.Fatalf("want 3 deduped prefixes, got %d: %v", len(got), got) + } +} + +func TestExpandPropagatesError(t *testing.T) { + r := &Refresher{Expander: fakeExpander{err: errors.New("boom")}} + if _, err := r.expand(context.Background(), model.AddressGroup{Members: []string{"13335"}}); err == nil { + t.Fatal("expected error to propagate so last-good is kept") + } +} diff --git a/internal/asnexpand/iplocate.go b/internal/asnexpand/iplocate.go new file mode 100644 index 0000000..ac7b519 --- /dev/null +++ b/internal/asnexpand/iplocate.go @@ -0,0 +1,94 @@ +package asnexpand + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +// IPLocate expands ASNs via the iplocate.io IP-intelligence API. +// +// NOTE: iplocate's documented ASN data type is returned as part of an IP lookup +// (it carries the single `route` for that IP), so the exact ASN->prefixes +// endpoint depends on the account/plan. This client is deliberately tolerant of +// response shape and endpoint-configurable so it can be pointed at the correct +// endpoint once the API key and plan are confirmed. It expects a JSON body +// containing a list of prefixes under any of: "prefixes", "routes", "cidrs". +type IPLocate struct { + // BaseURL is the API root, default https://iplocate.io/api. The ASN path is + // BaseURL + "/asn/" + asn. + BaseURL string + // APIKey is sent as the ?apikey= query parameter (iplocate's scheme). + APIKey string + HTTP *http.Client +} + +// NewIPLocate builds a client with sane defaults. +func NewIPLocate(apiKey string) *IPLocate { + return &IPLocate{ + BaseURL: "https://iplocate.io/api", + APIKey: apiKey, + HTTP: &http.Client{Timeout: 15 * time.Second}, + } +} + +// asnResponse tolerates the several field names an ASN-prefix payload might use. +type asnResponse struct { + Prefixes []string `json:"prefixes"` + Routes []string `json:"routes"` + CIDRs []string `json:"cidrs"` +} + +func (r asnResponse) list() []string { + switch { + case len(r.Prefixes) > 0: + return r.Prefixes + case len(r.Routes) > 0: + return r.Routes + default: + return r.CIDRs + } +} + +// Prefixes implements Expander. +func (c *IPLocate) Prefixes(ctx context.Context, asn string) ([]string, error) { + if c.APIKey == "" { + return nil, fmt.Errorf("iplocate: no API key configured") + } + asn = strings.TrimPrefix(strings.ToUpper(strings.TrimSpace(asn)), "AS") + + u, err := url.Parse(fmt.Sprintf("%s/asn/AS%s", strings.TrimRight(c.BaseURL, "/"), asn)) + if err != nil { + return nil, err + } + q := u.Query() + q.Set("apikey", c.APIKey) + u.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + + resp, err := c.HTTP.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("iplocate: AS%s: status %d: %s", asn, resp.StatusCode, strings.TrimSpace(string(body))) + } + + var parsed asnResponse + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, fmt.Errorf("iplocate: AS%s: decoding response: %w", asn, err) + } + return parsed.list(), nil +} diff --git a/internal/compiler/compiler.go b/internal/compiler/compiler.go index da439fd..b35c83a 100644 --- a/internal/compiler/compiler.go +++ b/internal/compiler/compiler.go @@ -243,7 +243,8 @@ func renderSet(g model.AddressGroup) RenderedSet { case model.GroupDNS: rs.FQDNs = g.Members case model.GroupASN: - rs.ASNs = g.Members // expanded prefixes are attached out-of-band by the ASN expander + rs.ASNs = g.Members // source ASNs + rs.Members = g.Resolved // concrete prefixes from the central expander (may be empty until first expansion) } return rs } diff --git a/internal/database/migrations/0002_asn_resolution.sql b/internal/database/migrations/0002_asn_resolution.sql new file mode 100644 index 0000000..acf50e5 --- /dev/null +++ b/internal/database/migrations/0002_asn_resolution.sql @@ -0,0 +1,7 @@ +-- Cache expanded ASN prefixes on the address group. `resolved` holds the concrete +-- CIDRs the expander last produced; `resolved_at` timestamps the last successful +-- expansion. Membership churn here is deliberately separate from rule definition: +-- the compiler folds `resolved` into the rendered set without touching rules. +ALTER TABLE address_groups + ADD COLUMN resolved JSONB NOT NULL DEFAULT '[]'::jsonb, + ADD COLUMN resolved_at TIMESTAMPTZ; diff --git a/internal/model/model.go b/internal/model/model.go index d32e462..920b38d 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -5,6 +5,7 @@ package model import ( "fmt" "strings" + "time" ) // DeviceClass is either a routed-core member or a zone-boundary firewall. @@ -75,6 +76,11 @@ type AddressGroup struct { Members []string `json:"members"` Refresh string `json:"refresh,omitempty"` Description string `json:"description,omitempty"` + + // Resolved holds concrete CIDRs the ASN expander last produced (asn groups + // only); ResolvedAt timestamps that expansion. Both are server-managed. + Resolved []string `json:"resolved,omitempty"` + ResolvedAt *time.Time `json:"resolved_at,omitempty"` } // SetName returns the nftables set name for this group. ASN groups get the diff --git a/internal/store/store.go b/internal/store/store.go index e8bb6f0..8aaedbf 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -158,7 +158,8 @@ func (s *Store) UpsertZone(ctx context.Context, z model.Zone) error { func (s *Store) ListAddressGroups(ctx context.Context) ([]model.AddressGroup, error) { rows, err := s.pool.Query(ctx, - `SELECT name, type, members, refresh, description FROM address_groups ORDER BY name`) + `SELECT name, type, members, refresh, description, resolved, resolved_at + FROM address_groups ORDER BY name`) if err != nil { return nil, err } @@ -166,18 +167,43 @@ func (s *Store) ListAddressGroups(ctx context.Context) ([]model.AddressGroup, er var out []model.AddressGroup for rows.Next() { var g model.AddressGroup - var members []byte - if err := rows.Scan(&g.Name, &g.Type, &members, &g.Refresh, &g.Description); err != nil { + var members, resolved []byte + if err := rows.Scan(&g.Name, &g.Type, &members, &g.Refresh, &g.Description, &resolved, &g.ResolvedAt); err != nil { return nil, err } if err := json.Unmarshal(members, &g.Members); err != nil { return nil, err } + if err := json.Unmarshal(resolved, &g.Resolved); err != nil { + return nil, err + } out = append(out, g) } return out, rows.Err() } +// UpdateResolvedPrefixes stores expanded ASN prefixes for a group and bumps the +// generation so agents re-pull. Membership churn is stored separately from the +// group definition, so an UpsertAddressGroup never clobbers it. +func (s *Store) UpdateResolvedPrefixes(ctx context.Context, name string, prefixes []string) error { + resolved, err := jsonb(prefixes) + if err != nil { + return err + } + return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error { + tag, err := tx.Exec(ctx, + `UPDATE address_groups SET resolved = $2, resolved_at = now() WHERE name = $1 AND type = 'asn'`, + name, resolved) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return bump(ctx, tx) + }) +} + func (s *Store) UpsertAddressGroup(ctx context.Context, g model.AddressGroup) error { members, err := jsonb(g.Members) if err != nil { -- 2.47.3