diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index dd66d39db..eaf954600 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -392,6 +392,7 @@ func (d *Daemon) subscribeAll() error { {"ec2.CreateSubnet", d.handleEC2CreateSubnet, "spinifex-workers"}, {"ec2.DeleteSubnet", d.handleEC2DeleteSubnet, "spinifex-workers"}, {"ec2.DescribeSubnets", d.handleEC2DescribeSubnets, "spinifex-workers"}, + {"ec2.ModifySubnetAttribute", d.handleEC2ModifySubnetAttribute, "spinifex-workers"}, {"ec2.CreateNetworkInterface", d.handleEC2CreateNetworkInterface, "spinifex-workers"}, {"ec2.DeleteNetworkInterface", d.handleEC2DeleteNetworkInterface, "spinifex-workers"}, {"ec2.DescribeNetworkInterfaces", d.handleEC2DescribeNetworkInterfaces, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_handlers_vpc.go b/spinifex/daemon/daemon_handlers_vpc.go index b1ccd219e..114b98400 100644 --- a/spinifex/daemon/daemon_handlers_vpc.go +++ b/spinifex/daemon/daemon_handlers_vpc.go @@ -31,6 +31,10 @@ func (d *Daemon) handleEC2DescribeSubnets(msg *nats.Msg) { handleNATSRequest(msg, d.vpcService.DescribeSubnets) } +func (d *Daemon) handleEC2ModifySubnetAttribute(msg *nats.Msg) { + handleNATSRequest(msg, d.vpcService.ModifySubnetAttribute) +} + func (d *Daemon) handleEC2CreateNetworkInterface(msg *nats.Msg) { handleNATSRequest(msg, d.vpcService.CreateNetworkInterface) } diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index 0b674f377..6b7dd18c4 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -215,6 +215,9 @@ var ec2Actions = map[string]EC2Handler{ "DescribeSubnets": ec2Handler(func(input *ec2.DescribeSubnetsInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_vpc.DescribeSubnets(input, gw.NATSConn, accountID) }), + "ModifySubnetAttribute": ec2Handler(func(input *ec2.ModifySubnetAttributeInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_vpc.ModifySubnetAttribute(input, gw.NATSConn, accountID) + }), "CreateNetworkInterface": ec2Handler(func(input *ec2.CreateNetworkInterfaceInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_vpc.CreateNetworkInterface(input, gw.NATSConn, accountID) }), diff --git a/spinifex/gateway/ec2/vpc/ModifySubnetAttribute.go b/spinifex/gateway/ec2/vpc/ModifySubnetAttribute.go new file mode 100644 index 000000000..df651749d --- /dev/null +++ b/spinifex/gateway/ec2/vpc/ModifySubnetAttribute.go @@ -0,0 +1,30 @@ +package gateway_ec2_vpc + +import ( + "errors" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/nats-io/nats.go" +) + +// ModifySubnetAttribute handles the EC2 ModifySubnetAttribute API call +func ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, natsConn *nats.Conn, accountID string) (ec2.ModifySubnetAttributeOutput, error) { + var output ec2.ModifySubnetAttributeOutput + + if input == nil { + return output, errors.New(awserrors.ErrorInvalidParameterValue) + } + if input.SubnetId == nil || *input.SubnetId == "" { + return output, errors.New(awserrors.ErrorMissingParameter) + } + + svc := handlers_ec2_vpc.NewNATSVPCService(natsConn) + result, err := svc.ModifySubnetAttribute(input, accountID) + if err != nil { + return output, err + } + + return *result, nil +} diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index 997e0526e..4031e6744 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -1021,7 +1021,7 @@ func TestEC2ActionMapCompleteness(t *testing.T) { "CreateEgressOnlyInternetGateway", "DeleteEgressOnlyInternetGateway", "DescribeEgressOnlyInternetGateways", "CreateVpc", "DeleteVpc", "DescribeVpcs", - "CreateSubnet", "DeleteSubnet", "DescribeSubnets", + "CreateSubnet", "DeleteSubnet", "DescribeSubnets", "ModifySubnetAttribute", "CreateNetworkInterface", "DeleteNetworkInterface", "DescribeNetworkInterfaces", } diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index d0827dd30..e3049e72d 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -33,6 +33,8 @@ type ENIRecord struct { AttachmentId string `json:"attachment_id,omitempty"` InstanceId string `json:"instance_id,omitempty"` DeviceIndex int64 `json:"device_index"` + PublicIpAddress string `json:"public_ip_address,omitempty"` // Auto-assigned or EIP + PublicIpPool string `json:"public_ip_pool,omitempty"` // Pool name the public IP came from Tags map[string]string `json:"tags"` CreatedAt time.Time `json:"created_at"` } @@ -346,6 +348,12 @@ func (s *VPCServiceImpl) eniRecordToEC2(record *ENIRecord, accountID string) *ec Groups: []*ec2.GroupIdentifier{}, } + if record.PublicIpAddress != "" { + eni.Association = &ec2.NetworkInterfaceAssociation{ + PublicIp: aws.String(record.PublicIpAddress), + } + } + if record.AttachmentId != "" { eni.Attachment = &ec2.NetworkInterfaceAttachment{ AttachmentId: aws.String(record.AttachmentId), diff --git a/spinifex/handlers/ec2/vpc/service.go b/spinifex/handlers/ec2/vpc/service.go index 96df7a7c4..07503e1a2 100644 --- a/spinifex/handlers/ec2/vpc/service.go +++ b/spinifex/handlers/ec2/vpc/service.go @@ -2,7 +2,7 @@ package handlers_ec2_vpc import "github.com/aws/aws-sdk-go/service/ec2" -// VPCService defines the interface for VPC, Subnet, and ENI operations +// VPCService defines the interface for VPC, Subnet, ENI, and Security Group operations type VPCService interface { CreateVpc(input *ec2.CreateVpcInput, accountID string) (*ec2.CreateVpcOutput, error) DeleteVpc(input *ec2.DeleteVpcInput, accountID string) (*ec2.DeleteVpcOutput, error) @@ -10,7 +10,15 @@ type VPCService interface { CreateSubnet(input *ec2.CreateSubnetInput, accountID string) (*ec2.CreateSubnetOutput, error) DeleteSubnet(input *ec2.DeleteSubnetInput, accountID string) (*ec2.DeleteSubnetOutput, error) DescribeSubnets(input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error) + ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error) + CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error) + DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error) + DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) + AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) + AuthorizeSecurityGroupEgress(input *ec2.AuthorizeSecurityGroupEgressInput, accountID string) (*ec2.AuthorizeSecurityGroupEgressOutput, error) + RevokeSecurityGroupIngress(input *ec2.RevokeSecurityGroupIngressInput, accountID string) (*ec2.RevokeSecurityGroupIngressOutput, error) + RevokeSecurityGroupEgress(input *ec2.RevokeSecurityGroupEgressInput, accountID string) (*ec2.RevokeSecurityGroupEgressOutput, error) } diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 64ee019fa..89cd4635c 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -45,14 +45,15 @@ type VPCRecord struct { // SubnetRecord represents a stored Subnet type SubnetRecord struct { - SubnetId string `json:"subnet_id"` - VpcId string `json:"vpc_id"` - CidrBlock string `json:"cidr_block"` - AvailabilityZone string `json:"availability_zone"` - State string `json:"state"` - IsDefault bool `json:"is_default"` - Tags map[string]string `json:"tags"` - CreatedAt time.Time `json:"created_at"` + SubnetId string `json:"subnet_id"` + VpcId string `json:"vpc_id"` + CidrBlock string `json:"cidr_block"` + AvailabilityZone string `json:"availability_zone"` + State string `json:"state"` + IsDefault bool `json:"is_default"` + MapPublicIpOnLaunch bool `json:"map_public_ip_on_launch"` + Tags map[string]string `json:"tags"` + CreatedAt time.Time `json:"created_at"` } // VPCServiceImpl implements VPC, Subnet, and ENI operations with NATS JetStream persistence @@ -63,6 +64,7 @@ type VPCServiceImpl struct { subnetKV nats.KeyValue vniKV nats.KeyValue eniKV nats.KeyValue + sgKV nats.KeyValue ipam *IPAM } @@ -105,6 +107,14 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer return nil, fmt.Errorf("write version to %s: %w", KVBucketENIs, err) } + sgKV, err := utils.GetOrCreateKVBucket(js, KVBucketSecurityGroups, 10) + if err != nil { + return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err) + } + if err := utils.WriteVersion(sgKV, KVBucketSecurityGroupsVersion); err != nil { + return nil, fmt.Errorf("write version to %s: %w", KVBucketSecurityGroups, err) + } + ipam, err := NewIPAM(js) if err != nil { return nil, fmt.Errorf("failed to initialize IPAM: %w", err) @@ -114,7 +124,8 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer "vpcBucket", KVBucketVPCs, "subnetBucket", KVBucketSubnets, "vniBucket", KVBucketVNICounter, - "eniBucket", KVBucketENIs) + "eniBucket", KVBucketENIs, + "sgBucket", KVBucketSecurityGroups) return &VPCServiceImpl{ config: cfg, @@ -123,6 +134,7 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer subnetKV: subnetKV, vniKV: vniKV, eniKV: eniKV, + sgKV: sgKV, ipam: ipam, }, nil } @@ -632,7 +644,7 @@ func (s *VPCServiceImpl) subnetRecordToEC2(record *SubnetRecord, availableIPs in DefaultForAz: aws.Bool(record.IsDefault), AvailableIpAddressCount: aws.Int64(int64(availableIPs)), OwnerId: aws.String(accountID), - MapPublicIpOnLaunch: aws.Bool(false), + MapPublicIpOnLaunch: aws.Bool(record.MapPublicIpOnLaunch), } if len(record.Tags) > 0 { @@ -646,6 +658,42 @@ func (s *VPCServiceImpl) subnetRecordToEC2(record *SubnetRecord, availableIPs in return subnet } +// ModifySubnetAttribute modifies a subnet's attributes (e.g. MapPublicIpOnLaunch). +func (s *VPCServiceImpl) ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error) { + if input.SubnetId == nil || *input.SubnetId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + subnetID := *input.SubnetId + key := utils.AccountKey(accountID, subnetID) + + entry, err := s.subnetKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound) + } + + var record SubnetRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + if input.MapPublicIpOnLaunch != nil && input.MapPublicIpOnLaunch.Value != nil { + record.MapPublicIpOnLaunch = *input.MapPublicIpOnLaunch.Value + } + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal subnet record: %w", err) + } + if _, err := s.subnetKV.Update(key, data, entry.Revision()); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("ModifySubnetAttribute completed", "subnetId", subnetID, "mapPublicIpOnLaunch", record.MapPublicIpOnLaunch, "accountID", accountID) + + return &ec2.ModifySubnetAttributeOutput{}, nil +} + // Default VPC constants matching AWS defaults. const ( DefaultVPCCidr = "172.31.0.0/16" diff --git a/spinifex/handlers/ec2/vpc/service_nats.go b/spinifex/handlers/ec2/vpc/service_nats.go index fa3525a9e..9772a1bca 100644 --- a/spinifex/handlers/ec2/vpc/service_nats.go +++ b/spinifex/handlers/ec2/vpc/service_nats.go @@ -42,6 +42,10 @@ func (s *NATSVPCService) DescribeSubnets(input *ec2.DescribeSubnetsInput, accoun return utils.NATSRequest[ec2.DescribeSubnetsOutput](s.natsConn, "ec2.DescribeSubnets", input, 30*time.Second, accountID) } +func (s *NATSVPCService) ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error) { + return utils.NATSRequest[ec2.ModifySubnetAttributeOutput](s.natsConn, "ec2.ModifySubnetAttribute", input, 30*time.Second, accountID) +} + func (s *NATSVPCService) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) { return utils.NATSRequest[ec2.CreateNetworkInterfaceOutput](s.natsConn, "ec2.CreateNetworkInterface", input, 30*time.Second, accountID) } @@ -53,3 +57,31 @@ func (s *NATSVPCService) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfac func (s *NATSVPCService) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error) { return utils.NATSRequest[ec2.DescribeNetworkInterfacesOutput](s.natsConn, "ec2.DescribeNetworkInterfaces", input, 30*time.Second, accountID) } + +func (s *NATSVPCService) CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error) { + return utils.NATSRequest[ec2.CreateSecurityGroupOutput](s.natsConn, "ec2.CreateSecurityGroup", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error) { + return utils.NATSRequest[ec2.DeleteSecurityGroupOutput](s.natsConn, "ec2.DeleteSecurityGroup", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) { + return utils.NATSRequest[ec2.DescribeSecurityGroupsOutput](s.natsConn, "ec2.DescribeSecurityGroups", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) { + return utils.NATSRequest[ec2.AuthorizeSecurityGroupIngressOutput](s.natsConn, "ec2.AuthorizeSecurityGroupIngress", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) AuthorizeSecurityGroupEgress(input *ec2.AuthorizeSecurityGroupEgressInput, accountID string) (*ec2.AuthorizeSecurityGroupEgressOutput, error) { + return utils.NATSRequest[ec2.AuthorizeSecurityGroupEgressOutput](s.natsConn, "ec2.AuthorizeSecurityGroupEgress", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) RevokeSecurityGroupIngress(input *ec2.RevokeSecurityGroupIngressInput, accountID string) (*ec2.RevokeSecurityGroupIngressOutput, error) { + return utils.NATSRequest[ec2.RevokeSecurityGroupIngressOutput](s.natsConn, "ec2.RevokeSecurityGroupIngress", input, 30*time.Second, accountID) +} + +func (s *NATSVPCService) RevokeSecurityGroupEgress(input *ec2.RevokeSecurityGroupEgressInput, accountID string) (*ec2.RevokeSecurityGroupEgressOutput, error) { + return utils.NATSRequest[ec2.RevokeSecurityGroupEgressOutput](s.natsConn, "ec2.RevokeSecurityGroupEgress", input, 30*time.Second, accountID) +} diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index fade656a2..1b1e9fffa 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -23,6 +23,8 @@ const ( TopicPortStatus = "vpc.port-status" TopicIGWAttach = "vpc.igw-attach" TopicIGWDetach = "vpc.igw-detach" + TopicAddNAT = "vpc.add-nat" + TopicDeleteNAT = "vpc.delete-nat" ) // VPCEvent is published on vpc.create after a VPC is persisted. @@ -48,6 +50,15 @@ type PortEvent struct { MacAddress string `json:"mac_address"` } +// NATEvent is published on vpc.add-nat / vpc.delete-nat for 1:1 public IP NAT. +type NATEvent struct { + VpcId string `json:"vpc_id"` + ExternalIP string `json:"external_ip"` + LogicalIP string `json:"logical_ip"` + PortName string `json:"port_name"` // logical port for distributed NAT + MAC string `json:"mac"` // external MAC for distributed NAT +} + // TopologyHandler translates VPC lifecycle NATS events into OVN NB DB operations. type TopologyHandler struct { ovn OVNClient @@ -123,6 +134,11 @@ func (h *TopologyHandler) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) {TopicDeletePort, h.handleDeletePort, true}, {TopicIGWAttach, h.handleIGWAttach, true}, {TopicIGWDetach, h.handleIGWDetach, true}, + {TopicAddNAT, h.handleAddNAT, true}, + {TopicDeleteNAT, h.handleDeleteNAT, true}, + {TopicCreateSG, h.handleCreateSG, true}, + {TopicDeleteSG, h.handleDeleteSG, true}, + {TopicUpdateSG, h.handleUpdateSG, true}, } var result []*nats.Subscription @@ -765,6 +781,83 @@ func (h *TopologyHandler) handleIGWDetach(msg *nats.Msg) { respond(msg, nil) } +// --- NAT (dnat_and_snat for public IPs) --- + +func (h *TopologyHandler) handleAddNAT(msg *nats.Msg) { + if h.ovn == nil { + respond(msg, fmt.Errorf("OVN client not connected")) + return + } + + var evt NATEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.add-nat event", "err", err) + respond(msg, err) + return + } + + ctx := context.Background() + routerName := "vpc-" + evt.VpcId + + logicalPort := evt.PortName + externalMAC := evt.MAC + + natRule := &nbdb.NAT{ + Type: "dnat_and_snat", + ExternalIP: evt.ExternalIP, + LogicalIP: evt.LogicalIP, + LogicalPort: &logicalPort, + ExternalMAC: &externalMAC, + ExternalIDs: map[string]string{ + "spinifex:vpc_id": evt.VpcId, + "spinifex:public_ip": evt.ExternalIP, + }, + } + + if err := h.ovn.AddNAT(ctx, routerName, natRule); err != nil { + slog.Error("vpcd: failed to add dnat_and_snat rule", "router", routerName, "externalIP", evt.ExternalIP, "logicalIP", evt.LogicalIP, "err", err) + respond(msg, err) + return + } + + slog.Info("vpcd: added dnat_and_snat rule", + "router", routerName, + "external_ip", evt.ExternalIP, + "logical_ip", evt.LogicalIP, + "port", evt.PortName, + ) + respond(msg, nil) +} + +func (h *TopologyHandler) handleDeleteNAT(msg *nats.Msg) { + if h.ovn == nil { + respond(msg, fmt.Errorf("OVN client not connected")) + return + } + + var evt NATEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.delete-nat event", "err", err) + respond(msg, err) + return + } + + ctx := context.Background() + routerName := "vpc-" + evt.VpcId + + if err := h.ovn.DeleteNAT(ctx, routerName, "dnat_and_snat", evt.LogicalIP); err != nil { + slog.Error("vpcd: failed to delete dnat_and_snat rule", "router", routerName, "logicalIP", evt.LogicalIP, "err", err) + respond(msg, err) + return + } + + slog.Info("vpcd: deleted dnat_and_snat rule", + "router", routerName, + "logical_ip", evt.LogicalIP, + ) + respond(msg, nil) +} + // --- Helpers --- // subnetGateway computes the gateway IP (.1) from a CIDR string.