feat: NAT Gateway — CRUD + vpcd SNAT rules + route table integration (mulga-750)

Implements AWS-compatible NAT Gateways for private subnet outbound internet:

API (3 actions):
- CreateNatGateway: validates subnet exists, EIP allocated, stores in KV
- DeleteNatGateway: removes record, publishes delete events for SNAT cleanup
- DescribeNatGateways: filter by ID, vpc-id, state

vpcd SNAT rule management:
- handleAddNATGateway: creates OVN snat rule (subnet CIDR → NAT GW EIP)
- handleDeleteNATGateway: removes OVN snat rule
- Subscribes to vpc.add-nat-gateway / vpc.delete-nat-gateway topics

Route table integration:
- CreateRoute now accepts NatGatewayId target (in addition to GatewayId)
- Validates NAT GW exists and belongs to same VPC
- Publishes vpc.add-nat-gateway events for associated subnets

OVN mechanics:
- Single snat rule per private subnet CIDR (shared by all VMs)
- No policy-based routing needed — SNAT rewrites before route lookup
- Existing 0.0.0.0/0 default route handles all egress
- Connection tracking handles return traffic automatically

spinifex/daemon/daemon.go

@@ -38,6 +38,7 @@ import (
 	handlers_ec2_image "github.com/mulgadc/spinifex/spinifex/handlers/ec2/image"
 	handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
 	handlers_ec2_key "github.com/mulgadc/spinifex/spinifex/handlers/ec2/key"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
 	handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
 	handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
 	handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot"
@@ -113,6 +114,7 @@ type Daemon struct {
 	vpcService            *handlers_ec2_vpc.VPCServiceImpl
 	eipService            *handlers_ec2_eip.EIPServiceImpl
 	routeTableService     *handlers_ec2_routetable.RouteTableServiceImpl
+	natGatewayService     *handlers_ec2_natgw.NatGatewayServiceImpl
 	externalIPAM          *handlers_ec2_vpc.ExternalIPAM
 	ctx                   context.Context
 	cancel                context.CancelFunc
@@ -404,6 +406,9 @@ func (d *Daemon) subscribeAll() error {
 		{"ec2.RemoveInstanceFromPlacementGroup", d.handleEC2RemoveInstanceFromPlacementGroup, "spinifex-workers"},
 		{"ec2.ReserveClusterNode", d.handleEC2ReserveClusterNode, "spinifex-workers"},
 		{"ec2.FinalizeClusterInstances", d.handleEC2FinalizeClusterInstances, "spinifex-workers"},
+		{"ec2.CreateNatGateway", d.handleEC2CreateNatGateway, "spinifex-workers"},
+		{"ec2.DeleteNatGateway", d.handleEC2DeleteNatGateway, "spinifex-workers"},
+		{"ec2.DescribeNatGateways", d.handleEC2DescribeNatGateways, "spinifex-workers"},
 		{"ec2.CreateRouteTable", d.handleEC2CreateRouteTable, "spinifex-workers"},
 		{"ec2.DeleteRouteTable", d.handleEC2DeleteRouteTable, "spinifex-workers"},
 		{"ec2.DescribeRouteTables", d.handleEC2DescribeRouteTables, "spinifex-workers"},
@@ -576,6 +581,13 @@ func (d *Daemon) Start() error {
 		return fmt.Errorf("failed to initialize RouteTable service: %w", err)
 	}
 
+	d.natGatewayService, err = initServiceWithRetry("NatGateway service", func() (*handlers_ec2_natgw.NatGatewayServiceImpl, error) {
+		return handlers_ec2_natgw.NewNatGatewayServiceImplWithNATS(d.natsConn)
+	})
+	if err != nil {
+		return fmt.Errorf("failed to initialize NatGateway service: %w", err)
+	}
+
 	// Initialize external IPAM if pool mode is configured (per-VM public IPs).
 	// NAT mode only uses SNAT via a shared gateway IP — no per-VM allocation needed.
 	if d.clusterConfig != nil && d.clusterConfig.Network.ExternalMode == "pool" {

spinifex/daemon/daemon_handlers_natgw.go

@@ -0,0 +1,17 @@
+package daemon
+
+import (
+	"github.com/nats-io/nats.go"
+)
+
+func (d *Daemon) handleEC2CreateNatGateway(msg *nats.Msg) {
+	handleNATSRequest(msg, d.natGatewayService.CreateNatGateway)
+}
+
+func (d *Daemon) handleEC2DeleteNatGateway(msg *nats.Msg) {
+	handleNATSRequest(msg, d.natGatewayService.DeleteNatGateway)
+}
+
+func (d *Daemon) handleEC2DescribeNatGateways(msg *nats.Msg) {
+	handleNATSRequest(msg, d.natGatewayService.DescribeNatGateways)
+}

spinifex/gateway/ec2.go

@@ -19,6 +19,7 @@ import (
 	gateway_ec2_image "github.com/mulgadc/spinifex/spinifex/gateway/ec2/image"
 	gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
 	gateway_ec2_key "github.com/mulgadc/spinifex/spinifex/gateway/ec2/key"
+	gateway_ec2_natgw "github.com/mulgadc/spinifex/spinifex/gateway/ec2/natgw"
 	gateway_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/gateway/ec2/placementgroup"
 	gateway_ec2_routetable "github.com/mulgadc/spinifex/spinifex/gateway/ec2/routetable"
 	gateway_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/gateway/ec2/snapshot"
@@ -317,6 +318,15 @@ var ec2Actions = map[string]EC2Handler{
 	"DescribeAddresses": ec2Handler(func(input *ec2.DescribeAddressesInput, gw *GatewayConfig, accountID string) (any, error) {
 		return gateway_ec2_eip.DescribeAddresses(input, gw.NATSConn, accountID)
 	}),
+	"CreateNatGateway": ec2Handler(func(input *ec2.CreateNatGatewayInput, gw *GatewayConfig, accountID string) (any, error) {
+		return gateway_ec2_natgw.CreateNatGateway(input, gw.NATSConn, accountID)
+	}),
+	"DeleteNatGateway": ec2Handler(func(input *ec2.DeleteNatGatewayInput, gw *GatewayConfig, accountID string) (any, error) {
+		return gateway_ec2_natgw.DeleteNatGateway(input, gw.NATSConn, accountID)
+	}),
+	"DescribeNatGateways": ec2Handler(func(input *ec2.DescribeNatGatewaysInput, gw *GatewayConfig, accountID string) (any, error) {
+		return gateway_ec2_natgw.DescribeNatGateways(input, gw.NATSConn, accountID)
+	}),
 }
 
 // ec2LocalActions are actions that don't require a NATS connection.

spinifex/gateway/ec2/natgw/CreateNatGateway.go

@@ -0,0 +1,23 @@
+package gateway_ec2_natgw
+
+import (
+	"errors"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
+	"github.com/nats-io/nats.go"
+)
+
+func CreateNatGateway(input *ec2.CreateNatGatewayInput, natsConn *nats.Conn, accountID string) (ec2.CreateNatGatewayOutput, error) {
+	var output ec2.CreateNatGatewayOutput
+	if input == nil {
+		return output, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn)
+	result, err := svc.CreateNatGateway(input, accountID)
+	if err != nil {
+		return output, err
+	}
+	return *result, nil
+}

spinifex/gateway/ec2/natgw/DeleteNatGateway.go

@@ -0,0 +1,23 @@
+package gateway_ec2_natgw
+
+import (
+	"errors"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
+	"github.com/nats-io/nats.go"
+)
+
+func DeleteNatGateway(input *ec2.DeleteNatGatewayInput, natsConn *nats.Conn, accountID string) (ec2.DeleteNatGatewayOutput, error) {
+	var output ec2.DeleteNatGatewayOutput
+	if input == nil {
+		return output, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn)
+	result, err := svc.DeleteNatGateway(input, accountID)
+	if err != nil {
+		return output, err
+	}
+	return *result, nil
+}

spinifex/gateway/ec2/natgw/DescribeNatGateways.go

@@ -0,0 +1,23 @@
+package gateway_ec2_natgw
+
+import (
+	"errors"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
+	"github.com/nats-io/nats.go"
+)
+
+func DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, natsConn *nats.Conn, accountID string) (ec2.DescribeNatGatewaysOutput, error) {
+	var output ec2.DescribeNatGatewaysOutput
+	if input == nil {
+		return output, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn)
+	result, err := svc.DescribeNatGateways(input, accountID)
+	if err != nil {
+		return output, err
+	}
+	return *result, nil
+}

spinifex/gateway/gateway_test.go

@@ -1032,6 +1032,7 @@ func TestEC2ActionMapCompleteness(t *testing.T) {
 		"CreateRouteTable", "DeleteRouteTable", "DescribeRouteTables",
 		"CreateRoute", "DeleteRoute", "ReplaceRoute",
 		"AssociateRouteTable", "DisassociateRouteTable", "ReplaceRouteTableAssociation",
+		"CreateNatGateway", "DeleteNatGateway", "DescribeNatGateways",
 	}
 
 	for _, action := range expectedActions {

spinifex/handlers/ec2/natgw/service.go

@@ -0,0 +1,10 @@
+package handlers_ec2_natgw
+
+import "github.com/aws/aws-sdk-go/service/ec2"
+
+// NatGatewayService defines the interface for NAT Gateway operations
+type NatGatewayService interface {
+	CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error)
+	DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error)
+	DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error)
+}

spinifex/handlers/ec2/natgw/service_impl.go

@@ -0,0 +1,451 @@
+package handlers_ec2_natgw
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+var _ NatGatewayService = (*NatGatewayServiceImpl)(nil)
+
+// NatGatewayServiceImpl implements NAT Gateway operations with NATS JetStream persistence
+type NatGatewayServiceImpl struct {
+	natgwKV  nats.KeyValue
+	eipKV    nats.KeyValue
+	subnetKV nats.KeyValue
+	vpcKV    nats.KeyValue
+	natsConn *nats.Conn
+}
+
+// NewNatGatewayServiceImplWithNATS creates a NAT Gateway service
+func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceImpl, error) {
+	js, err := natsConn.JetStream()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
+	}
+
+	natgwKV, err := utils.GetOrCreateKVBucket(js, KVBucketNatGateways, 10)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err)
+	}
+	if err := utils.WriteVersion(natgwKV, KVBucketNatGatewaysVersion); err != nil {
+		return nil, fmt.Errorf("write version to %s: %w", KVBucketNatGateways, err)
+	}
+
+	eipKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_eip.KVBucketEIPs, 10)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err)
+	}
+
+	subnetKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketSubnets, 10)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
+	}
+
+	vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err)
+	}
+
+	slog.Info("NatGateway service initialized with JetStream KV", "bucket", KVBucketNatGateways)
+
+	return &NatGatewayServiceImpl{
+		natgwKV:  natgwKV,
+		eipKV:    eipKV,
+		subnetKV: subnetKV,
+		vpcKV:    vpcKV,
+		natsConn: natsConn,
+	}, nil
+}
+
+// natGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway topics.
+type natGatewayEvent struct {
+	VpcId        string `json:"vpc_id"`
+	NatGatewayId string `json:"nat_gateway_id"`
+	PublicIp     string `json:"public_ip"`
+	SubnetCidr   string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule
+}
+
+// CreateNatGateway creates a NAT Gateway in a public subnet with an EIP
+func (s *NatGatewayServiceImpl) CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error) {
+	if input.SubnetId == nil || *input.SubnetId == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+	if input.AllocationId == nil || *input.AllocationId == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	subnetID := *input.SubnetId
+	allocID := *input.AllocationId
+
+	// Validate subnet exists and get its VPC
+	subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, subnetID))
+	if err != nil {
+		return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
+	}
+	var subnetRecord handlers_ec2_vpc.SubnetRecord
+	if err := json.Unmarshal(subnetEntry.Value(), &subnetRecord); err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Validate EIP exists and is not already associated
+	eipEntry, err := s.eipKV.Get(utils.AccountKey(accountID, allocID))
+	if err != nil {
+		return nil, errors.New(awserrors.ErrorInvalidAllocationIDNotFound)
+	}
+	var eipRecord handlers_ec2_eip.EIPRecord
+	if err := json.Unmarshal(eipEntry.Value(), &eipRecord); err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if eipRecord.AssociationId != "" {
+		return nil, errors.New(awserrors.ErrorResourceAlreadyAssociated)
+	}
+
+	natgwID := utils.GenerateResourceID("nat")
+	record := NatGatewayRecord{
+		NatGatewayId: natgwID,
+		VpcId:        subnetRecord.VpcId,
+		SubnetId:     subnetID,
+		AllocationId: allocID,
+		PublicIp:     eipRecord.PublicIp,
+		State:        "available",
+		AccountID:    accountID,
+		Tags:         make(map[string]string),
+		CreatedAt:    time.Now(),
+	}
+
+	data, err := json.Marshal(record)
+	if err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	if _, err := s.natgwKV.Put(utils.AccountKey(accountID, natgwID), data); err != nil {
+		slog.Error("Failed to store NAT Gateway", "natGatewayId", natgwID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	slog.Info("CreateNatGateway completed", "natGatewayId", natgwID, "subnetId", subnetID,
+		"allocationId", allocID, "publicIp", eipRecord.PublicIp, "vpcId", subnetRecord.VpcId, "accountID", accountID)
+
+	return &ec2.CreateNatGatewayOutput{
+		NatGateway: recordToEC2(&record),
+	}, nil
+}
+
+// DeleteNatGateway deletes a NAT Gateway and removes OVN SNAT rules
+func (s *NatGatewayServiceImpl) DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error) {
+	if input.NatGatewayId == nil || *input.NatGatewayId == "" {
+		return nil, errors.New(awserrors.ErrorMissingParameter)
+	}
+
+	natgwID := *input.NatGatewayId
+	key := utils.AccountKey(accountID, natgwID)
+
+	entry, err := s.natgwKV.Get(key)
+	if err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
+		}
+		slog.Error("Failed to read NAT Gateway", "natGatewayId", natgwID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	var record NatGatewayRecord
+	if err := json.Unmarshal(entry.Value(), &record); err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	// Find all private subnets whose route tables have 0.0.0.0/0 → this NAT GW
+	// and publish delete events for their SNAT rules
+	s.publishDeleteEventsForNatGateway(&record, accountID)
+
+	// Delete from KV
+	if err := s.natgwKV.Delete(key); err != nil {
+		slog.Error("Failed to delete NAT Gateway", "natGatewayId", natgwID, "err", err)
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	slog.Info("DeleteNatGateway completed", "natGatewayId", natgwID, "accountID", accountID)
+
+	return &ec2.DeleteNatGatewayOutput{
+		NatGatewayId: aws.String(natgwID),
+	}, nil
+}
+
+// publishDeleteEventsForNatGateway finds all SNAT rules associated with this NAT GW
+// by scanning route tables for routes targeting it, and publishes delete events.
+func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(record *NatGatewayRecord, accountID string) {
+	if s.natsConn == nil {
+		return
+	}
+
+	// Scan route tables in this VPC for routes targeting this NAT GW
+	rtbKV, err := utils.GetOrCreateKVBucket(mustJS(s.natsConn), "spinifex-vpc-route-tables", 10)
+	if err != nil {
+		slog.Warn("Cannot scan route tables for NAT GW cleanup", "err", err)
+		return
+	}
+
+	keys, err := rtbKV.Keys()
+	if err != nil {
+		return
+	}
+
+	prefix := accountID + "."
+	for _, key := range keys {
+		if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
+			continue
+		}
+		entry, err := rtbKV.Get(key)
+		if err != nil {
+			continue
+		}
+
+		// Minimal route table parse — just need routes and associations
+		var rtbData struct {
+			VpcId  string `json:"vpc_id"`
+			Routes []struct {
+				NatGatewayId string `json:"nat_gateway_id"`
+			} `json:"routes"`
+			Associations []struct {
+				SubnetId string `json:"subnet_id"`
+			} `json:"associations"`
+		}
+		if err := json.Unmarshal(entry.Value(), &rtbData); err != nil {
+			continue
+		}
+		if rtbData.VpcId != record.VpcId {
+			continue
+		}
+
+		// Check if any route targets this NAT GW
+		hasNatRoute := false
+		for _, r := range rtbData.Routes {
+			if r.NatGatewayId == record.NatGatewayId {
+				hasNatRoute = true
+				break
+			}
+		}
+		if !hasNatRoute {
+			continue
+		}
+
+		// Publish delete events for each associated subnet's CIDR
+		for _, assoc := range rtbData.Associations {
+			if assoc.SubnetId == "" {
+				continue
+			}
+			subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, assoc.SubnetId))
+			if err != nil {
+				continue
+			}
+			var subnet handlers_ec2_vpc.SubnetRecord
+			if err := json.Unmarshal(subnetEntry.Value(), &subnet); err != nil {
+				continue
+			}
+
+			evt := natGatewayEvent{
+				VpcId:        record.VpcId,
+				NatGatewayId: record.NatGatewayId,
+				PublicIp:     record.PublicIp,
+				SubnetCidr:   subnet.CidrBlock,
+			}
+			eventData, err := json.Marshal(evt)
+			if err != nil {
+				slog.Warn("Failed to marshal NAT GW delete event", "err", err)
+				continue
+			}
+			if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil {
+				slog.Warn("Failed to publish NAT GW delete event", "subnet", subnet.CidrBlock, "err", err)
+			}
+		}
+	}
+}
+
+func mustJS(nc *nats.Conn) nats.JetStreamContext {
+	js, _ := nc.JetStream()
+	return js
+}
+
+// DescribeNatGateways lists NAT Gateways, optionally filtered
+func (s *NatGatewayServiceImpl) DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error) {
+	natgwIDs := make(map[string]bool)
+	for _, id := range input.NatGatewayIds {
+		if id != nil {
+			natgwIDs[*id] = true
+		}
+	}
+
+	// Build filter map
+	filters := make(map[string][]string)
+	for _, f := range input.Filter {
+		if f.Name != nil {
+			for _, v := range f.Values {
+				if v != nil {
+					filters[*f.Name] = append(filters[*f.Name], *v)
+				}
+			}
+		}
+	}
+
+	prefix := accountID + "."
+	keys, err := s.natgwKV.Keys()
+	if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+
+	var natGateways []*ec2.NatGateway
+	foundIDs := make(map[string]bool)
+
+	for _, key := range keys {
+		if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
+			continue
+		}
+
+		entry, err := s.natgwKV.Get(key)
+		if err != nil {
+			if errors.Is(err, nats.ErrKeyNotFound) {
+				continue
+			}
+			slog.Error("Failed to read NAT Gateway", "key", key, "err", err)
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+
+		var record NatGatewayRecord
+		if err := json.Unmarshal(entry.Value(), &record); err != nil {
+			slog.Error("Corrupt NAT Gateway record", "key", key, "err", err)
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+
+		if len(natgwIDs) > 0 && !natgwIDs[record.NatGatewayId] {
+			continue
+		}
+
+		// Apply filters
+		if vpcFilter, ok := filters["vpc-id"]; ok {
+			found := slices.Contains(vpcFilter, record.VpcId)
+			if !found {
+				continue
+			}
+		}
+		if stateFilter, ok := filters["state"]; ok {
+			found := slices.Contains(stateFilter, record.State)
+			if !found {
+				continue
+			}
+		}
+
+		natGateways = append(natGateways, recordToEC2(&record))
+		foundIDs[record.NatGatewayId] = true
+	}
+
+	for id := range natgwIDs {
+		if !foundIDs[id] {
+			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
+		}
+	}
+
+	return &ec2.DescribeNatGatewaysOutput{
+		NatGateways: natGateways,
+	}, nil
+}
+
+// PublishAddEvent publishes a vpc.add-nat-gateway event for vpcd to create the SNAT rule.
+// Called by the route table service when CreateRoute targets a NAT GW.
+func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
+	if s.natsConn == nil {
+		return
+	}
+	evt := natGatewayEvent{
+		VpcId:        vpcId,
+		NatGatewayId: natGatewayId,
+		PublicIp:     publicIp,
+		SubnetCidr:   subnetCidr,
+	}
+	eventData, err := json.Marshal(evt)
+	if err != nil {
+		slog.Error("Failed to marshal NAT GW add event", "err", err)
+		return
+	}
+	if err := s.natsConn.Publish("vpc.add-nat-gateway", eventData); err != nil {
+		slog.Warn("Failed to publish NAT GW add event", "err", err)
+	}
+}
+
+// PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule.
+func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
+	if s.natsConn == nil {
+		return
+	}
+	evt := natGatewayEvent{
+		VpcId:        vpcId,
+		NatGatewayId: natGatewayId,
+		PublicIp:     publicIp,
+		SubnetCidr:   subnetCidr,
+	}
+	eventData, err := json.Marshal(evt)
+	if err != nil {
+		slog.Error("Failed to marshal NAT GW delete event", "err", err)
+		return
+	}
+	if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil {
+		slog.Warn("Failed to publish NAT GW delete event", "err", err)
+	}
+}
+
+// GetNatGateway retrieves a NAT Gateway record by ID
+func (s *NatGatewayServiceImpl) GetNatGateway(accountID, natgwID string) (*NatGatewayRecord, error) {
+	entry, err := s.natgwKV.Get(utils.AccountKey(accountID, natgwID))
+	if err != nil {
+		if errors.Is(err, nats.ErrKeyNotFound) {
+			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
+		}
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	var record NatGatewayRecord
+	if err := json.Unmarshal(entry.Value(), &record); err != nil {
+		return nil, errors.New(awserrors.ErrorServerInternal)
+	}
+	return &record, nil
+}
+
+func recordToEC2(record *NatGatewayRecord) *ec2.NatGateway {
+	ngw := &ec2.NatGateway{
+		NatGatewayId:     aws.String(record.NatGatewayId),
+		VpcId:            aws.String(record.VpcId),
+		SubnetId:         aws.String(record.SubnetId),
+		State:            aws.String(record.State),
+		ConnectivityType: aws.String("public"),
+		CreateTime:       aws.Time(record.CreatedAt),
+	}
+
+	if record.PublicIp != "" {
+		ngw.NatGatewayAddresses = []*ec2.NatGatewayAddress{
+			{
+				AllocationId: aws.String(record.AllocationId),
+				PublicIp:     aws.String(record.PublicIp),
+			},
+		}
+	}
+
+	if len(record.Tags) > 0 {
+		tags := make([]*ec2.Tag, 0, len(record.Tags))
+		for k, v := range record.Tags {
+			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
+		}
+		ngw.Tags = tags
+	}
+
+	return ngw
+}

spinifex/handlers/ec2/natgw/service_nats.go

@@ -0,0 +1,31 @@
+package handlers_ec2_natgw
+
+import (
+	"time"
+
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// NATSNatGatewayService handles NAT Gateway operations via NATS messaging
+type NATSNatGatewayService struct {
+	natsConn *nats.Conn
+}
+
+// NewNATSNatGatewayService creates a new NATS-based NAT Gateway service
+func NewNATSNatGatewayService(conn *nats.Conn) NatGatewayService {
+	return &NATSNatGatewayService{natsConn: conn}
+}
+
+func (s *NATSNatGatewayService) CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error) {
+	return utils.NATSRequest[ec2.CreateNatGatewayOutput](s.natsConn, "ec2.CreateNatGateway", input, 30*time.Second, accountID)
+}
+
+func (s *NATSNatGatewayService) DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error) {
+	return utils.NATSRequest[ec2.DeleteNatGatewayOutput](s.natsConn, "ec2.DeleteNatGateway", input, 30*time.Second, accountID)
+}
+
+func (s *NATSNatGatewayService) DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error) {
+	return utils.NATSRequest[ec2.DescribeNatGatewaysOutput](s.natsConn, "ec2.DescribeNatGateways", input, 30*time.Second, accountID)
+}

