From 5481ef436109c6b0e0053ee37dd6ca9a96e4f7ec Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Fri, 20 Mar 2026 04:24:49 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=204=20=E2=80=94=20security=20grou?= =?UTF-8?q?ps=20(OVN=20ACLs=20+=20Port=20Groups)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add security group CRUD handlers: CreateSecurityGroup, DeleteSecurityGroup, DescribeSecurityGroups, AuthorizeSecurityGroupIngress/Egress, RevokeSecurityGroupIngress/Egress. SG records stored in NATS KV. Add ACL match expression compiler (acl.go) translating AWS SG rules to OVN ACL match strings for TCP/UDP port/ranges, ICMP, all-protocols, CIDR filters, and intra-SG references. Add vpcd security.go handlers for OVN Port Group lifecycle and ACL management via vpc.create-sg/delete-sg/update-sg NATS topics. Extend OVNClient interface with CreatePortGroup, DeletePortGroup, SetPortGroupPorts, AddACL, ClearACLs. Add PortGroup and ACL to nbdb model. --- spinifex/handlers/ec2/vpc/security_group.go | 607 ++++++++++++++++++++ spinifex/services/vpcd/acl.go | 99 ++++ spinifex/services/vpcd/mock_ovn.go | 82 +++ spinifex/services/vpcd/nbdb/model.go | 22 + spinifex/services/vpcd/nbdb/model_test.go | 2 + spinifex/services/vpcd/ovn.go | 32 ++ spinifex/services/vpcd/security.go | 167 ++++++ 7 files changed, 1011 insertions(+) create mode 100644 spinifex/handlers/ec2/vpc/security_group.go create mode 100644 spinifex/services/vpcd/acl.go create mode 100644 spinifex/services/vpcd/security.go diff --git a/spinifex/handlers/ec2/vpc/security_group.go b/spinifex/handlers/ec2/vpc/security_group.go new file mode 100644 index 000000000..a1d1018ef --- /dev/null +++ b/spinifex/handlers/ec2/vpc/security_group.go @@ -0,0 +1,607 @@ +package handlers_ec2_vpc + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +const ( + KVBucketSecurityGroups = "spinifex-vpc-security-groups" + KVBucketSecurityGroupsVersion = 1 +) + +// SGRule represents a single ingress or egress rule in a security group. +type SGRule struct { + IpProtocol string `json:"ip_protocol"` // "tcp", "udp", "icmp", "-1" (all) + FromPort int64 `json:"from_port"` + ToPort int64 `json:"to_port"` + CidrIp string `json:"cidr_ip,omitempty"` + SourceSG string `json:"source_sg,omitempty"` // Another SG ID for intra-SG rules +} + +// SecurityGroupRecord represents a stored security group. +type SecurityGroupRecord struct { + GroupId string `json:"group_id"` + GroupName string `json:"group_name"` + Description string `json:"description"` + VpcId string `json:"vpc_id"` + IngressRules []SGRule `json:"ingress_rules"` + EgressRules []SGRule `json:"egress_rules"` + Tags map[string]string `json:"tags"` + CreatedAt time.Time `json:"created_at"` +} + +// SGEvent is published on vpc.create-sg / vpc.delete-sg / vpc.update-sg for vpcd consumption. +type SGEvent struct { + GroupId string `json:"group_id"` + VpcId string `json:"vpc_id"` + IngressRules []SGRule `json:"ingress_rules,omitempty"` + EgressRules []SGRule `json:"egress_rules,omitempty"` +} + +// CreateSecurityGroup creates a new security group in a VPC. +func (s *VPCServiceImpl) CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error) { + if input.GroupName == nil || *input.GroupName == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + if input.VpcId == nil || *input.VpcId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + vpcId := *input.VpcId + groupName := *input.GroupName + + // Verify VPC exists + if _, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcId)); err != nil { + return nil, errors.New(awserrors.ErrorInvalidVpcIDNotFound) + } + + // Check for duplicate group name in the same VPC + prefix := accountID + "." + sgKeys, err := s.sgKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + return nil, errors.New(awserrors.ErrorServerInternal) + } + for _, k := range sgKeys { + if k == utils.VersionKey { + continue + } + if !strings.HasPrefix(k, prefix) { + continue + } + entry, err := s.sgKV.Get(k) + if err != nil { + continue + } + var existing SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &existing); err != nil { + continue + } + if existing.VpcId == vpcId && existing.GroupName == groupName { + return nil, errors.New(awserrors.ErrorInvalidGroupDuplicate) + } + } + + groupId := utils.GenerateResourceID("sg") + + description := "" + if input.Description != nil { + description = *input.Description + } + + // Default egress rule: allow all outbound traffic + defaultEgress := []SGRule{ + {IpProtocol: "-1", FromPort: 0, ToPort: 0, CidrIp: "0.0.0.0/0"}, + } + + record := SecurityGroupRecord{ + GroupId: groupId, + GroupName: groupName, + Description: description, + VpcId: vpcId, + IngressRules: []SGRule{}, + EgressRules: defaultEgress, + Tags: make(map[string]string), + CreatedAt: time.Now(), + } + + for _, tagSpec := range input.TagSpecifications { + if tagSpec.ResourceType != nil && *tagSpec.ResourceType == "security-group" { + for _, tag := range tagSpec.Tags { + if tag.Key != nil && tag.Value != nil { + record.Tags[*tag.Key] = *tag.Value + } + } + } + } + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal security group record: %w", err) + } + if _, err := s.sgKV.Put(utils.AccountKey(accountID, groupId), data); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("CreateSecurityGroup completed", "groupId", groupId, "groupName", groupName, "vpcId", vpcId, "accountID", accountID) + + // Publish vpc.create-sg event for vpcd + s.publishSGEvent("vpc.create-sg", SGEvent{ + GroupId: groupId, + VpcId: vpcId, + IngressRules: record.IngressRules, + EgressRules: record.EgressRules, + }) + + return &ec2.CreateSecurityGroupOutput{ + GroupId: aws.String(groupId), + }, nil +} + +// DeleteSecurityGroup deletes a security group. +func (s *VPCServiceImpl) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error) { + if input.GroupId == nil || *input.GroupId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + groupId := *input.GroupId + key := utils.AccountKey(accountID, groupId) + + entry, err := s.sgKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + if err := s.sgKV.Delete(key); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("DeleteSecurityGroup completed", "groupId", groupId, "accountID", accountID) + + // Publish vpc.delete-sg event for vpcd + s.publishSGEvent("vpc.delete-sg", SGEvent{ + GroupId: groupId, + VpcId: record.VpcId, + }) + + return &ec2.DeleteSecurityGroupOutput{}, nil +} + +// DescribeSecurityGroups lists security groups with optional filters. +func (s *VPCServiceImpl) DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) { + var groups []*ec2.SecurityGroup + + groupIDs := make(map[string]bool) + for _, id := range input.GroupIds { + if id != nil { + groupIDs[*id] = true + } + } + + // Extract VPC ID filter + vpcIDFilter := "" + groupNameFilter := "" + for _, f := range input.Filters { + if f.Name == nil || len(f.Values) == 0 || f.Values[0] == nil { + continue + } + switch *f.Name { + case "vpc-id": + vpcIDFilter = *f.Values[0] + case "group-name": + groupNameFilter = *f.Values[0] + } + } + + prefix := accountID + "." + keys, err := s.sgKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + for _, key := range keys { + if key == utils.VersionKey { + continue + } + if !strings.HasPrefix(key, prefix) { + continue + } + + entry, err := s.sgKV.Get(key) + if err != nil { + slog.Warn("Failed to get security group record", "key", key, "error", err) + continue + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + slog.Warn("Failed to unmarshal security group record", "key", key, "error", err) + continue + } + + if len(groupIDs) > 0 && !groupIDs[record.GroupId] { + continue + } + if vpcIDFilter != "" && record.VpcId != vpcIDFilter { + continue + } + if groupNameFilter != "" && record.GroupName != groupNameFilter { + continue + } + + groups = append(groups, s.sgRecordToEC2(&record, accountID)) + } + + // If specific group IDs were requested but not found, return error + if len(groupIDs) > 0 { + found := make(map[string]bool) + for _, sg := range groups { + if sg.GroupId != nil { + found[*sg.GroupId] = true + } + } + for id := range groupIDs { + if !found[id] { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + } + } + + slog.Info("DescribeSecurityGroups completed", "count", len(groups), "accountID", accountID) + + return &ec2.DescribeSecurityGroupsOutput{ + SecurityGroups: groups, + }, nil +} + +// AuthorizeSecurityGroupIngress adds ingress rules to a security group. +func (s *VPCServiceImpl) AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) { + if input.GroupId == nil || *input.GroupId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + groupId := *input.GroupId + key := utils.AccountKey(accountID, groupId) + + entry, err := s.sgKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Convert AWS IpPermissions to SGRules + newRules := ipPermissionsToSGRules(input.IpPermissions) + record.IngressRules = append(record.IngressRules, newRules...) + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal security group record: %w", err) + } + if _, err := s.sgKV.Update(key, data, entry.Revision()); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("AuthorizeSecurityGroupIngress completed", "groupId", groupId, "newRules", len(newRules), "accountID", accountID) + + // Publish vpc.update-sg event for vpcd + s.publishSGEvent("vpc.update-sg", SGEvent{ + GroupId: groupId, + VpcId: record.VpcId, + IngressRules: record.IngressRules, + EgressRules: record.EgressRules, + }) + + return &ec2.AuthorizeSecurityGroupIngressOutput{ + Return: aws.Bool(true), + }, nil +} + +// AuthorizeSecurityGroupEgress adds egress rules to a security group. +func (s *VPCServiceImpl) AuthorizeSecurityGroupEgress(input *ec2.AuthorizeSecurityGroupEgressInput, accountID string) (*ec2.AuthorizeSecurityGroupEgressOutput, error) { + if input.GroupId == nil || *input.GroupId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + groupId := *input.GroupId + key := utils.AccountKey(accountID, groupId) + + entry, err := s.sgKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + newRules := ipPermissionsToSGRules(input.IpPermissions) + record.EgressRules = append(record.EgressRules, newRules...) + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal security group record: %w", err) + } + if _, err := s.sgKV.Update(key, data, entry.Revision()); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("AuthorizeSecurityGroupEgress completed", "groupId", groupId, "newRules", len(newRules), "accountID", accountID) + + s.publishSGEvent("vpc.update-sg", SGEvent{ + GroupId: groupId, + VpcId: record.VpcId, + IngressRules: record.IngressRules, + EgressRules: record.EgressRules, + }) + + return &ec2.AuthorizeSecurityGroupEgressOutput{ + Return: aws.Bool(true), + }, nil +} + +// RevokeSecurityGroupIngress removes ingress rules from a security group. +func (s *VPCServiceImpl) RevokeSecurityGroupIngress(input *ec2.RevokeSecurityGroupIngressInput, accountID string) (*ec2.RevokeSecurityGroupIngressOutput, error) { + if input.GroupId == nil || *input.GroupId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + groupId := *input.GroupId + key := utils.AccountKey(accountID, groupId) + + entry, err := s.sgKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + revokeRules := ipPermissionsToSGRules(input.IpPermissions) + record.IngressRules = removeSGRules(record.IngressRules, revokeRules) + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal security group record: %w", err) + } + if _, err := s.sgKV.Update(key, data, entry.Revision()); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("RevokeSecurityGroupIngress completed", "groupId", groupId, "revokedRules", len(revokeRules), "accountID", accountID) + + s.publishSGEvent("vpc.update-sg", SGEvent{ + GroupId: groupId, + VpcId: record.VpcId, + IngressRules: record.IngressRules, + EgressRules: record.EgressRules, + }) + + return &ec2.RevokeSecurityGroupIngressOutput{ + Return: aws.Bool(true), + }, nil +} + +// RevokeSecurityGroupEgress removes egress rules from a security group. +func (s *VPCServiceImpl) RevokeSecurityGroupEgress(input *ec2.RevokeSecurityGroupEgressInput, accountID string) (*ec2.RevokeSecurityGroupEgressOutput, error) { + if input.GroupId == nil || *input.GroupId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + groupId := *input.GroupId + key := utils.AccountKey(accountID, groupId) + + entry, err := s.sgKV.Get(key) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidGroupNotFound) + } + + var record SecurityGroupRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + revokeRules := ipPermissionsToSGRules(input.IpPermissions) + record.EgressRules = removeSGRules(record.EgressRules, revokeRules) + + data, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal security group record: %w", err) + } + if _, err := s.sgKV.Update(key, data, entry.Revision()); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("RevokeSecurityGroupEgress completed", "groupId", groupId, "revokedRules", len(revokeRules), "accountID", accountID) + + s.publishSGEvent("vpc.update-sg", SGEvent{ + GroupId: groupId, + VpcId: record.VpcId, + IngressRules: record.IngressRules, + EgressRules: record.EgressRules, + }) + + return &ec2.RevokeSecurityGroupEgressOutput{ + Return: aws.Bool(true), + }, nil +} + +// sgRecordToEC2 converts a SecurityGroupRecord to an EC2 SecurityGroup. +func (s *VPCServiceImpl) sgRecordToEC2(record *SecurityGroupRecord, accountID string) *ec2.SecurityGroup { + sg := &ec2.SecurityGroup{ + GroupId: aws.String(record.GroupId), + GroupName: aws.String(record.GroupName), + Description: aws.String(record.Description), + VpcId: aws.String(record.VpcId), + OwnerId: aws.String(accountID), + IpPermissions: sgRulesToIpPermissions(record.IngressRules), + IpPermissionsEgress: sgRulesToIpPermissions(record.EgressRules), + } + + 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)}) + } + sg.Tags = tags + } + + return sg +} + +// ipPermissionsToSGRules converts AWS IpPermission slice to SGRule slice. +func ipPermissionsToSGRules(perms []*ec2.IpPermission) []SGRule { + var rules []SGRule + for _, perm := range perms { + if perm == nil { + continue + } + + proto := "-1" + if perm.IpProtocol != nil { + proto = *perm.IpProtocol + } + + var fromPort, toPort int64 + if perm.FromPort != nil { + fromPort = *perm.FromPort + } + if perm.ToPort != nil { + toPort = *perm.ToPort + } + + // One rule per CIDR range + for _, ipRange := range perm.IpRanges { + if ipRange.CidrIp == nil { + continue + } + rules = append(rules, SGRule{ + IpProtocol: proto, + FromPort: fromPort, + ToPort: toPort, + CidrIp: *ipRange.CidrIp, + }) + } + + // One rule per source security group + for _, pair := range perm.UserIdGroupPairs { + if pair.GroupId == nil { + continue + } + rules = append(rules, SGRule{ + IpProtocol: proto, + FromPort: fromPort, + ToPort: toPort, + SourceSG: *pair.GroupId, + }) + } + + // If no ranges and no groups specified, add a rule with no source filter + if len(perm.IpRanges) == 0 && len(perm.UserIdGroupPairs) == 0 { + rules = append(rules, SGRule{ + IpProtocol: proto, + FromPort: fromPort, + ToPort: toPort, + }) + } + } + return rules +} + +// sgRulesToIpPermissions converts SGRule slice to AWS IpPermission slice. +func sgRulesToIpPermissions(rules []SGRule) []*ec2.IpPermission { + // Group rules by protocol+port range + type permKey struct { + IpProtocol string + FromPort int64 + ToPort int64 + } + + grouped := make(map[permKey]*ec2.IpPermission) + for _, rule := range rules { + key := permKey{IpProtocol: rule.IpProtocol, FromPort: rule.FromPort, ToPort: rule.ToPort} + perm, exists := grouped[key] + if !exists { + perm = &ec2.IpPermission{ + IpProtocol: aws.String(rule.IpProtocol), + FromPort: aws.Int64(rule.FromPort), + ToPort: aws.Int64(rule.ToPort), + } + grouped[key] = perm + } + + if rule.CidrIp != "" { + perm.IpRanges = append(perm.IpRanges, &ec2.IpRange{ + CidrIp: aws.String(rule.CidrIp), + }) + } + if rule.SourceSG != "" { + perm.UserIdGroupPairs = append(perm.UserIdGroupPairs, &ec2.UserIdGroupPair{ + GroupId: aws.String(rule.SourceSG), + }) + } + } + + result := make([]*ec2.IpPermission, 0, len(grouped)) + for _, perm := range grouped { + result = append(result, perm) + } + return result +} + +// removeSGRules removes matching rules from the existing set. +func removeSGRules(existing, toRemove []SGRule) []SGRule { + removeSet := make(map[string]bool) + for _, r := range toRemove { + removeSet[sgRuleKey(r)] = true + } + + var result []SGRule + for _, r := range existing { + if !removeSet[sgRuleKey(r)] { + result = append(result, r) + } + } + return result +} + +// sgRuleKey returns a string key for deduplication/matching of SG rules. +func sgRuleKey(r SGRule) string { + return fmt.Sprintf("%s:%d:%d:%s:%s", r.IpProtocol, r.FromPort, r.ToPort, r.CidrIp, r.SourceSG) +} + +// publishSGEvent publishes a security group lifecycle event to NATS for vpcd consumption. +func (s *VPCServiceImpl) publishSGEvent(topic string, evt SGEvent) { + if s.natsConn == nil { + return + } + data, err := json.Marshal(evt) + if err != nil { + slog.Error("Failed to marshal SG event", "topic", topic, "err", err) + return + } + if err := s.natsConn.Publish(topic, data); err != nil { + slog.Error("Failed to publish SG event", "topic", topic, "err", err) + } +} diff --git a/spinifex/services/vpcd/acl.go b/spinifex/services/vpcd/acl.go new file mode 100644 index 000000000..1c868d85a --- /dev/null +++ b/spinifex/services/vpcd/acl.go @@ -0,0 +1,99 @@ +package vpcd + +import ( + "fmt" + "strings" +) + +// SGRuleForACL mirrors the SGRule from handlers for vpcd's use. +type SGRuleForACL struct { + IpProtocol string `json:"ip_protocol"` // "tcp", "udp", "icmp", "-1" (all) + FromPort int64 `json:"from_port"` + ToPort int64 `json:"to_port"` + CidrIp string `json:"cidr_ip,omitempty"` + SourceSG string `json:"source_sg,omitempty"` +} + +// portGroupName converts a security group ID to an OVN port group name. +// OVN port group names must match [a-zA-Z_][a-zA-Z0-9_]*, so hyphens +// in sg-xxx IDs are replaced with underscores. +func portGroupName(groupId string) string { + return strings.ReplaceAll(groupId, "-", "_") +} + +// BuildIngressACLMatch generates an OVN ACL match expression for an ingress rule. +// Ingress rules use "outport == @{pgName}" because OVN ACLs with direction "to-lport" +// match on the destination port (outport from the pipeline's perspective). +func BuildIngressACLMatch(pgName string, rule SGRuleForACL) string { + parts := []string{fmt.Sprintf("outport == @%s", pgName), "ip4"} + + // Protocol filter + parts = appendProtocolMatch(parts, rule) + + // Source filter + if rule.CidrIp != "" && rule.CidrIp != "0.0.0.0/0" { + parts = append(parts, fmt.Sprintf("ip4.src == %s", rule.CidrIp)) + } + if rule.SourceSG != "" { + sgPG := portGroupName(rule.SourceSG) + parts = append(parts, fmt.Sprintf("ip4.src == $%s_ip4", sgPG)) + } + + return strings.Join(parts, " && ") +} + +// BuildEgressACLMatch generates an OVN ACL match expression for an egress rule. +// Egress rules use "inport == @{pgName}" because OVN ACLs with direction "from-lport" +// match on the source port (inport from the pipeline's perspective). +func BuildEgressACLMatch(pgName string, rule SGRuleForACL) string { + parts := []string{fmt.Sprintf("inport == @%s", pgName), "ip4"} + + // Protocol filter + parts = appendProtocolMatch(parts, rule) + + // Destination filter + if rule.CidrIp != "" && rule.CidrIp != "0.0.0.0/0" { + parts = append(parts, fmt.Sprintf("ip4.dst == %s", rule.CidrIp)) + } + if rule.SourceSG != "" { + sgPG := portGroupName(rule.SourceSG) + parts = append(parts, fmt.Sprintf("ip4.dst == $%s_ip4", sgPG)) + } + + return strings.Join(parts, " && ") +} + +// appendProtocolMatch adds protocol-specific match clauses to the parts slice. +func appendProtocolMatch(parts []string, rule SGRuleForACL) []string { + switch rule.IpProtocol { + case "tcp": + parts = appendPortMatch(parts, "tcp", rule.FromPort, rule.ToPort) + case "udp": + parts = appendPortMatch(parts, "udp", rule.FromPort, rule.ToPort) + case "icmp": + parts = append(parts, "icmp4") + case "-1", "": + // All protocols — no additional filter needed beyond "ip4" + default: + // Unknown protocol — treat as all traffic + } + return parts +} + +// appendPortMatch adds TCP/UDP port match clauses. +func appendPortMatch(parts []string, proto string, fromPort, toPort int64) []string { + if fromPort == 0 && toPort == 0 { + // All ports for this protocol + parts = append(parts, proto) + return parts + } + if fromPort == toPort { + // Single port + parts = append(parts, fmt.Sprintf("%s.dst == %d", proto, fromPort)) + } else { + // Port range + parts = append(parts, fmt.Sprintf("%s.dst >= %d", proto, fromPort)) + parts = append(parts, fmt.Sprintf("%s.dst <= %d", proto, toPort)) + } + return parts +} diff --git a/spinifex/services/vpcd/mock_ovn.go b/spinifex/services/vpcd/mock_ovn.go index 310437d19..f08c6948f 100644 --- a/spinifex/services/vpcd/mock_ovn.go +++ b/spinifex/services/vpcd/mock_ovn.go @@ -21,6 +21,8 @@ type MockOVNClient struct { dhcpOpts map[string]*nbdb.DHCPOptions nats map[string]*nbdb.NAT // keyed by UUID staticRoutes map[string]*nbdb.LogicalRouterStaticRoute // keyed by UUID + portGroups map[string]*nbdb.PortGroup // keyed by name + acls map[string]*nbdb.ACL // keyed by UUID } // NewMockOVNClient creates a new MockOVNClient for testing. @@ -33,6 +35,8 @@ func NewMockOVNClient() *MockOVNClient { dhcpOpts: make(map[string]*nbdb.DHCPOptions), nats: make(map[string]*nbdb.NAT), staticRoutes: make(map[string]*nbdb.LogicalRouterStaticRoute), + portGroups: make(map[string]*nbdb.PortGroup), + acls: make(map[string]*nbdb.ACL), } } @@ -417,3 +421,81 @@ func (m *MockOVNClient) DeleteStaticRoute(_ context.Context, routerName string, delete(m.staticRoutes, foundUUID) return nil } + +// Port Groups + +func (m *MockOVNClient) CreatePortGroup(_ context.Context, name string, ports []string) error { + m.mu.Lock() + defer m.mu.Unlock() + if _, exists := m.portGroups[name]; exists { + return fmt.Errorf("port group %q already exists", name) + } + pg := &nbdb.PortGroup{ + UUID: utils.GenerateResourceID("pg"), + Name: name, + Ports: ports, + } + m.portGroups[name] = pg + return nil +} + +func (m *MockOVNClient) DeletePortGroup(_ context.Context, name string) error { + m.mu.Lock() + defer m.mu.Unlock() + if _, exists := m.portGroups[name]; !exists { + return fmt.Errorf("port group %q not found", name) + } + // Remove ACLs associated with this port group + pg := m.portGroups[name] + for _, aclUUID := range pg.ACLs { + delete(m.acls, aclUUID) + } + delete(m.portGroups, name) + return nil +} + +func (m *MockOVNClient) SetPortGroupPorts(_ context.Context, name string, ports []string) error { + m.mu.Lock() + defer m.mu.Unlock() + pg, exists := m.portGroups[name] + if !exists { + return fmt.Errorf("port group %q not found", name) + } + pg.Ports = ports + return nil +} + +// ACLs + +func (m *MockOVNClient) AddACL(_ context.Context, portGroupName string, direction string, priority int, match string, action string) error { + m.mu.Lock() + defer m.mu.Unlock() + pg, exists := m.portGroups[portGroupName] + if !exists { + return fmt.Errorf("port group %q not found", portGroupName) + } + acl := &nbdb.ACL{ + UUID: utils.GenerateResourceID("acl"), + Direction: direction, + Priority: priority, + Match: match, + Action: action, + } + m.acls[acl.UUID] = acl + pg.ACLs = append(pg.ACLs, acl.UUID) + return nil +} + +func (m *MockOVNClient) ClearACLs(_ context.Context, portGroupName string) error { + m.mu.Lock() + defer m.mu.Unlock() + pg, exists := m.portGroups[portGroupName] + if !exists { + return fmt.Errorf("port group %q not found", portGroupName) + } + for _, aclUUID := range pg.ACLs { + delete(m.acls, aclUUID) + } + pg.ACLs = nil + return nil +} diff --git a/spinifex/services/vpcd/nbdb/model.go b/spinifex/services/vpcd/nbdb/model.go index d2d374a6f..08551e848 100644 --- a/spinifex/services/vpcd/nbdb/model.go +++ b/spinifex/services/vpcd/nbdb/model.go @@ -91,6 +91,26 @@ type LogicalRouterStaticRoute struct { ExternalIDs map[string]string `ovsdb:"external_ids"` } +// PortGroup represents an OVN Port_Group for grouping logical switch ports +// (used for security group enforcement). +type PortGroup struct { + UUID string `ovsdb:"_uuid"` + Name string `ovsdb:"name"` + Ports []string `ovsdb:"ports"` // UUIDs of logical switch ports + ACLs []string `ovsdb:"acls"` // UUIDs of ACLs + ExternalIDs map[string]string `ovsdb:"external_ids"` +} + +// ACL represents an OVN ACL rule attached to a port group or logical switch. +type ACL struct { + UUID string `ovsdb:"_uuid"` + Direction string `ovsdb:"direction"` // "to-lport" or "from-lport" + Priority int `ovsdb:"priority"` + Match string `ovsdb:"match"` + Action string `ovsdb:"action"` // "allow-related", "drop", "allow", "reject" + ExternalIDs map[string]string `ovsdb:"external_ids"` +} + // GatewayChassis represents an OVN Gateway_Chassis for HA scheduling. type GatewayChassis struct { UUID string `ovsdb:"_uuid"` @@ -113,5 +133,7 @@ func FullDatabaseModel() (model.ClientDBModel, error) { "NAT": &NAT{}, "Logical_Router_Static_Route": &LogicalRouterStaticRoute{}, "Gateway_Chassis": &GatewayChassis{}, + "Port_Group": &PortGroup{}, + "ACL": &ACL{}, }) } diff --git a/spinifex/services/vpcd/nbdb/model_test.go b/spinifex/services/vpcd/nbdb/model_test.go index 62410e99a..af23d864a 100644 --- a/spinifex/services/vpcd/nbdb/model_test.go +++ b/spinifex/services/vpcd/nbdb/model_test.go @@ -23,6 +23,8 @@ func TestFullDatabaseModel(t *testing.T) { "NAT", "Logical_Router_Static_Route", "Gateway_Chassis", + "Port_Group", + "ACL", } types := dbModel.Types() diff --git a/spinifex/services/vpcd/ovn.go b/spinifex/services/vpcd/ovn.go index 9c74dcd85..8fef8c3ef 100644 --- a/spinifex/services/vpcd/ovn.go +++ b/spinifex/services/vpcd/ovn.go @@ -79,6 +79,15 @@ type OVNClient interface { // Static routes AddStaticRoute(ctx context.Context, routerName string, route *nbdb.LogicalRouterStaticRoute) error DeleteStaticRoute(ctx context.Context, routerName string, ipPrefix string) error + + // Port Groups (security group enforcement) + CreatePortGroup(ctx context.Context, name string, ports []string) error + DeletePortGroup(ctx context.Context, name string) error + SetPortGroupPorts(ctx context.Context, name string, ports []string) error + + // ACLs (attached to port groups) + AddACL(ctx context.Context, portGroupName string, direction string, priority int, match string, action string) error + ClearACLs(ctx context.Context, portGroupName string) error } // namedUUID generates a valid OVSDB named-uuid from a prefix and name. @@ -653,3 +662,26 @@ func (c *LiveOVNClient) DeleteStaticRoute(ctx context.Context, routerName string } return nil } + +// Port Group and ACL methods — stub implementations for LiveOVNClient. +// Full libovsdb implementation will come in a future phase. + +func (c *LiveOVNClient) CreatePortGroup(_ context.Context, _ string, _ []string) error { + return fmt.Errorf("CreatePortGroup: not yet implemented for live OVN client") +} + +func (c *LiveOVNClient) DeletePortGroup(_ context.Context, _ string) error { + return fmt.Errorf("DeletePortGroup: not yet implemented for live OVN client") +} + +func (c *LiveOVNClient) SetPortGroupPorts(_ context.Context, _ string, _ []string) error { + return fmt.Errorf("SetPortGroupPorts: not yet implemented for live OVN client") +} + +func (c *LiveOVNClient) AddACL(_ context.Context, _ string, _ string, _ int, _ string, _ string) error { + return fmt.Errorf("AddACL: not yet implemented for live OVN client") +} + +func (c *LiveOVNClient) ClearACLs(_ context.Context, _ string) error { + return fmt.Errorf("ClearACLs: not yet implemented for live OVN client") +} diff --git a/spinifex/services/vpcd/security.go b/spinifex/services/vpcd/security.go new file mode 100644 index 000000000..a8cd05ca2 --- /dev/null +++ b/spinifex/services/vpcd/security.go @@ -0,0 +1,167 @@ +package vpcd + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "github.com/nats-io/nats.go" +) + +// NATS topics for security group lifecycle events. +const ( + TopicCreateSG = "vpc.create-sg" + TopicDeleteSG = "vpc.delete-sg" + TopicUpdateSG = "vpc.update-sg" +) + +// SGEvent carries security group state from the handler to vpcd. +type SGEvent struct { + GroupId string `json:"group_id"` + VpcId string `json:"vpc_id"` + IngressRules []SGRuleForACL `json:"ingress_rules,omitempty"` + EgressRules []SGRuleForACL `json:"egress_rules,omitempty"` +} + +// handleCreateSG creates an OVN Port Group and initial ACLs for a new security group. +func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) { + if h.ovn == nil { + respond(msg, fmt.Errorf("OVN client not connected")) + return + } + + var evt SGEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.create-sg event", "err", err) + respond(msg, err) + return + } + + ctx := context.Background() + pgName := portGroupName(evt.GroupId) + + // Create port group (initially empty — ports are added when ENIs are assigned to the SG) + if err := h.ovn.CreatePortGroup(ctx, pgName, nil); err != nil { + slog.Error("vpcd: failed to create port group", "pg", pgName, "err", err) + respond(msg, err) + return + } + + // Add default deny ACL (priority 900) — drop all traffic not explicitly allowed + if err := h.ovn.AddACL(ctx, pgName, "to-lport", 900, fmt.Sprintf("outport == @%s && ip4", pgName), "drop"); err != nil { + slog.Warn("vpcd: failed to add default deny ingress ACL", "pg", pgName, "err", err) + } + if err := h.ovn.AddACL(ctx, pgName, "from-lport", 900, fmt.Sprintf("inport == @%s && ip4", pgName), "drop"); err != nil { + slog.Warn("vpcd: failed to add default deny egress ACL", "pg", pgName, "err", err) + } + + // Add ACLs for initial rules (priority 1000 — higher than deny) + h.addRuleACLs(ctx, pgName, evt.IngressRules, evt.EgressRules) + + slog.Info("vpcd: created security group port group", + "pg", pgName, + "group_id", evt.GroupId, + "vpc_id", evt.VpcId, + "ingress_rules", len(evt.IngressRules), + "egress_rules", len(evt.EgressRules), + ) + respond(msg, nil) +} + +// handleDeleteSG deletes the OVN Port Group and all associated ACLs. +func (h *TopologyHandler) handleDeleteSG(msg *nats.Msg) { + if h.ovn == nil { + respond(msg, fmt.Errorf("OVN client not connected")) + return + } + + var evt SGEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.delete-sg event", "err", err) + respond(msg, err) + return + } + + ctx := context.Background() + pgName := portGroupName(evt.GroupId) + + // Clear all ACLs before deleting the port group + if err := h.ovn.ClearACLs(ctx, pgName); err != nil { + slog.Warn("vpcd: failed to clear ACLs", "pg", pgName, "err", err) + } + + if err := h.ovn.DeletePortGroup(ctx, pgName); err != nil { + slog.Error("vpcd: failed to delete port group", "pg", pgName, "err", err) + respond(msg, err) + return + } + + slog.Info("vpcd: deleted security group port group", + "pg", pgName, + "group_id", evt.GroupId, + "vpc_id", evt.VpcId, + ) + respond(msg, nil) +} + +// handleUpdateSG replaces all ACLs for a security group with the current rule set. +func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) { + if h.ovn == nil { + respond(msg, fmt.Errorf("OVN client not connected")) + return + } + + var evt SGEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.update-sg event", "err", err) + respond(msg, err) + return + } + + ctx := context.Background() + pgName := portGroupName(evt.GroupId) + + // Clear existing ACLs + if err := h.ovn.ClearACLs(ctx, pgName); err != nil { + slog.Warn("vpcd: failed to clear ACLs for update", "pg", pgName, "err", err) + } + + // Re-add default deny ACLs (priority 900) + if err := h.ovn.AddACL(ctx, pgName, "to-lport", 900, fmt.Sprintf("outport == @%s && ip4", pgName), "drop"); err != nil { + slog.Warn("vpcd: failed to re-add default deny ingress ACL", "pg", pgName, "err", err) + } + if err := h.ovn.AddACL(ctx, pgName, "from-lport", 900, fmt.Sprintf("inport == @%s && ip4", pgName), "drop"); err != nil { + slog.Warn("vpcd: failed to re-add default deny egress ACL", "pg", pgName, "err", err) + } + + // Add ACLs for current rules + h.addRuleACLs(ctx, pgName, evt.IngressRules, evt.EgressRules) + + slog.Info("vpcd: updated security group ACLs", + "pg", pgName, + "group_id", evt.GroupId, + "vpc_id", evt.VpcId, + "ingress_rules", len(evt.IngressRules), + "egress_rules", len(evt.EgressRules), + ) + respond(msg, nil) +} + +// addRuleACLs adds OVN ACLs for a set of ingress and egress rules at priority 1000 +// (higher than the default deny at 900, so allow rules take precedence). +func (h *TopologyHandler) addRuleACLs(ctx context.Context, pgName string, ingress, egress []SGRuleForACL) { + for _, rule := range ingress { + match := BuildIngressACLMatch(pgName, rule) + if err := h.ovn.AddACL(ctx, pgName, "to-lport", 1000, match, "allow-related"); err != nil { + slog.Warn("vpcd: failed to add ingress ACL", "pg", pgName, "match", match, "err", err) + } + } + + for _, rule := range egress { + match := BuildEgressACLMatch(pgName, rule) + if err := h.ovn.AddACL(ctx, pgName, "from-lport", 1000, match, "allow-related"); err != nil { + slog.Warn("vpcd: failed to add egress ACL", "pg", pgName, "match", match, "err", err) + } + } +}