Add central ASN address-group expander #2

Merged
benvin merged 1 commits from benvin/asn-expander into main 2026-07-19 22:10:33 +10:00
8 changed files with 355 additions and 4 deletions
+15
View File
@@ -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,
+126
View File
@@ -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
}
+76
View File
@@ -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")
}
}
+94
View File
@@ -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
}
+2 -1
View File
@@ -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
}
@@ -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;
+6
View File
@@ -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
+29 -3
View File
@@ -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 {