spinifex/handlers/ec2/natgw/types.go

@@ -0,0 +1,21 @@
+package handlers_ec2_natgw
+
+import "time"
+
+const (
+	KVBucketNatGateways        = "spinifex-vpc-nat-gateways"
+	KVBucketNatGatewaysVersion = 1
+)
+
+// NatGatewayRecord represents a stored NAT Gateway
+type NatGatewayRecord struct {
+	NatGatewayId string            `json:"nat_gateway_id"`
+	VpcId        string            `json:"vpc_id"`
+	SubnetId     string            `json:"subnet_id"`     // public subnet where NAT GW lives
+	AllocationId string            `json:"allocation_id"` // EIP allocation
+	PublicIp     string            `json:"public_ip"`     // EIP address
+	State        string            `json:"state"`         // pending, available, deleting, deleted, failed
+	AccountID    string            `json:"account_id"`
+	Tags         map[string]string `json:"tags"`
+	CreatedAt    time.Time         `json:"created_at"`
+}

spinifex/handlers/ec2/routetable/service_impl.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
+	handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
@@ -29,6 +30,7 @@ type RouteTableServiceImpl struct {
 	vpcKV    nats.KeyValue
 	igwKV    nats.KeyValue
 	subnetKV nats.KeyValue
+	natgwKV  nats.KeyValue
 	natsConn *nats.Conn
 }
 
