feat: wire ExternalIPAM to NATS DHCP + move AF_PACKET to vpcd (mulga-991)

Replaces the sudo-dhcpcd shell-out in ObtainDHCPLease / ReleaseDHCPLease
with NATS request/reply against vpc.dhcp.acquire / vpc.dhcp.release.
The daemon-side IPAM passes its nats.Conn into NewExternalIPAM and
threads hostname (option 12) / vendor-class (option 60) through to
vpcd; ExternalIPAllocation now records LeaseExpiresUnix, DHCPServerID
and HWAddr on DHCP-sourced entries. allocateFromPool is restructured
to acquire the lease once up-front and rely on vpcd's idempotent
Manager for CAS-retry safety; a single release fires only if every
CAS attempt fails. Drops the stdout parser, the per-bridge mutex, and
the parse test.

Systemd: spinifex-daemon loses AF_PACKET, ProtectKernelTunables and
MemoryDenyWriteExecute are restored to =yes; spinifex-vpcd gains
AF_PACKET in RestrictAddressFamilies, AmbientCapabilities=CAP_NET_RAW,
and CAP_NET_ADMIN / CAP_NET_RAW / CAP_NET_BIND_SERVICE in the bounding
set. Net privilege footprint narrows.

See docs/development/bugs/dhcp-allocate-error.md.
This commit is contained in:
Ben Duncan
2026-04-23 12:21:38 +10:00
parent 265556dc4e
commit d7c95bda5e
10 changed files with 664 additions and 192 deletions
+6 -6
View File
@@ -17,7 +17,7 @@ Environment=SPINIFEX_CONFIG_PATH=/etc/spinifex/spinifex.toml
Environment=SPINIFEX_BASE_DIR=/var/lib/spinifex/spinifex/
Environment=SPINIFEX_WAL_DIR=/var/lib/spinifex/spinifex/
# Filesystem access — NoNewPrivileges omitted (daemon uses sudo for ip, ovs-vsctl, dhcpcd)
# Filesystem access — NoNewPrivileges omitted (daemon uses sudo for ip, ovs-vsctl)
ProtectSystem=full
PrivateDevices=no
DevicePolicy=closed
@@ -34,16 +34,16 @@ LockPersonality=yes
RestrictRealtime=yes
RestrictNamespaces=yes
ProtectClock=yes
# Required for dhcpcd (issue dhcpcd[867]: br-wan: if_init: Read-only file system)
ProtectKernelTunables=no
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectKernelLogs=yes
ProtectControlGroups=yes
ProtectProc=invisible
SystemCallArchitectures=native
MemoryDenyWriteExecute=no
# AF_PACKET is required by dhcpcd (raw L2 BOOTP before the iface has an IP).
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK AF_PACKET
MemoryDenyWriteExecute=yes
# DHCP lease acquisition moved to spinifex-vpcd (see dhcp-allocate-error.md);
# the daemon no longer needs AF_PACKET or the softened kernel protections.
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK
ProtectHostname=yes
KeyringMode=private
RemoveIPC=yes
+9 -1
View File
@@ -37,7 +37,15 @@ ProtectControlGroups=yes
ProtectProc=invisible
SystemCallArchitectures=native
MemoryDenyWriteExecute=yes
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK
# AF_PACKET: required by the in-process DHCP client (github.com/insomniacslk/dhcp
# / nclient4) for raw L2 BOOTP on br-wan before an IP is configured on the
# interface. See docs/development/bugs/dhcp-allocate-error.md.
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX AF_NETLINK AF_PACKET
# CAP_NET_RAW: open AF_PACKET sockets.
# CAP_NET_BIND_SERVICE: covers unicast RENEW to UDP :68.
# CAP_NET_ADMIN: already used for ovs-vsctl / ip link wiring on the bridge.
AmbientCapabilities=CAP_NET_RAW
CapabilityBoundingSet=CAP_NET_ADMIN CAP_NET_RAW CAP_NET_BIND_SERVICE
ProtectHostname=yes
KeyringMode=private
RemoveIPC=yes
+1 -1
View File
@@ -667,7 +667,7 @@ func (d *Daemon) Start() error {
WanBridge: wanBridge,
})
}
d.externalIPAM, err = handlers_ec2_vpc.NewExternalIPAM(js, pools)
d.externalIPAM, err = handlers_ec2_vpc.NewExternalIPAM(d.natsConn, js, pools)
if err != nil {
slog.Warn("Failed to initialize external IPAM", "err", err)
} else {
+1 -1
View File
@@ -3898,7 +3898,7 @@ func TestDelegateHandlers_EIP(t *testing.T) {
js, err := nc.JetStream()
require.NoError(t, err)
ipam, err := handlers_ec2_vpc.NewExternalIPAM(js, []handlers_ec2_vpc.ExternalPoolConfig{
ipam, err := handlers_ec2_vpc.NewExternalIPAM(nc, js, []handlers_ec2_vpc.ExternalPoolConfig{
{Name: "test-pool", RangeStart: "192.168.100.2", RangeEnd: "192.168.100.254", Gateway: "192.168.100.1", PrefixLen: 24},
})
require.NoError(t, err)
@@ -29,7 +29,7 @@ func setupTestEIP(t *testing.T) (*EIPServiceImpl, *handlers_ec2_vpc.ExternalIPAM
_, nc, js := testutil.StartTestJetStream(t)
pool := testPool()
ipam, err := handlers_ec2_vpc.NewExternalIPAM(js, []handlers_ec2_vpc.ExternalPoolConfig{pool})
ipam, err := handlers_ec2_vpc.NewExternalIPAM(nc, js, []handlers_ec2_vpc.ExternalPoolConfig{pool})
require.NoError(t, err)
svc, err := NewEIPServiceImpl(nc, ipam, nil)
+108 -23
View File
@@ -26,6 +26,14 @@ type ExternalIPAllocation struct {
ENIId string `json:"eni_id,omitempty"` // ENI owning this IP
InstanceId string `json:"instance_id,omitempty"` // Instance owning this IP
Note string `json:"note,omitempty"` // Human-readable note
// DHCP-sourced metadata — set only for allocations whose pool has
// source="dhcp" (see ObtainDHCPLease). These fields let describe
// surfaces and operators trace a given IP back to the upstream lease
// without re-querying vpcd.
LeaseExpiresUnix int64 `json:"lease_expires_unix,omitempty"` // Absolute expiry from the server's ACK
DHCPServerID string `json:"dhcp_server_id,omitempty"` // Option 54 of the ACK
HWAddr string `json:"hw_addr,omitempty"` // chaddr placed in the DHCP payload
}
// ExternalIPAMRecord tracks allocated external IPs for a single pool.
@@ -64,10 +72,16 @@ func (p *ExternalPoolConfig) IsDHCP() bool {
type ExternalIPAM struct {
kv nats.KeyValue
pools []ExternalPoolConfig
// nc is used to talk to spinifex-vpcd for DHCP-sourced pools via
// vpc.dhcp.acquire / vpc.dhcp.release. May be nil when no pool has
// Source="dhcp" (callers should still pass it if available).
nc *nats.Conn
}
// NewExternalIPAM creates a new ExternalIPAM backed by NATS JetStream KV.
func NewExternalIPAM(js nats.JetStreamContext, pools []ExternalPoolConfig) (*ExternalIPAM, error) {
// nc is required for DHCP-sourced pools and may be nil for static-only
// deployments.
func NewExternalIPAM(nc *nats.Conn, js nats.JetStreamContext, pools []ExternalPoolConfig) (*ExternalIPAM, error) {
kv, err := utils.GetOrCreateKVBucket(js, KVBucketExternalIPAM, 5)
if err != nil {
return nil, fmt.Errorf("create external IPAM KV bucket: %w", err)
@@ -75,16 +89,17 @@ func NewExternalIPAM(js nats.JetStreamContext, pools []ExternalPoolConfig) (*Ext
if err := migrate.DefaultRegistry.RunKV(KVBucketExternalIPAM, kv, KVBucketExternalIPAMVersion); err != nil {
return nil, fmt.Errorf("migrate %s: %w", KVBucketExternalIPAM, err)
}
ipam := &ExternalIPAM{kv: kv, pools: pools}
ipam := &ExternalIPAM{kv: kv, pools: pools, nc: nc}
if err := ipam.initPools(); err != nil {
return nil, fmt.Errorf("init external IPAM pools: %w", err)
}
return ipam, nil
}
// NewExternalIPAMWithKV creates an ExternalIPAM with an existing KV bucket (for testing).
func NewExternalIPAMWithKV(kv nats.KeyValue, pools []ExternalPoolConfig) *ExternalIPAM {
return &ExternalIPAM{kv: kv, pools: pools}
// NewExternalIPAMWithKV creates an ExternalIPAM with an existing KV bucket
// (for testing). nc may be nil when the test exercises only static pools.
func NewExternalIPAMWithKV(nc *nats.Conn, kv nats.KeyValue, pools []ExternalPoolConfig) *ExternalIPAM {
return &ExternalIPAM{kv: kv, pools: pools, nc: nc}
}
// initPools ensures each configured pool has a KV record. Idempotent — safe to call
@@ -113,16 +128,32 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error {
gwIP = pool.RangeStart
}
// For DHCP pools, obtain the gateway IP from router DHCP if not set
// For DHCP pools, obtain the gateway IP from router DHCP if not set.
// The gateway lease is long-lived: vpcd's renewal goroutine keeps it
// refreshed until the pool is torn down.
var gatewayLease *dhcpLeaseResult
if gwIP == "" && pool.IsDHCP() {
dhcpIP, dhcpErr := ObtainDHCPLease(pool.WanBridge, "gateway-"+pool.Name)
lease, dhcpErr := ObtainDHCPLease(m.nc, pool.WanBridge,
"gateway-"+pool.Name,
"gateway-"+pool.Name,
"mulga-spinifex-gw",
pool.Name,
)
if dhcpErr != nil {
return fmt.Errorf("DHCP gateway lease for pool %q: %w", pool.Name, dhcpErr)
}
gwIP = dhcpIP
gatewayLease = &lease
gwIP = lease.IP
slog.Info("external IPAM obtained gateway IP via DHCP", "pool", pool.Name, "gateway_ip", gwIP)
}
gatewayAlloc := ExternalIPAllocation{Type: "gateway", Note: "OVN router SNAT address"}
if gatewayLease != nil {
gatewayAlloc.LeaseExpiresUnix = gatewayLease.ExpiresUnix
gatewayAlloc.DHCPServerID = gatewayLease.ServerID
gatewayAlloc.HWAddr = gatewayLease.HWAddr
}
record := &ExternalIPAMRecord{
PoolName: pool.Name,
RangeStart: pool.RangeStart,
@@ -133,7 +164,7 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error {
Region: pool.Region,
AZ: pool.AZ,
Allocated: map[string]ExternalIPAllocation{
gwIP: {Type: "gateway", Note: "OVN router SNAT address"},
gwIP: gatewayAlloc,
},
}
@@ -188,18 +219,32 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins
clientID = instanceID
}
// For DHCP-sourced pools: acquire the lease once, up front. vpcd's
// Manager is idempotent on ClientID, so CAS retries below will not
// trigger a second DORA — we only release at the end if no CAS
// attempt succeeded.
var dhcpLease *dhcpLeaseResult
if pool != nil && pool.IsDHCP() {
hostname, vendorClass := dhcpIdentityOptions(eniID, instanceID, poolName)
lease, err := ObtainDHCPLease(m.nc, pool.WanBridge, clientID, hostname, vendorClass, poolName)
if err != nil {
return "", fmt.Errorf("DHCP lease for %s: %w", clientID, err)
}
dhcpLease = &lease
}
for attempt := range 5 {
record, revision, err := m.getRecord(poolName)
if err != nil {
if dhcpLease != nil {
_ = ReleaseDHCPLease(m.nc, clientID)
}
return "", fmt.Errorf("get external IPAM record: %w", err)
}
var ip string
if pool != nil && pool.IsDHCP() {
ip, err = ObtainDHCPLease(pool.WanBridge, clientID)
if err != nil {
return "", fmt.Errorf("DHCP lease for %s: %w", clientID, err)
}
if dhcpLease != nil {
ip = dhcpLease.IP
} else {
// Static source: pick next IP from range
ip, err = nextAvailableExternalIP(record)
@@ -208,34 +253,69 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins
}
}
record.Allocated[ip] = ExternalIPAllocation{
alloc := ExternalIPAllocation{
Type: allocType,
AllocationID: allocID,
ENIId: eniID,
InstanceId: instanceID,
}
if dhcpLease != nil {
alloc.LeaseExpiresUnix = dhcpLease.ExpiresUnix
alloc.DHCPServerID = dhcpLease.ServerID
alloc.HWAddr = dhcpLease.HWAddr
}
record.Allocated[ip] = alloc
data, err := json.Marshal(record)
if err != nil {
if dhcpLease != nil {
_ = ReleaseDHCPLease(m.nc, clientID)
}
return "", fmt.Errorf("marshal external IPAM record: %w", err)
}
if _, err := m.kv.Update(poolName, data, revision); err != nil {
// CAS conflict — if we obtained a DHCP lease, release it before retrying
if pool != nil && pool.IsDHCP() {
_ = ReleaseDHCPLease(pool.WanBridge, clientID)
}
slog.Debug("external IPAM CAS conflict, retrying", "pool", poolName, "attempt", attempt)
continue
}
slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "type", allocType, "source", pool.Source)
source := "static"
if pool != nil {
source = pool.Source
}
slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "type", allocType, "source", source)
return ip, nil
}
// All CAS attempts exhausted — release any DHCP lease we acquired so
// vpcd doesn't hold a binding for an IP we won't use.
if dhcpLease != nil {
_ = ReleaseDHCPLease(m.nc, clientID)
}
return "", fmt.Errorf("external IPAM allocation failed after CAS retries for pool %s", poolName)
}
// dhcpIdentityOptions composes the option 12 (hostname) and option 60
// (vendor class) values for a per-allocation lease. Empty eniID /
// instanceID fall through to identifiers that still group usefully on
// the upstream server's lease table. Instance and ENI identifiers are
// used as-is (they already carry their "i-" / "eni-" prefixes).
func dhcpIdentityOptions(eniID, instanceID, poolName string) (hostname, vendorClass string) {
hostname = eniID
if hostname == "" {
if instanceID != "" {
hostname = "spinifex-" + instanceID
} else {
hostname = "spinifex-" + poolName
}
}
vendorClass = "mulga-spinifex"
if instanceID != "" {
vendorClass = instanceID
}
return hostname, vendorClass
}
// ReleaseIP releases a previously allocated external IP back to its pool.
func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error {
pool := m.findPoolByName(poolName)
@@ -254,13 +334,18 @@ func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error {
return fmt.Errorf("cannot release gateway IP %s in pool %s", ip, poolName)
}
// Release DHCP lease if this is a DHCP-sourced pool
// Release DHCP lease if this is a DHCP-sourced pool. clientID
// selection mirrors allocateFromPool so vpcd's Manager can find the
// lease it originally acquired.
if pool != nil && pool.IsDHCP() {
clientID := alloc.ENIId
clientID := alloc.AllocationID
if clientID == "" {
clientID = alloc.ENIId
}
if clientID == "" {
clientID = alloc.InstanceId
}
if releaseErr := ReleaseDHCPLease(pool.WanBridge, clientID); releaseErr != nil {
if releaseErr := ReleaseDHCPLease(m.nc, clientID); releaseErr != nil {
slog.Warn("Failed to release DHCP lease", "pool", poolName, "ip", ip, "err", releaseErr)
}
}
+94 -97
View File
@@ -1,127 +1,124 @@
package handlers_ec2_vpc
import (
"encoding/json"
"fmt"
"log/slog"
"os/exec"
"strings"
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp"
"github.com/nats-io/nats.go"
)
const dhcpcdBin = "/usr/sbin/dhcpcd"
// dhcpNATSTimeout bounds the request/reply to spinifex-vpcd. Picked to
// comfortably exceed the DORA handshake (15 s default + server jitter +
// ~10 s of slack for NAK + fresh DISCOVER fallbacks). A var rather than
// a const so tests can drive the timeout error path without waiting
// 20 s per table entry.
var dhcpNATSTimeout = 20 * time.Second
// bridgeMu serialises dhcpcd invocations per bridge. Concurrent invocations on
// the same bridge race: the second one sees the first still running and is
// treated by dhcpcd as a control command rather than a fresh DISCOVER, so we
// scrape stdout and find no lease. Serialising per bridge costs us nothing
// (DHCP is inherently slow) and eliminates the race.
var (
bridgeMuMap = map[string]*sync.Mutex{}
bridgeMuGuard sync.Mutex
)
func bridgeLock(bridge string) *sync.Mutex {
bridgeMuGuard.Lock()
defer bridgeMuGuard.Unlock()
mu, ok := bridgeMuMap[bridge]
if !ok {
mu = &sync.Mutex{}
bridgeMuMap[bridge] = mu
}
return mu
// dhcpLeaseResult is the data ExternalIPAM needs from a successful acquire.
// It maps directly from dhcp.AcquireReplyMsg; kept as a handler-side type
// so the IPAM ledger records the fields without coupling to the NATS wire
// struct.
type dhcpLeaseResult struct {
IP string
SubnetMask string
Routers []string
DNS []string
ServerID string
HWAddr string
ExpiresUnix int64
}
// ObtainDHCPLease requests a DHCP lease from the router on the given OVS bridge
// using a unique client ID. Returns the leased IP. The lease is obtained with
// --noconfigure so the IP is NOT added to the bridge interface — OVN handles
// traffic via its NAT rules.
func ObtainDHCPLease(bridge, clientID string) (string, error) {
// ObtainDHCPLease asks spinifex-vpcd to acquire a DHCP lease on the given
// bridge, identifying us with option 61 (client-id), option 12 (hostname)
// and option 60 (vendor class). Blocks until vpcd replies or
// dhcpNATSTimeout expires. The Manager-side handler is idempotent: a
// second call with the same clientID while a live lease exists returns
// the same lease without a fresh DORA, so CAS retry loops on the caller
// are safe.
func ObtainDHCPLease(nc *nats.Conn, bridge, clientID, hostname, vendorClass, poolName string) (dhcpLeaseResult, error) {
if nc == nil {
return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: NATS connection is required")
}
if bridge == "" {
return "", fmt.Errorf("DHCP lease: bridge name is required")
return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: bridge name is required")
}
if clientID == "" {
return "", fmt.Errorf("DHCP lease: client ID is required")
return dhcpLeaseResult{}, fmt.Errorf("DHCP lease: client ID is required")
}
mu := bridgeLock(bridge)
mu.Lock()
defer mu.Unlock()
req := dhcp.AcquireRequestMsg{
Bridge: bridge,
ClientID: clientID,
Hostname: hostname,
VendorClass: vendorClass,
PoolName: poolName,
}
data, err := json.Marshal(req)
if err != nil {
return dhcpLeaseResult{}, fmt.Errorf("marshal dhcp acquire request: %w", err)
}
// Run dhcpcd with:
// --noconfigure = don't add IP to interface (OVN handles traffic)
// -1 = exit after obtaining a lease (don't daemonize)
// -4 = IPv4 only
// -T = TEST MODE
// -I clientID = unique client identifier per ENI (Required to generate different IP from the DHCP server)
// -t 15 = 15 second timeout
cmd := exec.Command("sudo", dhcpcdBin,
"--noconfigure",
"-1",
"-4",
msg, err := nc.Request(dhcp.TopicAcquire, data, dhcpNATSTimeout)
if err != nil {
return dhcpLeaseResult{}, fmt.Errorf("dhcp acquire NATS request (client %s): %w", clientID, err)
}
// Previous bug, if dhcpcd already running (e.g dhcp mode set on an interface, will send cmd to daemon, not STDOUT)
"-T",
var reply dhcp.AcquireReplyMsg
if err := json.Unmarshal(msg.Data, &reply); err != nil {
return dhcpLeaseResult{}, fmt.Errorf("unmarshal dhcp acquire reply: %w", err)
}
if reply.Error != "" {
return dhcpLeaseResult{}, fmt.Errorf("dhcp acquire (client %s): %s", clientID, reply.Error)
}
"-I", clientID,
"-t", "15",
bridge,
slog.Info("DHCP lease obtained",
"bridge", bridge,
"client_id", clientID,
"ip", reply.IP,
"server_id", reply.ServerID,
"expires_unix", reply.ExpiresUnix,
)
output, err := cmd.CombinedOutput()
slog.Debug("dhcpcd output", "bridge", bridge, "clientID", clientID, "output", string(output), "err", err)
// Parse the leased IP from dhcpcd output.
// dhcpcd prints: "br-ext: leased 192.168.1.75 for 1800 seconds"
// Note: dhcpcd may exit non-zero with --noconfigure even after obtaining
// a lease (it reports "timed out" waiting for interface configuration).
// We check for a lease in the output regardless of exit code.
ip := parseDHCPCDLeasedIP(string(output))
if ip == "" {
if err != nil {
return "", fmt.Errorf("dhcpcd failed on %s (client %s): %w\noutput: %s", bridge, clientID, err, string(output))
}
return "", fmt.Errorf("dhcpcd produced no lease on %s (client %s): %s", bridge, clientID, string(output))
}
slog.Info("DHCP lease obtained", "bridge", bridge, "clientID", clientID, "ip", ip)
return ip, nil
return dhcpLeaseResult{
IP: reply.IP,
SubnetMask: reply.SubnetMask,
Routers: reply.Routers,
DNS: reply.DNS,
ServerID: reply.ServerID,
HWAddr: reply.HWAddr,
ExpiresUnix: reply.ExpiresUnix,
}, nil
}
// ReleaseDHCPLease releases a previously obtained DHCP lease.
func ReleaseDHCPLease(bridge, clientID string) error {
if bridge == "" || clientID == "" {
// ReleaseDHCPLease asks spinifex-vpcd to release the lease identified by
// clientID. Returns nil (silently) when nc or clientID is empty so that
// callers in clean-up paths can invoke it unconditionally.
func ReleaseDHCPLease(nc *nats.Conn, clientID string) error {
if nc == nil || clientID == "" {
return nil
}
mu := bridgeLock(bridge)
mu.Lock()
defer mu.Unlock()
cmd := exec.Command("sudo", dhcpcdBin,
"--release",
"-4",
"-I", clientID,
bridge,
)
output, err := cmd.CombinedOutput()
data, err := json.Marshal(dhcp.ReleaseRequestMsg{ClientID: clientID})
if err != nil {
slog.Debug("dhcpcd release", "bridge", bridge, "clientID", clientID, "output", string(output), "err", err)
// Not fatal — lease will expire naturally
return fmt.Errorf("dhcpcd release failed on %s (client %s): %w", bridge, clientID, err)
return fmt.Errorf("marshal dhcp release request: %w", err)
}
slog.Info("DHCP lease released", "bridge", bridge, "clientID", clientID)
msg, err := nc.Request(dhcp.TopicRelease, data, dhcpNATSTimeout)
if err != nil {
return fmt.Errorf("dhcp release NATS request (client %s): %w", clientID, err)
}
var reply dhcp.ReleaseReplyMsg
if err := json.Unmarshal(msg.Data, &reply); err != nil {
return fmt.Errorf("unmarshal dhcp release reply: %w", err)
}
if reply.Error != "" {
return fmt.Errorf("dhcp release (client %s): %s", clientID, reply.Error)
}
slog.Info("DHCP lease released", "client_id", clientID)
return nil
}
// parseDHCPCDLeasedIP extracts the leased IP from dhcpcd stdout.
// Looks for: "<iface>: leased <IP> for <N> seconds"
func parseDHCPCDLeasedIP(output string) string {
for line := range strings.SplitSeq(output, "\n") {
if before, after, found := strings.Cut(line, ": leased "); found && before != "" {
if spaceIdx := strings.IndexByte(after, ' '); spaceIdx > 0 {
return after[:spaceIdx]
}
}
}
return ""
}
+325 -62
View File
@@ -1,19 +1,24 @@
package handlers_ec2_vpc
import (
"encoding/json"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func setupTestExternalIPAM(t *testing.T, pools []ExternalPoolConfig) *ExternalIPAM {
t.Helper()
_, _, js := testutil.StartTestJetStream(t)
_, nc, js := testutil.StartTestJetStream(t)
ipam, err := NewExternalIPAM(js, pools)
ipam, err := NewExternalIPAM(nc, js, pools)
require.NoError(t, err)
return ipam
}
@@ -334,10 +339,10 @@ func TestExternalIPAM_RangeValidation(t *testing.T) {
func TestExternalIPAM_InitFromConfig(t *testing.T) {
pool := testPool()
// Create IPAM twice — second init should be idempotent
_, _, js := testutil.StartTestJetStream(t)
_, nc, js := testutil.StartTestJetStream(t)
// First init
ipam1, err := NewExternalIPAM(js, []ExternalPoolConfig{pool})
ipam1, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
// Allocate an IP
@@ -346,7 +351,7 @@ func TestExternalIPAM_InitFromConfig(t *testing.T) {
assert.Equal(t, "192.168.1.151", ip)
// Second init (simulating restart) — should not lose allocation
ipam2, err := NewExternalIPAM(js, []ExternalPoolConfig{pool})
ipam2, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
record, err := ipam2.GetPoolRecord("wan")
@@ -388,7 +393,7 @@ func TestExternalIPAM_FindPoolByName_NotFound(t *testing.T) {
// The function returns nil when no pool matches, which means static allocation path.
// We verify by using NewExternalIPAMWithKV to directly check findPoolByName returns nil.
kv := ipam.kv
ipam2 := NewExternalIPAMWithKV(kv, []ExternalPoolConfig{pool})
ipam2 := NewExternalIPAMWithKV(nil, kv, []ExternalPoolConfig{pool})
// AllocateFromPool with a non-existent pool name: findPoolByName returns nil,
// so pool.IsDHCP() is skipped and static allocation is used.
// The pool "nonexistent" has no KV record, so getRecord fails.
@@ -396,62 +401,6 @@ func TestExternalIPAM_FindPoolByName_NotFound(t *testing.T) {
assert.Error(t, err)
}
func TestParseDHCPCDLeasedIP(t *testing.T) {
tests := []struct {
name string
output string
want string
}{
{
name: "valid lease line",
output: "br-ext: leased 192.168.1.75 for 1800 seconds",
want: "192.168.1.75",
},
{
name: "lease in middle of multiline output",
output: "br-ext: soliciting a DHCP lease\nbr-ext: offered 192.168.1.75 from 192.168.1.1\nbr-ext: leased 192.168.1.75 for 1800 seconds\nbr-ext: adding route",
want: "192.168.1.75",
},
{
name: "no lease line",
output: "br-ext: soliciting a DHCP lease\nbr-ext: timed out",
want: "",
},
{
name: "empty output",
output: "",
want: "",
},
{
name: "malformed line - no space after IP",
output: ": leased 192.168.1.75",
want: "",
},
{
name: "empty before colon",
output: ": leased 192.168.1.75 for 600 seconds",
want: "",
},
{
name: "lease keyword but no IP after",
output: "br-ext: leased ",
want: "",
},
{
name: "different bridge name",
output: "br-wan: leased 10.0.0.50 for 3600 seconds",
want: "10.0.0.50",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := parseDHCPCDLeasedIP(tt.output)
assert.Equal(t, tt.want, got)
})
}
}
func TestExternalPoolConfig_IsDHCP(t *testing.T) {
p := ExternalPoolConfig{Source: "dhcp"}
assert.True(t, p.IsDHCP())
@@ -473,3 +422,317 @@ func TestValidatePoolConfig_DHCPPool(t *testing.T) {
err := ValidatePoolConfig(pool)
assert.NoError(t, err)
}
// dhcpStub captures acquire/release requests and replies with a
// deterministic lease so the handler-side IPAM code can be exercised
// without running vpcd.
type dhcpStub struct {
mu sync.Mutex
acquired []dhcp.AcquireRequestMsg
released []dhcp.ReleaseRequestMsg
nextIP string
acquireErr string
acquireCalls atomic.Int32
releaseCalls atomic.Int32
serverID string
leaseSeconds int64
hwAddrOverride string
releaseErrOnce atomic.Bool
}
func newDHCPStub(t *testing.T, nc *nats.Conn) *dhcpStub {
t.Helper()
s := &dhcpStub{
nextIP: "192.168.1.151",
serverID: "192.168.1.1",
leaseSeconds: 3600,
}
acquire, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) {
s.acquireCalls.Add(1)
var req dhcp.AcquireRequestMsg
if err := json.Unmarshal(msg.Data, &req); err != nil {
respondTestJSON(t, msg, dhcp.AcquireReplyMsg{Error: "malformed request"})
return
}
s.mu.Lock()
s.acquired = append(s.acquired, req)
ip := s.nextIP
errMsg := s.acquireErr
serverID := s.serverID
expires := time.Now().Add(time.Duration(s.leaseSeconds) * time.Second).Unix()
hw := s.hwAddrOverride
s.mu.Unlock()
if errMsg != "" {
respondTestJSON(t, msg, dhcp.AcquireReplyMsg{Error: errMsg})
return
}
if hw == "" {
hw = "02:00:00:aa:bb:cc"
}
respondTestJSON(t, msg, dhcp.AcquireReplyMsg{
IP: ip,
SubnetMask: "255.255.255.0",
Routers: []string{"192.168.1.1"},
DNS: []string{"192.168.1.1"},
ServerID: serverID,
HWAddr: hw,
ExpiresUnix: expires,
})
})
require.NoError(t, err)
t.Cleanup(func() { _ = acquire.Unsubscribe() })
release, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) {
s.releaseCalls.Add(1)
var req dhcp.ReleaseRequestMsg
_ = json.Unmarshal(msg.Data, &req)
s.mu.Lock()
s.released = append(s.released, req)
s.mu.Unlock()
reply := dhcp.ReleaseReplyMsg{}
if s.releaseErrOnce.CompareAndSwap(true, false) {
reply.Error = "first release fails"
}
respondTestJSON(t, msg, reply)
})
require.NoError(t, err)
t.Cleanup(func() { _ = release.Unsubscribe() })
return s
}
func respondTestJSON(t *testing.T, msg *nats.Msg, v any) {
t.Helper()
data, err := json.Marshal(v)
require.NoError(t, err)
require.NoError(t, msg.Respond(data))
}
func TestExternalIPAM_DHCPAllocateAndRelease(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
stub := newDHCPStub(t, nc)
stub.nextIP = "192.168.1.200"
pool := ExternalPoolConfig{
Name: "wan-dhcp",
Source: "dhcp",
Gateway: "192.168.1.1",
GatewayIP: "192.168.1.1", // pre-set so initPool skips gateway DORA
PrefixLen: 24,
WanBridge: "br-wan",
}
ipam, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
ip, err := ipam.AllocateFromPool("wan-dhcp", "elastic_ip", "eipalloc-123", "eni-abc", "i-xyz")
require.NoError(t, err)
assert.Equal(t, "192.168.1.200", ip)
require.EqualValues(t, 1, stub.acquireCalls.Load())
stub.mu.Lock()
got := stub.acquired[0]
stub.mu.Unlock()
assert.Equal(t, "eipalloc-123", got.ClientID, "clientID precedence: allocID first")
assert.Equal(t, "eni-abc", got.Hostname, "hostname defaults to eniID")
assert.Equal(t, "i-xyz", got.VendorClass, "vendor class i-<instanceID>")
assert.Equal(t, "wan-dhcp", got.PoolName)
assert.Equal(t, "br-wan", got.Bridge)
record, err := ipam.GetPoolRecord("wan-dhcp")
require.NoError(t, err)
alloc, ok := record.Allocated["192.168.1.200"]
require.True(t, ok)
assert.Equal(t, "192.168.1.1", alloc.DHCPServerID)
assert.NotZero(t, alloc.LeaseExpiresUnix)
assert.Equal(t, "02:00:00:aa:bb:cc", alloc.HWAddr)
require.NoError(t, ipam.ReleaseIP("wan-dhcp", "192.168.1.200"))
require.EqualValues(t, 1, stub.releaseCalls.Load())
stub.mu.Lock()
assert.Equal(t, "eipalloc-123", stub.released[0].ClientID)
stub.mu.Unlock()
}
func TestExternalIPAM_DHCPAcquireErrorPropagates(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
stub := newDHCPStub(t, nc)
stub.acquireErr = "server unreachable"
pool := ExternalPoolConfig{
Name: "wan-dhcp-err", Source: "dhcp", Gateway: "192.168.1.1",
GatewayIP: "192.168.1.1", PrefixLen: 24, WanBridge: "br-wan",
}
ipam, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
_, err = ipam.AllocateFromPool("wan-dhcp-err", "elastic_ip", "eipalloc-err", "", "")
require.Error(t, err)
assert.Contains(t, err.Error(), "server unreachable")
}
func TestExternalIPAM_DHCPGatewayFromLease(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
stub := newDHCPStub(t, nc)
stub.nextIP = "192.168.3.247"
pool := ExternalPoolConfig{
Name: "wan-dhcp-gw", Source: "dhcp", Gateway: "192.168.3.1",
PrefixLen: 24, WanBridge: "br-wan",
// GatewayIP intentionally empty — initPool must pull it from DHCP.
}
_, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
assert.EqualValues(t, 1, stub.acquireCalls.Load())
stub.mu.Lock()
got := stub.acquired[0]
stub.mu.Unlock()
assert.Equal(t, "gateway-wan-dhcp-gw", got.ClientID)
assert.Equal(t, "mulga-spinifex-gw", got.VendorClass)
}
func TestObtainDHCPLease_RequestTimeout(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
// Stub never responds — verifies that a hung vpcd surfaces as a
// wrapped NATS timeout error rather than hanging the caller.
sub, err := nc.Subscribe(dhcp.TopicAcquire, func(*nats.Msg) {})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
prev := dhcpNATSTimeout
dhcpNATSTimeout = 150 * time.Millisecond
t.Cleanup(func() { dhcpNATSTimeout = prev })
_, err = ObtainDHCPLease(nc, "br-wan", "eni-timeout", "eni-timeout", "mulga-spinifex", "wan")
assert.ErrorContains(t, err, "dhcp acquire NATS request")
}
func TestReleaseDHCPLease_RequestTimeout(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, err := nc.Subscribe(dhcp.TopicRelease, func(*nats.Msg) {})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
prev := dhcpNATSTimeout
dhcpNATSTimeout = 150 * time.Millisecond
t.Cleanup(func() { dhcpNATSTimeout = prev })
err = ReleaseDHCPLease(nc, "eni-timeout")
assert.ErrorContains(t, err, "dhcp release NATS request")
}
func TestObtainDHCPLease_MalformedReply(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, err := nc.Subscribe(dhcp.TopicAcquire, func(msg *nats.Msg) {
_ = msg.Respond([]byte("not json"))
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
_, err = ObtainDHCPLease(nc, "br-wan", "eni-malformed", "eni-malformed", "mulga-spinifex", "wan")
assert.ErrorContains(t, err, "unmarshal dhcp acquire reply")
}
func TestReleaseDHCPLease_MalformedReply(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) {
_ = msg.Respond([]byte("not json"))
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
err = ReleaseDHCPLease(nc, "eni-malformed")
assert.ErrorContains(t, err, "unmarshal dhcp release reply")
}
func TestReleaseDHCPLease_ReplyError(t *testing.T) {
_, nc := testutil.StartTestNATS(t)
sub, err := nc.Subscribe(dhcp.TopicRelease, func(msg *nats.Msg) {
data, _ := json.Marshal(dhcp.ReleaseReplyMsg{Error: "vpcd rejected release"})
_ = msg.Respond(data)
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
err = ReleaseDHCPLease(nc, "eni-rejected")
assert.ErrorContains(t, err, "vpcd rejected release")
}
func TestObtainDHCPLease_Guards(t *testing.T) {
_, err := ObtainDHCPLease(nil, "br-wan", "eni-1", "eni-1", "mulga-spinifex", "wan")
assert.ErrorContains(t, err, "NATS connection is required")
_, nc := testutil.StartTestNATS(t)
_, err = ObtainDHCPLease(nc, "", "eni-1", "eni-1", "mulga-spinifex", "wan")
assert.ErrorContains(t, err, "bridge name is required")
_, err = ObtainDHCPLease(nc, "br-wan", "", "eni-1", "mulga-spinifex", "wan")
assert.ErrorContains(t, err, "client ID is required")
}
func TestReleaseDHCPLease_NoopWhenNilOrEmpty(t *testing.T) {
assert.NoError(t, ReleaseDHCPLease(nil, "eni-1"))
_, nc := testutil.StartTestNATS(t)
assert.NoError(t, ReleaseDHCPLease(nc, ""))
}
func TestExternalIPAM_DHCPGatewayErrorPropagates(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
stub := newDHCPStub(t, nc)
stub.acquireErr = "no response from upstream"
pool := ExternalPoolConfig{
Name: "wan-gw-err", Source: "dhcp", Gateway: "192.168.1.1",
PrefixLen: 24, WanBridge: "br-wan",
}
_, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.Error(t, err)
assert.Contains(t, err.Error(), "no response from upstream")
}
func TestExternalIPAM_DHCPReleaseErrorIsNonFatal(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
stub := newDHCPStub(t, nc)
stub.nextIP = "192.168.1.155"
stub.releaseErrOnce.Store(true)
pool := ExternalPoolConfig{
Name: "wan-rel-err", Source: "dhcp", Gateway: "192.168.1.1",
GatewayIP: "192.168.1.1", PrefixLen: 24, WanBridge: "br-wan",
}
ipam, err := NewExternalIPAM(nc, js, []ExternalPoolConfig{pool})
require.NoError(t, err)
ip, err := ipam.AllocateFromPool("wan-rel-err", "elastic_ip", "eipalloc-rel", "", "")
require.NoError(t, err)
// Release reports vpcd failure via slog.Warn but ReleaseIP itself
// still succeeds — the allocation is removed from KV.
require.NoError(t, ipam.ReleaseIP("wan-rel-err", ip))
rec, err := ipam.GetPoolRecord("wan-rel-err")
require.NoError(t, err)
_, stillThere := rec.Allocated[ip]
assert.False(t, stillThere)
}
func TestDHCPIdentityOptions(t *testing.T) {
tests := []struct {
name string
eniID string
instanceID string
pool string
wantHost string
wantVendor string
}{
{"eni+instance", "eni-1", "i-abcd", "wan", "eni-1", "i-abcd"},
{"instance only", "", "i-xyz", "wan", "spinifex-i-xyz", "i-xyz"},
{"neither", "", "", "wan", "spinifex-wan", "mulga-spinifex"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
host, vendor := dhcpIdentityOptions(tc.eniID, tc.instanceID, tc.pool)
assert.Equal(t, tc.wantHost, host)
assert.Equal(t, tc.wantVendor, vendor)
})
}
}
@@ -111,3 +111,64 @@ func TestFakeRenewHookSurfacesError(t *testing.T) {
t.Fatal("expected hook error")
}
}
func TestFakeReleaseHookSurfacesError(t *testing.T) {
f := dhcp.NewFake()
f.ReleaseHook = func(*dhcp.Lease) error { return errors.New("server unreachable") }
lease, _ := f.Acquire(context.Background(), dhcp.AcquireRequest{Bridge: "br-wan", ClientID: "eni-rel"})
if err := f.Release(context.Background(), lease); err == nil {
t.Fatal("expected hook error")
}
}
func TestFakeRenewNilLease(t *testing.T) {
f := dhcp.NewFake()
if _, err := f.Renew(context.Background(), nil); err == nil {
t.Fatal("expected error for nil lease")
}
}
func TestFakeReleaseNilIsNoop(t *testing.T) {
f := dhcp.NewFake()
if err := f.Release(context.Background(), nil); err != nil {
t.Fatalf("expected nil, got %v", err)
}
}
func TestNClient4ValidatesInputs(t *testing.T) {
c := dhcp.NewNClient4(0, 0) // defaults applied
if _, err := c.Acquire(context.Background(), dhcp.AcquireRequest{}); err == nil {
t.Fatal("expected bridge-required error")
}
mac, _ := net.ParseMAC("02:00:00:aa:bb:cc")
if _, err := c.Acquire(context.Background(), dhcp.AcquireRequest{Bridge: "br-wan"}); err == nil {
t.Fatal("expected hw_addr-required error")
}
_ = mac
}
func TestNClient4RenewRejectsNilAndMissingRaw(t *testing.T) {
c := dhcp.NewNClient4(0, 0)
if _, err := c.Renew(context.Background(), nil); err == nil {
t.Fatal("expected nil-lease error")
}
lease := &dhcp.Lease{Bridge: "br-wan", ClientID: "x"} // no Raw bytes
if _, err := c.Renew(context.Background(), lease); err == nil {
t.Fatal("expected missing-raw-bytes error")
}
}
func TestNClient4ReleaseNilIsNoop(t *testing.T) {
c := dhcp.NewNClient4(0, 0)
if err := c.Release(context.Background(), nil); err != nil {
t.Fatalf("expected nil, got %v", err)
}
}
func TestNClient4ReleaseRejectsMissingRaw(t *testing.T) {
c := dhcp.NewNClient4(0, 0)
lease := &dhcp.Lease{Bridge: "br-wan", ClientID: "x"}
if err := c.Release(context.Background(), lease); err == nil {
t.Fatal("expected missing-raw-bytes error")
}
}
@@ -323,6 +323,64 @@ func TestDHCPManager_JitterWithinBounds(t *testing.T) {
}
}
func TestNewDHCPManager_RejectsNilDeps(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
_, err := NewDHCPManager(nil, js, dhcp.NewFake())
require.ErrorContains(t, err, "nats.Conn is required")
_, err = NewDHCPManager(nc, nil, dhcp.NewFake())
require.ErrorContains(t, err, "JetStreamContext is required")
_, err = NewDHCPManager(nc, js, nil)
require.ErrorContains(t, err, "dhcp.Client is required")
}
func TestDHCPManager_OptionsApply(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
customMAC := func(string) net.HardwareAddr {
mac, _ := net.ParseMAC("02:de:ad:be:ef:01")
return mac
}
m, err := NewDHCPManager(nc, js, dhcp.NewFake(),
WithDHCPMACFunc(customMAC),
WithDHCPAcquireTimeout(1*time.Second),
WithDHCPJitterFraction(0.05),
)
require.NoError(t, err)
t.Cleanup(m.Close)
require.Equal(t, time.Second, m.acquireTimeout)
require.InDelta(t, 0.05, m.jitterFraction, 0.0001)
require.Equal(t, "02:de:ad:be:ef:01", m.macForClientID("anything").String())
}
func TestDHCPManager_BootstrapEmptyBucket(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
m, err := NewDHCPManager(nc, js, dhcp.NewFake())
require.NoError(t, err)
t.Cleanup(m.Close)
// No leases in KV → Bootstrap is a clean no-op.
require.NoError(t, m.Bootstrap(context.Background()))
}
func TestDHCPManager_JitterWithZeroFractionReturnsInput(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
m, err := NewDHCPManager(nc, js, dhcp.NewFake(), WithDHCPJitterFraction(0))
require.NoError(t, err)
t.Cleanup(m.Close)
require.Equal(t, 100*time.Millisecond, m.jitter(100*time.Millisecond))
require.Equal(t, time.Duration(0), m.jitter(0))
}
func TestDHCPManager_ClassifyRenewErr(t *testing.T) {
require.Equal(t, "unknown", classifyRenewErr(nil))
require.Equal(t, "nak", classifyRenewErr(errors.New("got NAK from server")))
require.Equal(t, "server_unreachable", classifyRenewErr(errors.New("i/o timeout")))
require.Equal(t, "renew_failed", classifyRenewErr(errors.New("something else")))
}
func TestDHCPManager_EncodeDecodeLeaseRoundTrip(t *testing.T) {
mac, _ := net.ParseMAC("02:00:00:de:ad:be")
in := &dhcp.Lease{