From 0b91dd908e2c80b5507fbe329629726a436ebe80 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Thu, 21 May 2026 21:36:03 +1000 Subject: [PATCH] refactor(vpcd): collapse SG subscribers onto network/{topology,policy} managers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 2.6 / mulga-siv-129 slice 2. Rewrites the vpc.create-sg / vpc.update-sg / vpc.delete-sg NATS subscribers in services/vpcd as thin wrappers over topology.Manager (port group lifecycle) and policy.SecurityGroupManager (ACL programming). Deletes the ~1000 lines of duplicated ACL builder and provisioning logic — its authoritative copy lives in network/policy. Surface additions: - ovn.ErrPortGroupNotFound sentinel; mock + live GetPortGroup wrap not-found via fmt.Errorf("%w: ...") so callers can use errors.Is. - topology.Manager grows EnsureSGPortGroup / DeleteSGPortGroup (resolves the apply.go TODO from §2.1 of the redesign plan). liveManager implements them with GetPortGroup-then-Create / GetPortGroup-then-Clear+Delete idempotency. Behaviour delta: the old provisionSG rolled back the OVN port group on AddACLs failure ("all-or-nothing"). The new path lets the port group survive and relies on network/reconcile to retry EnsureSG on its next sweep. The all-or-nothing fail-fast test cases are dropped; ACL emission semantics are covered by network/policy/sg_test.go. --- spinifex/network/ovn/client.go | 5 + spinifex/network/ovn/live.go | 2 +- spinifex/network/ovn/mock/mock.go | 5 +- spinifex/network/topology/manager.go | 6 + spinifex/network/topology/manager_live.go | 54 ++ .../network/topology/manager_live_test.go | 41 ++ spinifex/services/vpcd/acl.go | 110 ---- spinifex/services/vpcd/acl_test.go | 103 ---- spinifex/services/vpcd/security.go | 281 +++------- spinifex/services/vpcd/security_test.go | 523 ++++-------------- spinifex/services/vpcd/topology.go | 13 +- spinifex/services/vpcd/topology_manager.go | 21 + spinifex/services/vpcd/topology_test.go | 13 +- 13 files changed, 312 insertions(+), 865 deletions(-) delete mode 100644 spinifex/services/vpcd/acl.go delete mode 100644 spinifex/services/vpcd/acl_test.go diff --git a/spinifex/network/ovn/client.go b/spinifex/network/ovn/client.go index 1e3de03ef..8c7415f31 100644 --- a/spinifex/network/ovn/client.go +++ b/spinifex/network/ovn/client.go @@ -16,6 +16,11 @@ import ( // the same delete) match it with errors.Is and treat as success. var ErrNATNotFound = errors.New("NAT not found") +// ErrPortGroupNotFound is returned by GetPortGroup when the named port group +// is absent. EnsureSGPortGroup / DeleteSGPortGroup use errors.Is to make +// existence checks and delete-then-recheck flows idempotent. +var ErrPortGroupNotFound = errors.New("port group not found") + // ACLSpec describes an OVN ACL rule for attachment to a port group. // Name, Severity, and Meter are optional — set Severity when Log is true. type ACLSpec struct { diff --git a/spinifex/network/ovn/live.go b/spinifex/network/ovn/live.go index f51320bec..84699c32c 100644 --- a/spinifex/network/ovn/live.go +++ b/spinifex/network/ovn/live.go @@ -1068,7 +1068,7 @@ func (c *LiveClient) getPortGroup(ctx context.Context, name string) (*nbdb.PortG return nil, fmt.Errorf("get port group: %w", err) } if len(pgs) == 0 { - return nil, fmt.Errorf("port group %q not found", name) + return nil, fmt.Errorf("%w: %q", ErrPortGroupNotFound, name) } return &pgs[0], nil } diff --git a/spinifex/network/ovn/mock/mock.go b/spinifex/network/ovn/mock/mock.go index 01be0dc29..b7970ec46 100644 --- a/spinifex/network/ovn/mock/mock.go +++ b/spinifex/network/ovn/mock/mock.go @@ -700,13 +700,14 @@ func (m *Client) ListPortGroupsForPort(_ context.Context, lspName string) ([]str } // GetPortGroup returns the port group with the given name, or an error if it -// doesn't exist. Mirrors the live client. +// doesn't exist. Mirrors the live client; missing rows surface as +// ovn.ErrPortGroupNotFound so callers can use errors.Is for idempotent flows. func (m *Client) GetPortGroup(_ context.Context, name string) (*nbdb.PortGroup, error) { m.Mu.Lock() defer m.Mu.Unlock() pg, exists := m.PortGroups[name] if !exists { - return nil, fmt.Errorf("port group %q not found", name) + return nil, fmt.Errorf("%w: %q", ovn.ErrPortGroupNotFound, name) } return pg, nil } diff --git a/spinifex/network/topology/manager.go b/spinifex/network/topology/manager.go index db393dd29..a174b4593 100644 --- a/spinifex/network/topology/manager.go +++ b/spinifex/network/topology/manager.go @@ -34,6 +34,12 @@ type Manager interface { // memberships for a port. Manager computes the add/remove diff against // current OVN state. SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error + + // SG port-group lifecycle. ACL programming on a port group lives in + // network/policy.SecurityGroupManager — topology only owns the empty + // OVN port-group row keyed by SecurityGroupPortGroup(groupID). + EnsureSGPortGroup(ctx context.Context, groupID string) error + DeleteSGPortGroup(ctx context.Context, groupID string) error } // VPCSpec describes a VPC at L2. AZSlice is reserved for Phase 3 per-AZ CIDR diff --git a/spinifex/network/topology/manager_live.go b/spinifex/network/topology/manager_live.go index 405a8f680..c67c07cb1 100644 --- a/spinifex/network/topology/manager_live.go +++ b/spinifex/network/topology/manager_live.go @@ -2,6 +2,7 @@ package topology import ( "context" + "errors" "fmt" "log/slog" "net/netip" @@ -333,6 +334,59 @@ func (m *liveManager) DeletePort(ctx context.Context, spec PortSpec) error { return nil } +// EnsureSGPortGroup creates the empty OVN port-group row for a security +// group. Idempotent: a second call with the same groupID is a no-op. ACL +// programming lives in network/policy.SecurityGroupManager — callers must +// invoke it after EnsureSGPortGroup to attach the infrastructure and tenant +// rule sets. +func (m *liveManager) EnsureSGPortGroup(ctx context.Context, groupID string) error { + if m.ovn == nil { + return fmt.Errorf("OVN client not connected") + } + if groupID == "" { + return fmt.Errorf("EnsureSGPortGroup: empty groupID") + } + pgName := SecurityGroupPortGroup(groupID) + if _, err := m.ovn.GetPortGroup(ctx, pgName); err == nil { + return nil + } else if !errors.Is(err, ovn.ErrPortGroupNotFound) { + return fmt.Errorf("get port group %s: %w", pgName, err) + } + if err := m.ovn.CreatePortGroup(ctx, pgName, nil); err != nil { + return fmt.Errorf("create port group %s: %w", pgName, err) + } + slog.Info("topology: created SG port group", "pg", pgName, "group_id", groupID) + return nil +} + +// DeleteSGPortGroup clears every ACL on the SG's port group and removes the +// port group row. Idempotent: missing port group is treated as success. +// Reference-integrity dictates ClearACLs before DeletePortGroup — libovsdb +// rejects deleting a port group with dangling ACL references. +func (m *liveManager) DeleteSGPortGroup(ctx context.Context, groupID string) error { + if m.ovn == nil { + return fmt.Errorf("OVN client not connected") + } + if groupID == "" { + return fmt.Errorf("DeleteSGPortGroup: empty groupID") + } + pgName := SecurityGroupPortGroup(groupID) + if _, err := m.ovn.GetPortGroup(ctx, pgName); err != nil { + if errors.Is(err, ovn.ErrPortGroupNotFound) { + return nil + } + return fmt.Errorf("get port group %s: %w", pgName, err) + } + if err := m.ovn.ClearACLs(ctx, pgName); err != nil { + return fmt.Errorf("clear ACLs on %s: %w", pgName, err) + } + if err := m.ovn.DeletePortGroup(ctx, pgName); err != nil { + return fmt.Errorf("delete port group %s: %w", pgName, err) + } + slog.Info("topology: deleted SG port group", "pg", pgName, "group_id", groupID) + return nil +} + // SetPortSecurityGroups reconciles the port's port-group memberships against // the declared list. func (m *liveManager) SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error { diff --git a/spinifex/network/topology/manager_live_test.go b/spinifex/network/topology/manager_live_test.go index d9a12ff88..35471ab27 100644 --- a/spinifex/network/topology/manager_live_test.go +++ b/spinifex/network/topology/manager_live_test.go @@ -160,6 +160,47 @@ func TestLiveManager_DeleteSubnetAndVPC(t *testing.T) { } } +func TestLiveManager_EnsureSGPortGroup(t *testing.T) { + mgr, mockClient := newLiveManagerForTest(t) + ctx := context.Background() + + if err := mgr.EnsureSGPortGroup(ctx, "sg-pg1"); err != nil { + t.Fatalf("EnsureSGPortGroup: %v", err) + } + pg, err := mockClient.GetPortGroup(ctx, SecurityGroupPortGroup("sg-pg1")) + if err != nil { + t.Fatalf("port group not present: %v", err) + } + if pg.Name != SecurityGroupPortGroup("sg-pg1") { + t.Errorf("port group name mismatch: %q", pg.Name) + } + + // Idempotent: second call must not fail or churn state. + if err := mgr.EnsureSGPortGroup(ctx, "sg-pg1"); err != nil { + t.Fatalf("EnsureSGPortGroup second call: %v", err) + } +} + +func TestLiveManager_DeleteSGPortGroup(t *testing.T) { + mgr, mockClient := newLiveManagerForTest(t) + ctx := context.Background() + + if err := mgr.EnsureSGPortGroup(ctx, "sg-del1"); err != nil { + t.Fatalf("EnsureSGPortGroup: %v", err) + } + if err := mgr.DeleteSGPortGroup(ctx, "sg-del1"); err != nil { + t.Fatalf("DeleteSGPortGroup: %v", err) + } + if _, err := mockClient.GetPortGroup(ctx, SecurityGroupPortGroup("sg-del1")); err == nil { + t.Fatalf("port group still present after delete") + } + + // Idempotent on already-absent. + if err := mgr.DeleteSGPortGroup(ctx, "sg-del1"); err != nil { + t.Fatalf("DeleteSGPortGroup second call: %v", err) + } +} + func TestLiveManager_SetPortSecurityGroups(t *testing.T) { mgr, mockClient := newLiveManagerForTest(t) ctx := context.Background() diff --git a/spinifex/services/vpcd/acl.go b/spinifex/services/vpcd/acl.go deleted file mode 100644 index 5391688c7..000000000 --- a/spinifex/services/vpcd/acl.go +++ /dev/null @@ -1,110 +0,0 @@ -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, "-", "_") -} - -// addressSetName is the OVN address set holding the IPv4 addresses of every -// port that belongs to the given port group. SG-to-SG ACL match expressions -// reference it via "ip4.src == $" / "ip4.dst == $". Lifecycle is -// tied to the port group (created in handleCreateSG, deleted in handleDeleteSG); -// reconcilePortSGs maintains its membership in step with port-group joins. -func addressSetName(pgName string) string { - return pgName + "_ip4" -} - -// 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). -// -// Tenant-supplied CidrIp/SourceSG values MUST be validated by the handler before -// reaching this builder (see ec2/vpc/security_group.go ipPermissionsToSGRules). This -// builder interpolates them verbatim. -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 != "" { - parts = append(parts, fmt.Sprintf("ip4.src == $%s", addressSetName(portGroupName(rule.SourceSG)))) - } - - 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 != "" { - parts = append(parts, fmt.Sprintf("ip4.dst == $%s", addressSetName(portGroupName(rule.SourceSG)))) - } - - 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/acl_test.go b/spinifex/services/vpcd/acl_test.go deleted file mode 100644 index 6348c4d4d..000000000 --- a/spinifex/services/vpcd/acl_test.go +++ /dev/null @@ -1,103 +0,0 @@ -package vpcd - -import ( - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestACL_TCPPortFromCIDR(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "tcp", - FromPort: 22, - ToPort: 22, - CidrIp: "10.0.0.0/8", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "tcp.dst == 22") - assert.Contains(t, match, "ip4.src == 10.0.0.0/8") - assert.Contains(t, match, "outport == @sg_test") - assert.Contains(t, match, "ip4") -} - -func TestACL_AllTrafficFromSG(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "-1", - SourceSG: "sg-abc123", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "ip4.src == $sg_abc123_ip4") - assert.Contains(t, match, "outport == @sg_test") - assert.Contains(t, match, "ip4") -} - -func TestACL_PortRange(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "udp", - FromPort: 1024, - ToPort: 65535, - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "udp.dst >= 1024") - assert.Contains(t, match, "udp.dst <= 65535") -} - -func TestACL_ICMP(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "icmp", - CidrIp: "0.0.0.0/0", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "icmp4") - assert.NotContains(t, match, "tcp.dst") - assert.NotContains(t, match, "udp.dst") -} - -func TestACL_AllProtocols(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "-1", - CidrIp: "10.0.0.0/16", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "ip4") - assert.Contains(t, match, "ip4.src == 10.0.0.0/16") - assert.NotContains(t, match, "tcp") - assert.NotContains(t, match, "udp") - assert.NotContains(t, match, "icmp") -} - -func TestACL_EgressAll(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "-1", - CidrIp: "0.0.0.0/0", - } - match := BuildEgressACLMatch("sg_test", rule) - assert.Contains(t, match, "inport == @sg_test") - assert.NotContains(t, match, "outport") - assert.Contains(t, match, "ip4") -} - -func TestACL_TCPSinglePort(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "tcp", - FromPort: 443, - ToPort: 443, - CidrIp: "10.0.0.0/8", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "tcp.dst == 443") - assert.NotContains(t, match, "tcp.dst >=") - assert.NotContains(t, match, "tcp.dst <=") -} - -func TestACL_NoSource(t *testing.T) { - rule := SGRuleForACL{ - IpProtocol: "tcp", - FromPort: 80, - ToPort: 80, - CidrIp: "0.0.0.0/0", - } - match := BuildIngressACLMatch("sg_test", rule) - assert.Contains(t, match, "tcp.dst == 80") - assert.NotContains(t, match, "ip4.src") -} diff --git a/spinifex/services/vpcd/security.go b/spinifex/services/vpcd/security.go index 6424f1381..72903a608 100644 --- a/spinifex/services/vpcd/security.go +++ b/spinifex/services/vpcd/security.go @@ -1,58 +1,88 @@ package vpcd +// Phase 2.6 (mulga-siv-129) thin NATS wrappers for SG lifecycle. Each handler +// deserialises the event, translates to the policy/topology layer's spec +// types, and calls the manager. ACL semantics, OVN port-group lifecycle and +// the full all-or-nothing convergence model live in network/topology and +// network/policy; this file owns NATS plumbing only. + import ( "context" "encoding/json" "fmt" "log/slog" + "github.com/mulgadc/spinifex/spinifex/network/policy" "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" -) - -// denyACLSeverity is the syslog severity OVN uses when a packet matches a -// default-deny ACL. "info" is loud enough to be captured by a syslog forwarder -// yet quiet enough to avoid paging on every dropped scan packet. Operators can -// promote it at their log collector if they want higher-priority alerts. -const denyACLSeverity = "info" - -// 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"` +// sgRule mirrors the on-wire payload from handlers/ec2/vpc.SGRule. Kept local +// so vpcd does not import the handlers package. +type sgRule struct { + IpProtocol string `json:"ip_protocol"` + FromPort int64 `json:"from_port"` + ToPort int64 `json:"to_port"` + CidrIp string `json:"cidr_ip,omitempty"` + SourceSG string `json:"source_sg,omitempty"` } -// handleCreateSG creates an OVN Port Group and initial ACLs for a new security group. +// sgEvent mirrors handlers/ec2/vpc.SGEvent. +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"` +} + +func (e sgEvent) toSpec() policy.SGSpec { + return policy.SGSpec{ + GroupID: e.GroupId, + VPCID: e.VpcId, + IngressRules: toPolicyRules(e.IngressRules), + EgressRules: toPolicyRules(e.EgressRules), + } +} + +func toPolicyRules(in []sgRule) []policy.Rule { + out := make([]policy.Rule, len(in)) + for i, r := range in { + out[i] = policy.Rule{ + IPProtocol: r.IpProtocol, + FromPort: r.FromPort, + ToPort: r.ToPort, + CIDR: r.CidrIp, + SourceSG: r.SourceSG, + } + } + return out +} + +// handleCreateSG ensures the OVN port group exists then applies the ACL set +// for a new security group. Two-step (topology + policy) because port-group +// lifecycle is L2 and ACL programming is L3 — see plan §8.1. func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) { if h.ovn == nil { respond(msg, fmt.Errorf("OVN client not connected")) return } - - var evt SGEvent + 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() - if err := h.provisionSG(ctx, evt.GroupId, evt.IngressRules, evt.EgressRules); err != nil { - slog.Error("vpcd: failed to provision security group", "group_id", evt.GroupId, "err", err) + if err := h.EnsureSGPortGroup(ctx, evt.GroupId); err != nil { + slog.Error("vpcd: EnsureSGPortGroup failed", "group_id", evt.GroupId, "err", err) respond(msg, err) return } - - slog.Info("vpcd: created security group port group", - "pg", portGroupName(evt.GroupId), + if err := h.securityGroupManager().EnsureSG(ctx, evt.toSpec()); err != nil { + slog.Error("vpcd: EnsureSG failed", "group_id", evt.GroupId, "err", err) + respond(msg, err) + return + } + slog.Info("vpcd: created security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId, "ingress_rules", len(evt.IngressRules), @@ -61,129 +91,49 @@ func (h *TopologyHandler) handleCreateSG(msg *nats.Msg) { respond(msg, nil) } -// provisionSG creates the OVN port group, default-deny ACLs, and allow ACLs -// for the given SG. Used by handleCreateSG (CreateSecurityGroup path) and the -// reconciler's scan-2 (SG record without OVN port group). -// -// SG-to-SG match expressions reference `$_ip4` / `$_ip6` — these -// resolve against the SB Address_Set rows that ovn-northd auto-derives from -// each port group's port addresses (ovn-nb(5): "For each port group, there -// are two address sets generated to the Address_Set table of the -// OVN_Southbound database, ... with name being the name of the Port_Group -// followed by a suffix _ip4 / _ip6"). vpcd must NOT create explicit NB -// Address_Set rows with those names; that produces a duplicate sync to SB -// and wedges northd in an `OVNSB commit failed` retry loop. -// -// All-or-nothing: on any failure after the port group exists, the port group -// and any partial ACLs are torn down so the reconciler's scan-2 can recreate -// the SG from scratch on its next pass. A partial state would otherwise look -// healthy to scanMissingPortGroups (port group exists) while silently -// dropping legitimate traffic via the default-deny ACL. -func (h *TopologyHandler) provisionSG(ctx context.Context, groupId string, ingress, egress []SGRuleForACL) error { - pgName := portGroupName(groupId) - - if err := h.ovn.CreatePortGroup(ctx, pgName, nil); err != nil { - return fmt.Errorf("create port group %s: %w", pgName, err) - } - - done := false - defer func() { - if done { - return - } - if err := h.ovn.ClearACLs(ctx, pgName); err != nil { - slog.Warn("vpcd: provisionSG cleanup ClearACLs failed", "pg", pgName, "err", err) - } - if err := h.ovn.DeletePortGroup(ctx, pgName); err != nil { - slog.Warn("vpcd: provisionSG cleanup DeletePortGroup failed", "pg", pgName, "err", err) - } - }() - - // Default deny ACLs (priority 900, logged for CMMC SC.L1-3.13.1), - // infrastructure DHCP allows (priority 1050), and tenant allow rules - // (priority 1000) go in one OVSDB transaction so a 60-rule SG creation - // is one round-trip, not 64. - specs := append(infrastructureACLs(pgName), ruleACLSpecs(pgName, ingress, egress)...) - if err := h.ovn.AddACLs(ctx, pgName, specs); err != nil { - return fmt.Errorf("add ACLs on %s: %w", pgName, err) - } - done = true - return nil -} - -// handleDeleteSG deletes the OVN Port Group and all associated ACLs. +// handleDeleteSG removes the SG's port group (and its ACLs in one OVN +// transaction). Idempotent on already-absent. func (h *TopologyHandler) handleDeleteSG(msg *nats.Msg) { if h.ovn == nil { respond(msg, fmt.Errorf("OVN client not connected")) return } - - var evt SGEvent + 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. Fail-fast — leaving stale - // ACLs causes DeletePortGroup to be rejected by libovsdb (dangling ref). - if err := h.ovn.ClearACLs(ctx, pgName); err != nil { - slog.Error("vpcd: failed to clear ACLs", "pg", pgName, "err", err) + if err := h.DeleteSGPortGroup(ctx, evt.GroupId); err != nil { + slog.Error("vpcd: DeleteSGPortGroup failed", "group_id", evt.GroupId, "err", err) respond(msg, err) return } - - 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, - ) + slog.Info("vpcd: deleted security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId) respond(msg, nil) } -// handleUpdateSG replaces all ACLs for a security group with the current rule set. +// handleUpdateSG replaces the SG's ACL set with the rules from the event. +// Port-group lifecycle is unaffected — UpdateSG only touches ACLs. func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) { if h.ovn == nil { respond(msg, fmt.Errorf("OVN client not connected")) return } - - var evt SGEvent + 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.Error("vpcd: failed to clear ACLs for update", "pg", pgName, "err", err) + if err := h.securityGroupManager().UpdateSG(ctx, evt.toSpec()); err != nil { + slog.Error("vpcd: UpdateSG failed", "group_id", evt.GroupId, "err", err) respond(msg, err) return } - - // Re-add default deny + DHCP-infra allow ACLs + current rules in one transaction. - specs := append(infrastructureACLs(pgName), ruleACLSpecs(pgName, evt.IngressRules, evt.EgressRules)...) - if err := h.ovn.AddACLs(ctx, pgName, specs); err != nil { - slog.Error("vpcd: failed to add ACLs for update", "pg", pgName, "err", err) - respond(msg, err) - return - } - - slog.Info("vpcd: updated security group ACLs", - "pg", pgName, + slog.Info("vpcd: updated security group", "group_id", evt.GroupId, "vpc_id", evt.VpcId, "ingress_rules", len(evt.IngressRules), @@ -191,94 +141,3 @@ func (h *TopologyHandler) handleUpdateSG(msg *nats.Msg) { ) respond(msg, nil) } - -// ruleACLSpecs builds the priority-1000 allow ACL specs for the given rules. -// Allow rules are not logged — accept logging on a private network is high -// volume and low signal; only denies carry Log=true. -func ruleACLSpecs(pgName string, ingress, egress []SGRuleForACL) []ACLSpec { - specs := make([]ACLSpec, 0, len(ingress)+len(egress)) - for _, rule := range ingress { - specs = append(specs, ACLSpec{Direction: "to-lport", Priority: 1000, Match: BuildIngressACLMatch(pgName, rule), Action: "allow-related"}) - } - for _, rule := range egress { - specs = append(specs, ACLSpec{Direction: "from-lport", Priority: 1000, Match: BuildEgressACLMatch(pgName, rule), Action: "allow-related"}) - } - return specs -} - -// denyIngressACL builds the default-deny ingress ACL for a port group with -// logging enabled (CMMC SC.L1-3.13.1 boundary monitoring). -func denyIngressACL(pgName string) ACLSpec { - return ACLSpec{ - Direction: "to-lport", - Priority: 900, - Match: fmt.Sprintf("outport == @%s && ip4", pgName), - Action: "drop", - Name: pgName + "-deny-ingress", - Log: true, - Severity: denyACLSeverity, - } -} - -// denyEgressACL is the egress counterpart to denyIngressACL. -func denyEgressACL(pgName string) ACLSpec { - return ACLSpec{ - Direction: "from-lport", - Priority: 900, - Match: fmt.Sprintf("inport == @%s && ip4", pgName), - Action: "drop", - Name: pgName + "-deny-egress", - Log: true, - Severity: denyACLSeverity, - } -} - -// infrastructureACLs returns the platform-level ACLs that every port group -// carries regardless of tenant rules: the priority-900 default-deny pair and -// the priority-1050 DHCPv4 client/server allow pair. OVN's built-in DHCP -// responder runs inside the LS pipeline *after* the port-group ACL stage, so -// without these allows a narrow-egress tenant SG (e.g. egress restricted to -// the VPC CIDR) silently drops the guest's DHCPDISCOVER (dst=255.255.255.255) -// against the default-deny and the VM never gets an IPv4 lease. AWS users -// can't write a rule for the limited broadcast address, so the platform must -// allow DHCP itself — matching AWS's implicit behaviour. -// -// Priority 1050 sits above the default-deny (900) and above tenant rules -// (1000) so DHCP works regardless of tenant configuration; tenants in our -// model don't write ACLs at >1000, so there is no collision. -func infrastructureACLs(pgName string) []ACLSpec { - return []ACLSpec{ - denyIngressACL(pgName), - denyEgressACL(pgName), - dhcpEgressACL(pgName), - dhcpIngressACL(pgName), - } -} - -// dhcpEgressACL allows the guest's DHCP client traffic (DHCPDISCOVER, -// DHCPREQUEST: udp.src=68, udp.dst=67) out of the port. Action is plain -// "allow", not "allow-related" — DHCP is single-shot request/reply and -// allow-related on broadcast UDP interacts oddly with OVN's CT zones. -func dhcpEgressACL(pgName string) ACLSpec { - return ACLSpec{ - Direction: "from-lport", - Priority: 1050, - Match: fmt.Sprintf("inport == @%s && udp && udp.src == 68 && udp.dst == 67", pgName), - Action: "allow", - Name: pgName + "-allow-dhcp-egress", - } -} - -// dhcpIngressACL allows the OVN-generated DHCP server reply (DHCPOFFER, -// DHCPACK: udp.src=67, udp.dst=68) back to the guest. OVN's DHCP responder -// emits the reply from inside the LS pipeline, so without this matching -// to-lport allow the reply hits the priority-900 ingress default-deny. -func dhcpIngressACL(pgName string) ACLSpec { - return ACLSpec{ - Direction: "to-lport", - Priority: 1050, - Match: fmt.Sprintf("outport == @%s && udp && udp.src == 67 && udp.dst == 68", pgName), - Action: "allow", - Name: pgName + "-allow-dhcp-ingress", - } -} diff --git a/spinifex/services/vpcd/security_test.go b/spinifex/services/vpcd/security_test.go index 2ea7da6f1..142713769 100644 --- a/spinifex/services/vpcd/security_test.go +++ b/spinifex/services/vpcd/security_test.go @@ -1,5 +1,11 @@ package vpcd +// Phase 2.6 (mulga-siv-129) tests for the SG NATS wrappers in security.go. +// ACL emission semantics live in network/policy/sg_test.go + +// network/policy/acl_test.go; tests here cover the wrapper layer: +// event-shape errors, OVN-client wiring, and end-to-end NATS → OVN happy +// paths through the new managers. + import ( "context" "encoding/json" @@ -9,12 +15,16 @@ import ( "github.com/stretchr/testify/require" ) -func TestSecurity_CreatePortGroup(t *testing.T) { - _, nc := startTestNATS(t) +func sgTopoFixture(t *testing.T) (*TopologyHandler, *MockOVNClient) { + t.Helper() mock := NewMockOVNClient() _ = mock.Connect(context.Background()) + return NewTopologyHandler(mock), mock +} - topo := NewTopologyHandler(mock) +func TestSG_CreateSG_HappyPath(t *testing.T) { + _, nc := startTestNATS(t) + topo, mock := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -23,125 +33,23 @@ func TestSecurity_CreatePortGroup(t *testing.T) { } }() - evt := SGEvent{ - GroupId: "sg-abc123", - VpcId: "vpc-test1", - } + evt := sgEvent{GroupId: "sg-create1", VpcId: "vpc-test"} data, _ := json.Marshal(evt) resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) require.NoError(t, err) - assertSuccess(t, resp, "create SG port group") + assertSuccess(t, resp, "create SG") - // Verify port group was created in the mock mock.Mu.Lock() - pg, exists := mock.PortGroups["sg_abc123"] + _, pgExists := mock.PortGroups["sg_create1"] + aclCount := len(mock.ACLs) mock.Mu.Unlock() - assert.True(t, exists, "port group sg_abc123 should exist") - assert.NotNil(t, pg) - - // Verify the infrastructure ACL set was emitted: 2 default-deny at - // priority 900 (logged for CMMC SC.L1-3.13.1) plus 2 DHCP-infra allows - // at priority 1050 (unlogged — high-volume, low-signal). - type aclSnapshot struct { - action string - log bool - severity string - name string - priority int - direction string - match string - } - var snapshots []aclSnapshot - mock.Mu.Lock() - aclCount := len(pg.ACLs) - for _, aclUUID := range pg.ACLs { - a := mock.ACLs[aclUUID] - if a == nil { - continue - } - snap := aclSnapshot{ - action: a.Action, - log: a.Log, - priority: a.Priority, - direction: a.Direction, - match: a.Match, - } - if a.Severity != nil { - snap.severity = *a.Severity - } - if a.Name != nil { - snap.name = *a.Name - } - snapshots = append(snapshots, snap) - } - mock.Mu.Unlock() - - assert.Equal(t, 4, aclCount, "should have 4 infra ACLs (2 deny + 2 DHCP allow)") - denyCount, dhcpAllowCount := 0, 0 - for _, s := range snapshots { - switch s.action { - case "drop": - denyCount++ - assert.True(t, s.log, "default deny ACL must have log=true for boundary monitoring") - assert.Equal(t, "info", s.severity, "default deny ACL must use info severity") - assert.Contains(t, s.name, "sg_abc123-deny-", "default deny ACL must have a name for syslog correlation") - assert.Equal(t, 900, s.priority, "default deny ACL must be priority 900") - case "allow": - dhcpAllowCount++ - assert.False(t, s.log, "DHCP infra allow must not be logged (high-volume, low-signal)") - assert.Equal(t, 1050, s.priority, "DHCP infra allow must be priority 1050 (above tenant rules and default-deny)") - assert.Contains(t, s.name, "sg_abc123-allow-dhcp-", "DHCP allow must have a name for operator correlation") - } - } - assert.Equal(t, 2, denyCount, "should have 2 drop ACLs") - assert.Equal(t, 2, dhcpAllowCount, "should have 2 DHCP allow ACLs (egress client→server, ingress server→client)") + assert.True(t, pgExists, "port group sg_create1 must exist after create-sg") + assert.Equal(t, 4, aclCount, "infra ACL set (2 deny + 2 DHCP) must be attached") } -// TestSecurity_DHCPInfraACLs locks the exact shape of the DHCPv4 infra-allow -// ACLs that vpcd emits on every port group. This is the fix for the -// "narrow-egress SG silently breaks DHCP" bug: OVN's DHCP responder runs -// inside the LS pipeline *after* the port-group ACL stage, so without these -// the guest's DHCPDISCOVER (dst=255.255.255.255) hits the priority-900 -// default-deny under any non-0.0.0.0/0 egress rule and the VM never gets a -// lease. -func TestSecurity_DHCPInfraACLs(t *testing.T) { - const pg = "sg_dhcp_shape" - - egress := dhcpEgressACL(pg) - assert.Equal(t, "from-lport", egress.Direction) - assert.Equal(t, 1050, egress.Priority, "must outrank tenant rules (1000) and default-deny (900)") - assert.Equal(t, "allow", egress.Action, "plain allow — DHCP is single-shot, no conntrack") - assert.Equal(t, "inport == @sg_dhcp_shape && udp && udp.src == 68 && udp.dst == 67", egress.Match) - assert.Equal(t, "sg_dhcp_shape-allow-dhcp-egress", egress.Name) - assert.False(t, egress.Log) - - ingress := dhcpIngressACL(pg) - assert.Equal(t, "to-lport", ingress.Direction) - assert.Equal(t, 1050, ingress.Priority) - assert.Equal(t, "allow", ingress.Action) - assert.Equal(t, "outport == @sg_dhcp_shape && udp && udp.src == 67 && udp.dst == 68", ingress.Match) - assert.Equal(t, "sg_dhcp_shape-allow-dhcp-ingress", ingress.Name) - assert.False(t, ingress.Log) -} - -// TestSecurity_InfrastructureACLs_Order locks the deny-before-allow ordering -// that provisionSG/handleUpdateSG rely on when concatenating infrastructure -// ACLs with tenant rule ACLs. -func TestSecurity_InfrastructureACLs_Order(t *testing.T) { - specs := infrastructureACLs("sg_order") - require.Len(t, specs, 4) - assert.Equal(t, "sg_order-deny-ingress", specs[0].Name) - assert.Equal(t, "sg_order-deny-egress", specs[1].Name) - assert.Equal(t, "sg_order-allow-dhcp-egress", specs[2].Name) - assert.Equal(t, "sg_order-allow-dhcp-ingress", specs[3].Name) -} - -func TestSecurity_DeletePortGroup(t *testing.T) { +func TestSG_CreateSG_Idempotent(t *testing.T) { _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - - topo := NewTopologyHandler(mock) + topo, mock := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -150,39 +58,51 @@ func TestSecurity_DeletePortGroup(t *testing.T) { } }() - // Create first - createEvt := SGEvent{GroupId: "sg-del1", VpcId: "vpc-test2"} - data, _ := json.Marshal(createEvt) + data, _ := json.Marshal(sgEvent{GroupId: "sg-idem", VpcId: "vpc-test"}) + for range 2 { + resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) + require.NoError(t, err) + assertSuccess(t, resp, "create SG retry") + } + + mock.Mu.Lock() + pgCount := len(mock.PortGroups) + mock.Mu.Unlock() + assert.Equal(t, 1, pgCount, "duplicate create-sg must not create a second port group") +} + +func TestSG_DeleteSG_HappyPath(t *testing.T) { + _, nc := startTestNATS(t) + topo, mock := sgTopoFixture(t) + subs, err := topo.Subscribe(nc) + require.NoError(t, err) + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + }() + + data, _ := json.Marshal(sgEvent{GroupId: "sg-del1", VpcId: "vpc-test"}) resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) require.NoError(t, err) - assertSuccess(t, resp, "create SG for delete test") + assertSuccess(t, resp, "create SG before delete") - // Verify it exists - mock.Mu.Lock() - _, exists := mock.PortGroups["sg_del1"] - mock.Mu.Unlock() - assert.True(t, exists) - - // Delete - delEvt := SGEvent{GroupId: "sg-del1", VpcId: "vpc-test2"} - data, _ = json.Marshal(delEvt) resp, err = nc.Request(TopicDeleteSG, data, 5_000_000_000) require.NoError(t, err) - assertSuccess(t, resp, "delete SG port group") + assertSuccess(t, resp, "delete SG") - // Verify removed mock.Mu.Lock() - _, exists = mock.PortGroups["sg_del1"] + _, pgExists := mock.PortGroups["sg_del1"] mock.Mu.Unlock() - assert.False(t, exists, "port group sg_del1 should be deleted") + assert.False(t, pgExists, "port group must be gone after delete-sg") } -func TestSecurity_UpdateSGAddRules(t *testing.T) { +// Delete of a never-created SG must succeed: the new wrapper delegates to +// topology.Manager.DeleteSGPortGroup which is idempotent on absent port +// groups (the reconciler retries delete events that race with vm-cleanup). +func TestSG_DeleteSG_IdempotentOnAbsent(t *testing.T) { _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - - topo := NewTopologyHandler(mock) + topo, _ := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -191,83 +111,15 @@ func TestSecurity_UpdateSGAddRules(t *testing.T) { } }() - // Create SG first (no rules) - createEvt := SGEvent{GroupId: "sg-upd1", VpcId: "vpc-test3"} - data, _ := json.Marshal(createEvt) - resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) + data, _ := json.Marshal(sgEvent{GroupId: "sg-ghost", VpcId: "vpc-test"}) + resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000) require.NoError(t, err) - assertSuccess(t, resp, "create SG for update test") - - // Update with ingress rules - updateEvt := SGEvent{ - GroupId: "sg-upd1", - VpcId: "vpc-test3", - IngressRules: []SGRuleForACL{ - {IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"}, - {IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"}, - }, - } - data, _ = json.Marshal(updateEvt) - resp, err = nc.Request(TopicUpdateSG, data, 5_000_000_000) - require.NoError(t, err) - assertSuccess(t, resp, "update SG with ingress rules") - - // Verify ACLs were created: 2 default deny + 2 ingress allow = 4 - type aclSnapshot struct { - action string - match string - log bool - } - var snapshots []aclSnapshot - mock.Mu.Lock() - pg := mock.PortGroups["sg_upd1"] - aclCount := len(pg.ACLs) - for _, aclUUID := range pg.ACLs { - a := mock.ACLs[aclUUID] - if a == nil { - continue - } - snapshots = append(snapshots, aclSnapshot{action: a.Action, match: a.Match, log: a.Log}) - } - mock.Mu.Unlock() - - assert.Equal(t, 6, aclCount, "should have 6 ACLs (2 deny + 2 DHCP infra allow + 2 ingress allow)") - - // Check that at least one match contains tcp.dst == 22. Also verify - // logging policy: denies logged, allows not logged (CMMC SC.L1-3.13.1). - foundSSH := false - foundHTTPS := false - for _, s := range snapshots { - switch s.action { - case "drop": - assert.True(t, s.log, "deny ACL must be logged") - case "allow-related": - assert.False(t, s.log, "allow ACL must not be logged (high volume, low signal)") - } - if containsAll(s.match, "tcp.dst == 22", "ip4.src == 10.0.0.0/8") { - foundSSH = true - } - if containsAll(s.match, "tcp.dst == 443") { - foundHTTPS = true - } - } - assert.True(t, foundSSH, "should have SSH ACL with source CIDR") - assert.True(t, foundHTTPS, "should have HTTPS ACL") + assertSuccess(t, resp, "delete absent SG must be a no-op") } -// TestSecurity_CreateSG_FailsFastOnAddACLError locks the Phase 1.1 invariant -// that handleCreateSG returns an AddACLs error to the caller and tears down -// the half-built SG. The mock fails the batched AddACLs call, and the test -// asserts the handler responds with failure AND that the port group is -// gone — so the next reconciler scan recreates from scratch instead of -// observing a half-built PG with no ACLs. -func TestSecurity_CreateSG_FailsFastOnAddACLError(t *testing.T) { +func TestSG_UpdateSG_ReplacesACLSet(t *testing.T) { _, nc := startTestNATS(t) - mock := NewMockOVNClient() - mock.AddACLErrAfter = 1 - _ = mock.Connect(context.Background()) - - topo := NewTopologyHandler(mock) + topo, mock := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -276,103 +128,40 @@ func TestSecurity_CreateSG_FailsFastOnAddACLError(t *testing.T) { } }() - evt := SGEvent{GroupId: "sg-failacl", VpcId: "vpc-failacl"} - data, _ := json.Marshal(evt) - resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) - require.NoError(t, err) - assertFailure(t, resp, "create SG must fail when AddACL fails") - - mock.Mu.Lock() - _, pgExists := mock.PortGroups["sg_failacl"] - mock.Mu.Unlock() - assert.False(t, pgExists, "port group must be torn down on failed provisionSG") -} - -// TestSecurity_UpdateSG_FailsFastOnAddACLError locks the same fail-fast policy -// in handleUpdateSG: clear ACLs, re-add deny ACLs, then add allow ACLs — any -// failure must propagate to the caller instead of being slog.Warn-ed away. -func TestSecurity_UpdateSG_FailsFastOnAddACLError(t *testing.T) { - _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - - topo := NewTopologyHandler(mock) - subs, err := topo.Subscribe(nc) - require.NoError(t, err) - defer func() { - for _, s := range subs { - _ = s.Unsubscribe() - } - }() - - createEvt := SGEvent{GroupId: "sg-updfail", VpcId: "vpc-updfail"} - data, _ := json.Marshal(createEvt) - resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) + createData, _ := json.Marshal(sgEvent{GroupId: "sg-upd", VpcId: "vpc-test"}) + resp, err := nc.Request(TopicCreateSG, createData, 5_000_000_000) require.NoError(t, err) assertSuccess(t, resp, "create SG before update") - mock.Mu.Lock() - mock.AddACLCalls = 0 - mock.AddACLErrAfter = 1 - mock.Mu.Unlock() - - updateEvt := SGEvent{ - GroupId: "sg-updfail", - VpcId: "vpc-updfail", - IngressRules: []SGRuleForACL{ + updateData, _ := json.Marshal(sgEvent{ + GroupId: "sg-upd", + VpcId: "vpc-test", + IngressRules: []sgRule{ {IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"}, - {IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"}, }, - } - data, _ = json.Marshal(updateEvt) - resp, err = nc.Request(TopicUpdateSG, data, 5_000_000_000) + }) + resp, err = nc.Request(TopicUpdateSG, updateData, 5_000_000_000) require.NoError(t, err) - assertFailure(t, resp, "update SG must fail when AddACL fails mid-batch") -} + assertSuccess(t, resp, "update SG") -// TestSecurity_CreateSG_WithRules_FailsFastOnAddACLError locks the fail-fast -// policy when handleCreateSG carries allow rules: the batched AddACLs call -// includes both deny ACLs and rule ACLs, and a failure must still tear down -// the PG/AS so the reconciler doesn't mistake a half-built PG as healthy. -func TestSecurity_CreateSG_WithRules_FailsFastOnAddACLError(t *testing.T) { - _, nc := startTestNATS(t) - mock := NewMockOVNClient() - mock.AddACLErrAfter = 1 - _ = mock.Connect(context.Background()) - - topo := NewTopologyHandler(mock) - subs, err := topo.Subscribe(nc) - require.NoError(t, err) - defer func() { - for _, s := range subs { - _ = s.Unsubscribe() + mock.Mu.Lock() + pg := mock.PortGroups["sg_upd"] + aclCount := 0 + hasTenantRule := false + if pg != nil { + aclCount = len(pg.ACLs) + for _, uuid := range pg.ACLs { + if acl := mock.ACLs[uuid]; acl != nil && acl.Priority == 1000 { + hasTenantRule = true + } } - }() - - evt := SGEvent{ - GroupId: "sg-rulefail", - VpcId: "vpc-rulefail", - IngressRules: []SGRuleForACL{ - {IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/8"}, - {IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"}, - }, } - data, _ := json.Marshal(evt) - resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) - require.NoError(t, err) - assertFailure(t, resp, "create SG with rules must fail when AddACLs fails") - - mock.Mu.Lock() - _, pgExists := mock.PortGroups["sg_rulefail"] mock.Mu.Unlock() - assert.False(t, pgExists, "port group must be torn down when AddACLs fails") + assert.Equal(t, 5, aclCount, "update must leave 4 infra + 1 tenant ACL on the port group") + assert.True(t, hasTenantRule, "tenant priority-1000 ACL must be present after update") } -// TestSecurity_HandleCreateSG_OVNNotConnected, ...HandleDeleteSG_..., and -// ...HandleUpdateSG_... lock the defensive guard at the top of each handler: -// when vpcd's OVN client failed to connect (or has been Closed), the SG event -// must surface a clear error to the caller instead of nil-deref'ing. -func TestSecurity_HandleCreateSG_OVNNotConnected(t *testing.T) { +func TestSG_HandleCreateSG_OVNNotConnected(t *testing.T) { _, nc := startTestNATS(t) topo := NewTopologyHandler(nil) subs, err := topo.Subscribe(nc) @@ -383,13 +172,13 @@ func TestSecurity_HandleCreateSG_OVNNotConnected(t *testing.T) { } }() - data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"}) + data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"}) resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) require.NoError(t, err) assertFailure(t, resp, "create SG must fail when ovn client is nil") } -func TestSecurity_HandleDeleteSG_OVNNotConnected(t *testing.T) { +func TestSG_HandleDeleteSG_OVNNotConnected(t *testing.T) { _, nc := startTestNATS(t) topo := NewTopologyHandler(nil) subs, err := topo.Subscribe(nc) @@ -400,13 +189,13 @@ func TestSecurity_HandleDeleteSG_OVNNotConnected(t *testing.T) { } }() - data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"}) + data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"}) resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000) require.NoError(t, err) assertFailure(t, resp, "delete SG must fail when ovn client is nil") } -func TestSecurity_HandleUpdateSG_OVNNotConnected(t *testing.T) { +func TestSG_HandleUpdateSG_OVNNotConnected(t *testing.T) { _, nc := startTestNATS(t) topo := NewTopologyHandler(nil) subs, err := topo.Subscribe(nc) @@ -417,20 +206,15 @@ func TestSecurity_HandleUpdateSG_OVNNotConnected(t *testing.T) { } }() - data, _ := json.Marshal(SGEvent{GroupId: "sg-noovn", VpcId: "vpc-noovn"}) + data, _ := json.Marshal(sgEvent{GroupId: "sg-noovn"}) resp, err := nc.Request(TopicUpdateSG, data, 5_000_000_000) require.NoError(t, err) assertFailure(t, resp, "update SG must fail when ovn client is nil") } -// TestSecurity_HandleCreateSG_BadJSON / ..._HandleDeleteSG_BadJSON / -// ..._HandleUpdateSG_BadJSON: malformed event payloads must surface as -// failures, not be silently dropped — the handler-side caller relies on the -// ack to know the request was processed. -func TestSecurity_HandleCreateSG_BadJSON(t *testing.T) { +func TestSG_HandleCreateSG_BadJSON(t *testing.T) { _, nc := startTestNATS(t) - topo := NewTopologyHandler(NewMockOVNClient()) - _ = topo.ovn.Connect(context.Background()) + topo, _ := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -444,10 +228,9 @@ func TestSecurity_HandleCreateSG_BadJSON(t *testing.T) { assertFailure(t, resp, "create SG with malformed payload must fail") } -func TestSecurity_HandleDeleteSG_BadJSON(t *testing.T) { +func TestSG_HandleDeleteSG_BadJSON(t *testing.T) { _, nc := startTestNATS(t) - topo := NewTopologyHandler(NewMockOVNClient()) - _ = topo.ovn.Connect(context.Background()) + topo, _ := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -461,10 +244,9 @@ func TestSecurity_HandleDeleteSG_BadJSON(t *testing.T) { assertFailure(t, resp, "delete SG with malformed payload must fail") } -func TestSecurity_HandleUpdateSG_BadJSON(t *testing.T) { +func TestSG_HandleUpdateSG_BadJSON(t *testing.T) { _, nc := startTestNATS(t) - topo := NewTopologyHandler(NewMockOVNClient()) - _ = topo.ovn.Connect(context.Background()) + topo, _ := sgTopoFixture(t) subs, err := topo.Subscribe(nc) require.NoError(t, err) defer func() { @@ -477,122 +259,3 @@ func TestSecurity_HandleUpdateSG_BadJSON(t *testing.T) { require.NoError(t, err) assertFailure(t, resp, "update SG with malformed payload must fail") } - -// TestSecurity_HandleDeleteSG_FailsFastOnClearACLsError: handleDeleteSG must -// surface the ClearACLs error rather than swallow it. We trigger this by -// requesting a delete for an SG that was never created — the mock's ClearACLs -// returns "port group not found", which must propagate as a failed response. -func TestSecurity_HandleDeleteSG_FailsFastOnClearACLsError(t *testing.T) { - _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - topo := NewTopologyHandler(mock) - subs, err := topo.Subscribe(nc) - require.NoError(t, err) - defer func() { - for _, s := range subs { - _ = s.Unsubscribe() - } - }() - - data, _ := json.Marshal(SGEvent{GroupId: "sg-never-created", VpcId: "vpc-x"}) - resp, err := nc.Request(TopicDeleteSG, data, 5_000_000_000) - require.NoError(t, err) - assertFailure(t, resp, "delete must fail when port group does not exist") -} - -// TestSecurity_HandleUpdateSG_FailsFastOnClearACLsError: handleUpdateSG must -// also surface the ClearACLs error — silently swallowing it would leave the -// caller thinking the rule update succeeded when in fact the old ACLs are -// still in place. -func TestSecurity_HandleUpdateSG_FailsFastOnClearACLsError(t *testing.T) { - _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - topo := NewTopologyHandler(mock) - subs, err := topo.Subscribe(nc) - require.NoError(t, err) - defer func() { - for _, s := range subs { - _ = s.Unsubscribe() - } - }() - - data, _ := json.Marshal(SGEvent{ - GroupId: "sg-never-created", - VpcId: "vpc-x", - IngressRules: []SGRuleForACL{{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "0.0.0.0/0"}}, - }) - resp, err := nc.Request(TopicUpdateSG, data, 5_000_000_000) - require.NoError(t, err) - assertFailure(t, resp, "update must fail when port group does not exist") -} - -// TestSecurity_HandleDeleteSG_Idempotent_AddressSetMissing: if the address set -// has already been cleaned up out-of-band but the port group still exists, the -// delete must NOT propagate the AS-not-found error — the goal state is "both -// gone", and we've already achieved it for the AS. Locks the matching -// fail-fast invariant in TestSecurity_DeleteSG_FailsFastOnAddressSetError -// (already covered) by exercising the inverse direction: a fresh delete after -// a successful create must succeed end-to-end without errors leaking. -// -// (kept here to avoid duplicating the create→delete fixture; the existing -// happy-path test exercises the success branch but does not assert that -// neither half-state lingers.) -func TestSecurity_HandleDeleteSG_LeavesNoResidualState(t *testing.T) { - _, nc := startTestNATS(t) - mock := NewMockOVNClient() - _ = mock.Connect(context.Background()) - topo := NewTopologyHandler(mock) - subs, err := topo.Subscribe(nc) - require.NoError(t, err) - defer func() { - for _, s := range subs { - _ = s.Unsubscribe() - } - }() - - data, _ := json.Marshal(SGEvent{GroupId: "sg-residual", VpcId: "vpc-x"}) - resp, err := nc.Request(TopicCreateSG, data, 5_000_000_000) - require.NoError(t, err) - assertSuccess(t, resp, "create SG") - - resp, err = nc.Request(TopicDeleteSG, data, 5_000_000_000) - require.NoError(t, err) - assertSuccess(t, resp, "delete SG") - - mock.Mu.Lock() - _, pgExists := mock.PortGroups["sg_residual"] - leftoverACLs := 0 - for _, a := range mock.ACLs { - if a != nil && (a.Match == "outport == @sg_residual && ip4" || a.Match == "inport == @sg_residual && ip4") { - leftoverACLs++ - } - } - mock.Mu.Unlock() - assert.False(t, pgExists, "port group must be gone") - assert.Zero(t, leftoverACLs, "no ACL rows must reference the deleted PG") -} - -// containsAll checks if s contains all substrings. -func containsAll(s string, subs ...string) bool { - for _, sub := range subs { - if !contains(s, sub) { - return false - } - } - return true -} - -func contains(s, sub string) bool { - return len(s) >= len(sub) && searchString(s, sub) -} - -func searchString(s, sub string) bool { - for i := 0; i <= len(s)-len(sub); i++ { - if s[i:i+len(sub)] == sub { - return true - } - } - return false -} diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index a79f4418e..d6eacef5d 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -12,6 +12,7 @@ import ( "sync" "time" + "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" @@ -69,6 +70,9 @@ const ( TopicDeleteNAT = "vpc.delete-nat" TopicAddNATGateway = "vpc.add-nat-gateway" TopicDeleteNATGateway = "vpc.delete-nat-gateway" + TopicCreateSG = "vpc.create-sg" + TopicDeleteSG = "vpc.delete-sg" + TopicUpdateSG = "vpc.update-sg" ) // gatewayPortNetwork is the link-local CIDR every IGW gateway-LRP carries @@ -179,6 +183,11 @@ type TopologyHandler struct { // adapter call doesn't re-construct (Phase 2.6, mulga-siv-129). lmOnce sync.Once lm topology.Manager + + // sgm + sgmOnce cache the policy.SecurityGroupManager backed by h.ovn so + // SG subscribers reuse one manager instance across events. + sgmOnce sync.Once + sgm policy.SecurityGroupManager } // NewTopologyHandler creates a new TopologyHandler with optional external network config. @@ -841,7 +850,7 @@ func (h *TopologyHandler) handleCreatePort(msg *nats.Msg) { // port group's port addresses once the LSP joins. pgNames := make([]string, 0, len(evt.SecurityGroupIds)) for _, sgId := range evt.SecurityGroupIds { - pgNames = append(pgNames, portGroupName(sgId)) + pgNames = append(pgNames, topology.SecurityGroupPortGroup(sgId)) } if err := h.ovn.CreateLogicalSwitchPortInGroups(ctx, switchName, lsp, pgNames); err != nil { slog.Error("vpcd: failed to create logical switch port", "port", portName, "switch", switchName, "err", err) @@ -959,7 +968,7 @@ func (h *TopologyHandler) handleUpdatePortSGs(msg *nats.Msg) { func (h *TopologyHandler) reconcilePortSGs(ctx context.Context, lspName string, desiredSGs []string) (bool, error) { desired := make(map[string]struct{}, len(desiredSGs)) for _, sgId := range desiredSGs { - desired[portGroupName(sgId)] = struct{}{} + desired[topology.SecurityGroupPortGroup(sgId)] = struct{}{} } currentNames, err := h.ovn.ListPortGroupsForPort(ctx, lspName) diff --git a/spinifex/services/vpcd/topology_manager.go b/spinifex/services/vpcd/topology_manager.go index 4e1330bd2..a935b9220 100644 --- a/spinifex/services/vpcd/topology_manager.go +++ b/spinifex/services/vpcd/topology_manager.go @@ -14,6 +14,7 @@ import ( "context" "strings" + "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/network/topology" ) @@ -77,3 +78,23 @@ func (h *TopologyHandler) DeletePort(ctx context.Context, spec topology.PortSpec func (h *TopologyHandler) SetPortSecurityGroups(ctx context.Context, portID string, sgIDs []string) error { return h.topologyManager().SetPortSecurityGroups(ctx, portID, sgIDs) } + +// EnsureSGPortGroup creates the OVN port group for a security group. +func (h *TopologyHandler) EnsureSGPortGroup(ctx context.Context, groupID string) error { + return h.topologyManager().EnsureSGPortGroup(ctx, groupID) +} + +// DeleteSGPortGroup removes the OVN port group for a security group. +func (h *TopologyHandler) DeleteSGPortGroup(ctx context.Context, groupID string) error { + return h.topologyManager().DeleteSGPortGroup(ctx, groupID) +} + +// securityGroupManager returns the lazily-constructed policy SG manager +// backed by the same OVN client as the live topology manager. Subscribers +// use it to apply ACL sets against the SG's port group. +func (h *TopologyHandler) securityGroupManager() policy.SecurityGroupManager { + h.sgmOnce.Do(func() { + h.sgm = policy.NewSecurityGroupManager(h.ovn) + }) + return h.sgm +} diff --git a/spinifex/services/vpcd/topology_test.go b/spinifex/services/vpcd/topology_test.go index 0576a12c6..743f604fb 100644 --- a/spinifex/services/vpcd/topology_test.go +++ b/spinifex/services/vpcd/topology_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/services/vpcd/dhcp" "github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb" "github.com/mulgadc/spinifex/spinifex/testutil" @@ -3570,7 +3571,7 @@ func setupPortSGFixture(t *testing.T, mock *MockOVNClient, ctx context.Context, t.Fatalf("CreateDHCPOptions: %v", err) } for _, sgId := range sgIds { - pg := portGroupName(sgId) + pg := topology.SecurityGroupPortGroup(sgId) if err := mock.CreatePortGroup(ctx, pg, nil); err != nil { t.Fatalf("CreatePortGroup %s: %v", sgId, err) } @@ -3617,7 +3618,7 @@ func TestTopologyHandler_CreatePort_JoinsPortGroups(t *testing.T) { } for _, sgId := range []string{"sg-1111", "sg-2222"} { - pg := mock.PortGroups[portGroupName(sgId)] + pg := mock.PortGroups[topology.SecurityGroupPortGroup(sgId)] if !slices.Contains(pg.Ports, lsp.UUID) { t.Errorf("expected %s membership in port group %s, got %v", lsp.UUID, sgId, pg.Ports) } @@ -3695,7 +3696,7 @@ func TestTopologyHandler_DeletePort_ClearsPortGroupMembership(t *testing.T) { resp, _ := nc.Request(TopicCreatePort, data, 5_000_000_000) assertSuccess(t, resp, "create port") - if got := len(mock.PortGroups[portGroupName("sg-del")].Ports); got != 1 { + if got := len(mock.PortGroups[topology.SecurityGroupPortGroup("sg-del")].Ports); got != 1 { t.Fatalf("setup: expected 1 member in port group, got %d", got) } @@ -3710,7 +3711,7 @@ func TestTopologyHandler_DeletePort_ClearsPortGroupMembership(t *testing.T) { resp, _ = nc.Request(TopicDeletePort, data, 5_000_000_000) assertSuccess(t, resp, "delete port") - if got := len(mock.PortGroups[portGroupName("sg-del")].Ports); got != 0 { + if got := len(mock.PortGroups[topology.SecurityGroupPortGroup("sg-del")].Ports); got != 0 { t.Errorf("expected 0 members after delete-port, got %d", got) } } @@ -3792,7 +3793,7 @@ func TestTopologyHandler_UpdatePortSGs_Transitions(t *testing.T) { } expected := make(map[string]struct{}, len(tc.expected)) for _, s := range tc.expected { - expected[portGroupName(s)] = struct{}{} + expected[topology.SecurityGroupPortGroup(s)] = struct{}{} } for name, pg := range mock.PortGroups { inGroup := slices.Contains(pg.Ports, lsp.UUID) @@ -3836,7 +3837,7 @@ func TestTopologyHandler_CreatePort_ExistingLSP_ReconcilesSGMemberships(t *testi Name: "port-eni-recover", Addresses: []string{"02:00:00:00:00:99 10.0.6.50"}, })) - pgName := portGroupName("sg-recover") + pgName := topology.SecurityGroupPortGroup("sg-recover") require.NotContains(t, mock.PortGroups[pgName].Ports, mock.Ports["port-eni-recover"].UUID) evt := PortEvent{