@@ -62,6 +64,11 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (
 		return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err)
 	}
 
+	natgwKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_natgw.KVBucketNatGateways, 10)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get NAT Gateway KV bucket: %w", err)
+	}
+
 	slog.Info("RouteTable service initialized with JetStream KV", "bucket", KVBucketRouteTables)
 
 	return &RouteTableServiceImpl{
@@ -70,6 +77,7 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (
 		vpcKV:    vpcKV,
 		igwKV:    igwKV,
 		subnetKV: subnetKV,
+		natgwKV:  natgwKV,
 		natsConn: natsConn,
 	}, nil
 }
@@ -347,39 +355,69 @@ func (s *RouteTableServiceImpl) CreateRoute(input *ec2.CreateRouteInput, account
 		}
 	}
 
-	// V1: only GatewayId target supported
-	if input.GatewayId == nil || *input.GatewayId == "" {
+	var route RouteRecord
+
+	switch {
+	case input.GatewayId != nil && *input.GatewayId != "":
+		igwID := *input.GatewayId
+		// Verify IGW exists and is attached to the same VPC
+		igwEntry, err := s.igwKV.Get(utils.AccountKey(accountID, igwID))
+		if err != nil {
+			return nil, errors.New(awserrors.ErrorInvalidInternetGatewayIDNotFound)
+		}
+		var igwRecord handlers_ec2_igw.IGWRecord
+		if err := json.Unmarshal(igwEntry.Value(), &igwRecord); err != nil {
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+		if igwRecord.VpcId != record.VpcId {
+			return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+		}
+		route = RouteRecord{
+			DestinationCidrBlock: destCidr,
+			GatewayId:            igwID,
+			State:                "active",
+			Origin:               "CreateRoute",
+		}
+
+	case input.NatGatewayId != nil && *input.NatGatewayId != "":
+		natgwID := *input.NatGatewayId
+		// Verify NAT GW exists and belongs to the same VPC
+		natgwEntry, err := s.natgwKV.Get(utils.AccountKey(accountID, natgwID))
+		if err != nil {
+			return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
+		}
+		var natgwRecord struct {
+			NatGatewayId string `json:"nat_gateway_id"`
+			VpcId        string `json:"vpc_id"`
+			PublicIp     string `json:"public_ip"`
+		}
+		if err := json.Unmarshal(natgwEntry.Value(), &natgwRecord); err != nil {
+			return nil, errors.New(awserrors.ErrorServerInternal)
+		}
+		if natgwRecord.VpcId != record.VpcId {
+			return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+		}
+		route = RouteRecord{
+			DestinationCidrBlock: destCidr,
+			NatGatewayId:         natgwID,
+			State:                "active",
+			Origin:               "CreateRoute",
+		}
+
+		// Publish vpc.add-nat-gateway events for each subnet associated with this route table
+		s.publishNatGatewayEvents(accountID, record, natgwRecord.VpcId, natgwID, natgwRecord.PublicIp)
+
+	default:
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
 
-	igwID := *input.GatewayId
-
-	// Verify IGW exists and is attached to the same VPC
-	igwEntry, err := s.igwKV.Get(utils.AccountKey(accountID, igwID))
-	if err != nil {
-		return nil, errors.New(awserrors.ErrorInvalidInternetGatewayIDNotFound)
-	}
-	var igwRecord handlers_ec2_igw.IGWRecord
-	if err := json.Unmarshal(igwEntry.Value(), &igwRecord); err != nil {
-		return nil, errors.New(awserrors.ErrorServerInternal)
-	}
-	if igwRecord.VpcId != record.VpcId {
-		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
-	}
-
-	route := RouteRecord{
-		DestinationCidrBlock: destCidr,
-		GatewayId:            igwID,
-		State:                "active",
-		Origin:               "CreateRoute",
-	}
 	record.Routes = append(record.Routes, route)
 
 	if err := s.putRouteTable(accountID, record); err != nil {
 		return nil, err
 	}
 
-	slog.Info("CreateRoute completed", "routeTableId", rtbID, "destination", destCidr, "gatewayId", igwID, "accountID", accountID)
+	slog.Info("CreateRoute completed", "routeTableId", rtbID, "destination", destCidr, "accountID", accountID)
 
 	return &ec2.CreateRouteOutput{
 		Return: aws.Bool(true),
@@ -784,6 +822,49 @@ func containsString(slice []string, s string) bool {
 	return slices.Contains(slice, s)
 }
 
+// publishNatGatewayEvents publishes vpc.add-nat-gateway events for each subnet
+// associated with this route table, so vpcd creates the SNAT rules.
+func (s *RouteTableServiceImpl) publishNatGatewayEvents(accountID string, record *RouteTableRecord, vpcID, natgwID, publicIp string) {
+	if s.natsConn == nil {
+		return
+	}
+	for _, assoc := range record.Associations {
+		if assoc.SubnetId == "" || assoc.Main {
+			continue
+		}
+		subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, assoc.SubnetId))
+		if err != nil {
+			continue
+		}
+		var subnet handlers_ec2_vpc.SubnetRecord
+		if err := json.Unmarshal(subnetEntry.Value(), &subnet); err != nil {
+			continue
+		}
+
+		evt := struct {
+			VpcId        string `json:"vpc_id"`
+			NatGatewayId string `json:"nat_gateway_id"`
+			PublicIp     string `json:"public_ip"`
+			SubnetCidr   string `json:"subnet_cidr"`
+		}{
+			VpcId:        vpcID,
+			NatGatewayId: natgwID,
+			PublicIp:     publicIp,
+			SubnetCidr:   subnet.CidrBlock,
+		}
+		eventData, err := json.Marshal(evt)
+		if err != nil {
+			slog.Warn("Failed to marshal NAT GW add event", "err", err)
+			continue
+		}
+		if err := s.natsConn.Publish("vpc.add-nat-gateway", eventData); err != nil {
+			slog.Warn("Failed to publish NAT GW add event", "subnet", assoc.SubnetId, "err", err)
+		} else {
+			slog.Info("Published NAT GW add event", "subnetCidr", subnet.CidrBlock, "publicIp", publicIp)
+		}
+	}
+}
+
 // recordToEC2 converts an internal record to an AWS SDK RouteTable struct
 func recordToEC2(record *RouteTableRecord) *ec2.RouteTable {
 	rtb := &ec2.RouteTable{

spinifex/services/vpcd/topology.go

@@ -15,17 +15,19 @@ import (
 
 // NATS topics for VPC lifecycle events published by the daemon.
 const (
-	TopicVPCCreate    = "vpc.create"
-	TopicVPCDelete    = "vpc.delete"
-	TopicSubnetCreate = "vpc.create-subnet"
-	TopicSubnetDelete = "vpc.delete-subnet"
-	TopicCreatePort   = "vpc.create-port"
-	TopicDeletePort   = "vpc.delete-port"
-	TopicPortStatus   = "vpc.port-status"
-	TopicIGWAttach    = "vpc.igw-attach"
-	TopicIGWDetach    = "vpc.igw-detach"
-	TopicAddNAT       = "vpc.add-nat"
-	TopicDeleteNAT    = "vpc.delete-nat"
+	TopicVPCCreate        = "vpc.create"
+	TopicVPCDelete        = "vpc.delete"
+	TopicSubnetCreate     = "vpc.create-subnet"
+	TopicSubnetDelete     = "vpc.delete-subnet"
+	TopicCreatePort       = "vpc.create-port"
+	TopicDeletePort       = "vpc.delete-port"
+	TopicPortStatus       = "vpc.port-status"
+	TopicIGWAttach        = "vpc.igw-attach"
+	TopicIGWDetach        = "vpc.igw-detach"
+	TopicAddNAT           = "vpc.add-nat"
+	TopicDeleteNAT        = "vpc.delete-nat"
+	TopicAddNATGateway    = "vpc.add-nat-gateway"
+	TopicDeleteNATGateway = "vpc.delete-nat-gateway"
 )
 
 // VPCEvent is published on vpc.create after a VPC is persisted.
@@ -60,6 +62,14 @@ type NATEvent struct {
 	MAC        string `json:"mac"`       // external MAC for distributed NAT
 }
 
+// NATGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway.
+type NATGatewayEvent struct {
+	VpcId        string `json:"vpc_id"`
+	NatGatewayId string `json:"nat_gateway_id"`
+	PublicIp     string `json:"public_ip"`
+	SubnetCidr   string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule
+}
+
 // Bridge mode constants for external connectivity.
 const (
 	// BridgeModeMacvlan uses a macvlan sub-interface on the WAN NIC. SSH-safe
@@ -185,6 +195,8 @@ func (h *TopologyHandler) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error)
 		{TopicIGWDetach, h.handleIGWDetach, true},
 		{TopicAddNAT, h.handleAddNAT, true},
 		{TopicDeleteNAT, h.handleDeleteNAT, true},
+		{TopicAddNATGateway, h.handleAddNATGateway, true},
+		{TopicDeleteNATGateway, h.handleDeleteNATGateway, true},
 		{TopicCreateSG, h.handleCreateSG, true},
 		{TopicDeleteSG, h.handleDeleteSG, true},
 		{TopicUpdateSG, h.handleUpdateSG, true},
@@ -1218,3 +1230,64 @@ func respond(msg *nats.Msg, err error) {
 		slog.Error("vpcd: failed to respond to NATS request", "err", err)
 	}
 }
+
+// handleAddNATGateway creates an OVN SNAT rule for a private subnet via a NAT Gateway.
+// The SNAT rule rewrites source IPs from the private subnet CIDR to the NAT GW's public IP.
+func (h *TopologyHandler) handleAddNATGateway(msg *nats.Msg) {
+	var evt NATGatewayEvent
+	if err := json.Unmarshal(msg.Data, &evt); err != nil {
+		slog.Error("vpcd: failed to unmarshal vpc.add-nat-gateway event", "err", err)
+		return
+	}
+
+	slog.Info("vpcd: adding NAT Gateway SNAT rule",
+		"vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId,
+		"publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr)
+
+	ctx := context.Background()
+	routerName := "vpc-" + evt.VpcId
+
+	snatRule := &nbdb.NAT{
+		Type:       "snat",
+		ExternalIP: evt.PublicIp,
+		LogicalIP:  evt.SubnetCidr,
+		ExternalIDs: map[string]string{
+			"spinifex:vpc_id":         evt.VpcId,
+			"spinifex:nat_gateway_id": evt.NatGatewayId,
+		},
+	}
+
+	if err := h.ovn.AddNAT(ctx, routerName, snatRule); err != nil {
+		slog.Error("vpcd: failed to add NAT Gateway SNAT rule",
+			"router", routerName, "publicIp", evt.PublicIp,
+			"subnetCidr", evt.SubnetCidr, "err", err)
+		return
+	}
+
+	slog.Info("vpcd: NAT Gateway SNAT rule added",
+		"router", routerName, "publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr)
+}
+
+// handleDeleteNATGateway removes the OVN SNAT rule for a private subnet's NAT Gateway.
+func (h *TopologyHandler) handleDeleteNATGateway(msg *nats.Msg) {
+	var evt NATGatewayEvent
+	if err := json.Unmarshal(msg.Data, &evt); err != nil {
+		slog.Error("vpcd: failed to unmarshal vpc.delete-nat-gateway event", "err", err)
+		return
+	}
+
+	slog.Info("vpcd: removing NAT Gateway SNAT rule",
+		"vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId, "subnetCidr", evt.SubnetCidr)
+
+	ctx := context.Background()
+	routerName := "vpc-" + evt.VpcId
+
+	if err := h.ovn.DeleteNAT(ctx, routerName, "snat", evt.SubnetCidr); err != nil {
+		slog.Warn("vpcd: failed to delete NAT Gateway SNAT rule",
+			"router", routerName, "subnetCidr", evt.SubnetCidr, "err", err)
+		return
+	}
+
+	slog.Info("vpcd: NAT Gateway SNAT rule removed",
+		"router", routerName, "subnetCidr", evt.SubnetCidr)
+}