Add central ASN address-group expander #2
@@ -13,9 +13,11 @@ import (
|
|||||||
"os/signal"
|
"os/signal"
|
||||||
"syscall"
|
"syscall"
|
||||||
|
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/asnexpand"
|
||||||
"git.unkin.net/unkin/tomswallapi/internal/config"
|
"git.unkin.net/unkin/tomswallapi/internal/config"
|
||||||
"git.unkin.net/unkin/tomswallapi/internal/database"
|
"git.unkin.net/unkin/tomswallapi/internal/database"
|
||||||
"git.unkin.net/unkin/tomswallapi/internal/server"
|
"git.unkin.net/unkin/tomswallapi/internal/server"
|
||||||
|
"git.unkin.net/unkin/tomswallapi/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
var version = "dev"
|
var version = "dev"
|
||||||
@@ -52,6 +54,19 @@ func main() {
|
|||||||
slog.Warn("TOMSWALLAPI_AGENT_TOKEN is not set; agent config endpoint is disabled")
|
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{
|
srv := server.New(server.Options{
|
||||||
DB: db,
|
DB: db,
|
||||||
WriteToken: cfg.WriteToken,
|
WriteToken: cfg.WriteToken,
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -243,7 +243,8 @@ func renderSet(g model.AddressGroup) RenderedSet {
|
|||||||
case model.GroupDNS:
|
case model.GroupDNS:
|
||||||
rs.FQDNs = g.Members
|
rs.FQDNs = g.Members
|
||||||
case model.GroupASN:
|
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
|
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;
|
||||||
@@ -5,6 +5,7 @@ package model
|
|||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// DeviceClass is either a routed-core member or a zone-boundary firewall.
|
// DeviceClass is either a routed-core member or a zone-boundary firewall.
|
||||||
@@ -75,6 +76,11 @@ type AddressGroup struct {
|
|||||||
Members []string `json:"members"`
|
Members []string `json:"members"`
|
||||||
Refresh string `json:"refresh,omitempty"`
|
Refresh string `json:"refresh,omitempty"`
|
||||||
Description string `json:"description,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
|
// SetName returns the nftables set name for this group. ASN groups get the
|
||||||
|
|||||||
+29
-3
@@ -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) {
|
func (s *Store) ListAddressGroups(ctx context.Context) ([]model.AddressGroup, error) {
|
||||||
rows, err := s.pool.Query(ctx,
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -166,18 +167,43 @@ func (s *Store) ListAddressGroups(ctx context.Context) ([]model.AddressGroup, er
|
|||||||
var out []model.AddressGroup
|
var out []model.AddressGroup
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var g model.AddressGroup
|
var g model.AddressGroup
|
||||||
var members []byte
|
var members, resolved []byte
|
||||||
if err := rows.Scan(&g.Name, &g.Type, &members, &g.Refresh, &g.Description); err != nil {
|
if err := rows.Scan(&g.Name, &g.Type, &members, &g.Refresh, &g.Description, &resolved, &g.ResolvedAt); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if err := json.Unmarshal(members, &g.Members); err != nil {
|
if err := json.Unmarshal(members, &g.Members); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
if err := json.Unmarshal(resolved, &g.Resolved); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
out = append(out, g)
|
out = append(out, g)
|
||||||
}
|
}
|
||||||
return out, rows.Err()
|
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 {
|
func (s *Store) UpsertAddressGroup(ctx context.Context, g model.AddressGroup) error {
|
||||||
members, err := jsonb(g.Members)
|
members, err := jsonb(g.Members)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user