22989950013

Committed 12 Mar 2026 06:43AM UTC coverage: 23.372% (+0.1%) from 23.262%

Build # 22989950013

Build Type

Pull #6330

github

Committed by

oilbeater

Commit Message

Fix minimum and maximum limits for SecurityGroup spec fields

Signed-off-by: Abhishek Pandey <abhpandey@microsoft.com>

Pull Request Pull Request #6330: feat: Extend SG API to have tiers, larger priority range, localAddress and port matches

Coverage Stats

119 of 144 new or added lines in 4 files covered. (82.64%)

6 existing lines in 3 files now uncovered.

12809 of 54804 relevant lines covered (23.37%)

0.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

14.44

/pkg/controller/security_group.go

package controller

import (
        "context"
        "encoding/hex"
        "errors"
        "fmt"
        "reflect"
        "slices"
        "strings"

        "github.com/cnf/structhash"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/cache"
        "k8s.io/klog/v2"

        kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
        "github.com/kubeovn/kube-ovn/pkg/ovs"
        "github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
        "github.com/kubeovn/kube-ovn/pkg/util"
)

func (c *Controller) enqueueAddSg(obj any) {
        key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()
        klog.V(3).Infof("enqueue add securityGroup %s", key)
        c.addOrUpdateSgQueue.Add(key)
}

func (c *Controller) enqueueUpdateSg(oldObj, newObj any) {
        oldSg := oldObj.(*kubeovnv1.SecurityGroup)
        newSg := newObj.(*kubeovnv1.SecurityGroup)
        if !reflect.DeepEqual(oldSg.Spec, newSg.Spec) {
                key := cache.MetaObjectToName(newSg).String()
                klog.V(3).Infof("enqueue update securityGroup %s", key)
                c.addOrUpdateSgQueue.Add(key)
        }
}

func (c *Controller) enqueueDeleteSg(obj any) {
        var sg *kubeovnv1.SecurityGroup
        switch t := obj.(type) {
        case *kubeovnv1.SecurityGroup:
                sg = t
        case cache.DeletedFinalStateUnknown:
                s, ok := t.Obj.(*kubeovnv1.SecurityGroup)
                if !ok {
                        klog.Warningf("unexpected object type: %T", t.Obj)
                        return
                }
                sg = s
        default:
                klog.Warningf("unexpected type: %T", obj)
                return
        }

        key := cache.MetaObjectToName(sg).String()
        klog.V(3).Infof("enqueue delete securityGroup %s", key)
        c.delSgQueue.Add(key)
}

func (c *Controller) initDefaultDenyAllSecurityGroup() error {
        pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)
        if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{
                "type": "security_group",
                sgKey:  util.DenyAllSecurityGroup,
        }); err != nil {
                klog.Errorf("create port group for sg %s: %v", util.DenyAllSecurityGroup, err)
                return err
        }

        if err := c.OVNNbClient.CreateSgDenyAllACL(util.DenyAllSecurityGroup); err != nil {
                klog.Errorf("create deny all acl for sg %s: %v", util.DenyAllSecurityGroup, err)
                return err
        }

        c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)
        return nil
}

func (c *Controller) syncSecurityGroup() error {
        sgs, err := c.sgsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list security groups: %v", err)
                return err
        }
        for _, sg := range sgs {
                lost, err := c.OVNNbClient.SGLostACL(sg)
                if err != nil {
                        err = fmt.Errorf("failed to check if security group %s lost acl: %w", sg.Name, err)
                        klog.Error(err)
                        return err
                }
                if lost {
                        if err := c.handleAddOrUpdateSg(sg.Name, true); err != nil {
                                klog.Errorf("failed to sync security group %s: %v", sg.Name, err)
                        }
                }
        }
        return nil
}

// updateDenyAllSgPorts set lsp to deny which security_groups is not empty
func (c *Controller) updateDenyAllSgPorts() error {
        // list all lsp which security_groups is not empty
        lsps, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{sgsKey: ""})
        if err != nil {
                klog.Errorf("list logical switch ports with security_groups is not empty: %v", err)
                return err
        }

        addPorts := make([]string, 0, len(lsps))
        for _, lsp := range lsps {
                /* skip lsp which security_group does not exist */
                // sgs format: sg1/sg2/sg3
                sgs := strings.Split(lsp.ExternalIDs[sgsKey], "/")
                allNotExist, err := c.securityGroupAllNotExist(sgs)
                if err != nil {
                        klog.Error(err)
                        return err
                }

                if allNotExist {
                        continue
                }

                addPorts = append(addPorts, lsp.Name)
        }
        pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)

        klog.V(6).Infof("setting ports of port group %s to %v", pgName, addPorts)
        if err = c.OVNNbClient.PortGroupSetPorts(pgName, addPorts); err != nil {
                klog.Error(err)
                return err
        }

        return nil
}

func (c *Controller) handleAddOrUpdateSg(key string, force bool) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("handle add/update security group %s", key)

        // set 'deny all' for port associated with security group
        if key == util.DenyAllSecurityGroup {
                if err := c.updateDenyAllSgPorts(); err != nil {
                        klog.Errorf("update sg deny all policy failed. %v", err)
                        return err
                }
        }

        cachedSg, err := c.sgsLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        sg := cachedSg.DeepCopy()

        if err = c.validateSgRule(sg); err != nil {
                klog.Error(err)
                return err
        }

        pgName := ovs.GetSgPortGroupName(sg.Name)
        if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{
                "type": "security_group",
                sgKey:  sg.Name,
        }); err != nil {
                klog.Errorf("create port group for sg %s: %v", sg.Name, err)
                return err
        }

        v4AsName := ovs.GetSgV4AssociatedName(sg.Name)
        v6AsName := ovs.GetSgV6AssociatedName(sg.Name)
        externalIDs := map[string]string{
                sgKey: sg.Name,
        }

        if err = c.OVNNbClient.CreateAddressSet(v4AsName, externalIDs); err != nil {
                klog.Errorf("create address set %s for sg %s: %v", v4AsName, key, err)
                return err
        }

        if err = c.OVNNbClient.CreateAddressSet(v6AsName, externalIDs); err != nil {
                klog.Errorf("create address set %s for sg %s: %v", v6AsName, key, err)
                return err
        }

        var ingressNeedUpdate, egressNeedUpdate bool
        var newIngressMd5, newEgressMd5 string
        if force {
                klog.Infof("force update sg %s", sg.Name)
                ingressNeedUpdate = true
                egressNeedUpdate = true
        } else {
                // check md5
                newIngressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.IngressRules, 1))
                if !sg.Status.IngressLastSyncSuccess || newIngressMd5 != sg.Status.IngressMd5 {
                        klog.Infof("ingress need update, sg:%s", sg.Name)
                        ingressNeedUpdate = true
                }
                newEgressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.EgressRules, 1))
                if !sg.Status.EgressLastSyncSuccess || newEgressMd5 != sg.Status.EgressMd5 {
                        klog.Infof("egress need update, sg:%s", sg.Name)
                        egressNeedUpdate = true
                }

                // check allowSameGroupTraffic switch
                if sg.Status.AllowSameGroupTraffic != sg.Spec.AllowSameGroupTraffic {
                        klog.Infof("both ingress && egress need update, sg:%s", sg.Name)
                        ingressNeedUpdate = true
                        egressNeedUpdate = true
                }
        }

        // update sg rule
        if ingressNeedUpdate {
                if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionToLport); err != nil {
                        sg.Status.IngressLastSyncSuccess = false
                        c.patchSgStatus(sg)
                        klog.Error(err)
                        return err
                }

                if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionToLport); err != nil {
                        klog.Error(err)
                        return err
                }
                sg.Status.IngressMd5 = newIngressMd5
                sg.Status.IngressLastSyncSuccess = true
                c.patchSgStatus(sg)
        }
        if egressNeedUpdate {
                if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionFromLport); err != nil {
                        sg.Status.IngressLastSyncSuccess = false
                        c.patchSgStatus(sg)
                        klog.Error(err)
                        return err
                }

                if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionFromLport); err != nil {
                        klog.Error(err)
                        return err
                }

                sg.Status.EgressMd5 = newEgressMd5
                sg.Status.EgressLastSyncSuccess = true
                c.patchSgStatus(sg)
        }

        // update status
        sg.Status.PortGroup = ovs.GetSgPortGroupName(sg.Name)
        sg.Status.AllowSameGroupTraffic = sg.Spec.AllowSameGroupTraffic
        c.patchSgStatus(sg)
        c.syncSgPortsQueue.Add(key)
        return nil
}

func (c *Controller) validateSgRule(sg *kubeovnv1.SecurityGroup) error {
        // check sg rules
        allRules := append(sg.Spec.IngressRules, sg.Spec.EgressRules...)
        if err := util.ValidateSecurityGroupTier(sg.Spec.Tier); err != nil {
                return err
        }

        for _, rule := range allRules {
                if rule.IPVersion != "ipv4" && rule.IPVersion != "ipv6" {
                        return errors.New("IPVersion should be 'ipv4' or 'ipv6'")
                }

                if rule.Priority < util.SecurityGroupPriorityMin || rule.Priority > util.SecurityGroupPriorityMax {
                        return fmt.Errorf("priority '%d' is not in the range of %d to %d", rule.Priority, util.SecurityGroupPriorityMin, util.SecurityGroupPriorityMax)
                }

                if sg.Spec.Tier == util.SecurityGroupAPITierMaximum && rule.Policy == kubeovnv1.SgPolicyPass {
                        return fmt.Errorf("policy pass not valid when the security group tier is maximum [%d]", util.SecurityGroupAPITierMaximum)
                }

                switch rule.RemoteType {
                case kubeovnv1.SgRemoteTypeAddress:
                        if strings.Contains(rule.RemoteAddress, "/") {
                                if err := util.CheckCidrs(rule.RemoteAddress); err != nil {
                                        return fmt.Errorf("invalid CIDR '%s'", rule.RemoteAddress)
                                }
                        } else {
                                if !util.IsValidIP(rule.RemoteAddress) {
                                        return fmt.Errorf("invalid ip address '%s'", rule.RemoteAddress)
                                }
                        }
                case kubeovnv1.SgRemoteTypeSg:
                        _, err := c.sgsLister.Get(rule.RemoteSecurityGroup)
                        if err != nil {
                                return fmt.Errorf("failed to get remote sg '%s', %w", rule.RemoteSecurityGroup, err)
                        }
                default:
                        return fmt.Errorf("not support sgRemoteType '%s'", rule.RemoteType)
                }

                if rule.LocalAddress != "" {
                        if strings.Contains(rule.LocalAddress, "/") {
                                if err := util.CheckCidrs(rule.LocalAddress); err != nil {
                                        return fmt.Errorf("invalid CIDR '%s'", rule.LocalAddress)
                                }
                        } else {
                                if !util.IsValidIP(rule.LocalAddress) {
                                        return fmt.Errorf("invalid ip address '%s'", rule.LocalAddress)
                                }
                        }
                }

                if rule.Protocol == kubeovnv1.SgProtocolTCP || rule.Protocol == kubeovnv1.SgProtocolUDP {
                        if rule.PortRangeMin < 1 || rule.PortRangeMin > 65535 || rule.PortRangeMax < 1 || rule.PortRangeMax > 65535 {
                                return errors.New("portRange is out of range")
                        }
                        if rule.PortRangeMin > rule.PortRangeMax {
                                return errors.New("portRange err, range Minimum value greater than maximum value")
                        }
                        if rule.LocalAddress != "" {
                                if rule.SourcePortRangeMin < 1 || rule.SourcePortRangeMin > 65535 || rule.SourcePortRangeMax < 1 || rule.SourcePortRangeMax > 65535 {
                                        return errors.New("sourcePortRange is out of range")
                                }
                                if rule.SourcePortRangeMin > rule.SourcePortRangeMax {
                                        return errors.New("sourcePortRange err, range Minimum value greater than maximum value")
                                }
                        }
                }
        }
        return nil
}

func (c *Controller) patchSgStatus(sg *kubeovnv1.SecurityGroup) {
        bytes, err := sg.Status.Bytes()
        if err != nil {
                klog.Error(err)
                return
        }
        if _, err = c.config.KubeOvnClient.KubeovnV1().SecurityGroups().Patch(context.Background(), sg.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {
                klog.Error("patch security group status failed", err)
        }
}

func (c *Controller) handleDeleteSg(key string) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("handle delete security group %s", key)

        if err := c.OVNNbClient.DeleteSecurityGroup(key); err != nil {
                klog.Errorf("delete sg %s: %v", key, err)
                return err
        }

        return nil
}

func (c *Controller) syncSgLogicalPort(key string) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("sync lsp for security group %s", key)

        sgPorts, err := c.OVNNbClient.ListLogicalSwitchPorts(false, map[string]string{"associated_sg_" + key: "true"}, nil)
        if err != nil {
                klog.Errorf("failed to find logical port, %v", err)
                return err
        }

        var ports, v4s, v6s, addresses []string
        for _, lsp := range sgPorts {
                ports = append(ports, lsp.Name)
                if len(lsp.PortSecurity) != 0 {
                        addresses = lsp.PortSecurity
                } else {
                        addresses = lsp.Addresses
                }
                for _, as := range addresses {
                        fields := strings.Fields(as)
                        if len(fields) < 2 {
                                continue
                        }
                        for _, address := range fields[1:] {
                                if strings.Contains(address, ":") {
                                        v6s = append(v6s, address)
                                } else {
                                        v4s = append(v4s, address)
                                }
                        }
                }
        }

        sg, err := c.sgsLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        klog.Warningf("no security group %s", key)
                        // The security group is gone, trigger an update of the deny-all security group
                        // to re-evaluate which ports should be included.
                        c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)
                        return nil
                }
                klog.Errorf("failed to get security group %s: %v", key, err)
                return err
        }

        if err = c.OVNNbClient.PortGroupSetPorts(sg.Status.PortGroup, ports); err != nil {
                klog.Errorf("add ports to port group %s: %v", sg.Status.PortGroup, err)
                return err
        }

        v4AsName := ovs.GetSgV4AssociatedName(key)
        if err := c.OVNNbClient.AddressSetUpdateAddress(v4AsName, v4s...); err != nil {
                klog.Errorf("set ips to address set %s: %v", v4AsName, err)
                return err
        }

        v6AsName := ovs.GetSgV6AssociatedName(key)
        if err := c.OVNNbClient.AddressSetUpdateAddress(v6AsName, v6s...); err != nil {
                klog.Errorf("set ips to address set %s: %v", v6AsName, err)
                return err
        }

        c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)
        return nil
}

func (c *Controller) getPortSg(port *ovnnb.LogicalSwitchPort) ([]string, error) {
        var sgList []string
        for key, value := range port.ExternalIDs {
                if strings.HasPrefix(key, "associated_sg_") && value == "true" {
                        sgName := strings.ReplaceAll(key, "associated_sg_", "")
                        sgList = append(sgList, sgName)
                }
        }
        return sgList, nil
}

func (c *Controller) reconcilePortSg(portName, securityGroups string) error {
        port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, false)
        if err != nil {
                klog.Errorf("failed to get logical switch port %s: %v", portName, err)
                return err
        }
        oldSgList, err := c.getPortSg(port)
        if err != nil {
                klog.Errorf("get port sg failed, %v", err)
                return err
        }
        klog.Infof("reconcile port sg, port='%s', oldSgList='%s', newSgList='%s'", portName, oldSgList, securityGroups)

        newSgList := strings.Split(securityGroups, ",")
        diffSgList := util.DiffStringSlice(oldSgList, newSgList)
        for _, sgName := range diffSgList {
                if sgName == "" {
                        continue
                }
                needAssociated := "false"
                if slices.Contains(newSgList, sgName) {
                        needAssociated = "true"
                }

                if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"associated_sg_" + sgName: needAssociated}); err != nil {
                        klog.Errorf("set logical switch port %s external_ids: %v", portName, err)
                        return err
                }
                c.syncSgPortsQueue.Add(sgName)
        }

        if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"security_groups": strings.ReplaceAll(securityGroups, ",", "/")}); err != nil {
                klog.Errorf("set logical switch port %s external_ids: %v", portName, err)
                return err
        }

        return nil
}

// securityGroupAllNotExist return true if all sgs does not exist
func (c *Controller) securityGroupAllNotExist(sgs []string) (bool, error) {
        if len(sgs) == 0 {
                return true, nil
        }

        notExistsCount := 0
        // sgs format: sg1/sg2/sg3
        for _, sg := range sgs {
                ok, err := c.OVNNbClient.PortGroupExists(ovs.GetSgPortGroupName(sg))
                if err != nil {
                        klog.Error(err)
                        return true, err
                }

                if !ok {
                        notExistsCount++
                }
        }

        return notExistsCount == len(sgs), nil
}

1	package controller
2
3	import (
4	"context"
5	"encoding/hex"
6	"errors"
7	"fmt"
8	"reflect"
9	"slices"
10	"strings"
11
12	"github.com/cnf/structhash"
13	k8serrors "k8s.io/apimachinery/pkg/api/errors"
14	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
15	"k8s.io/apimachinery/pkg/labels"
16	"k8s.io/apimachinery/pkg/types"
17	"k8s.io/client-go/tools/cache"
18	"k8s.io/klog/v2"
19
20	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
21	"github.com/kubeovn/kube-ovn/pkg/ovs"
22	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
23	"github.com/kubeovn/kube-ovn/pkg/util"
24	)
25
26	func (c *Controller) enqueueAddSg(obj any) {	×
27	key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()	×
28	klog.V(3).Infof("enqueue add securityGroup %s", key)	×
29	c.addOrUpdateSgQueue.Add(key)	×
30	}	×
31
32	func (c *Controller) enqueueUpdateSg(oldObj, newObj any) {	×
33	oldSg := oldObj.(*kubeovnv1.SecurityGroup)	×
34	newSg := newObj.(*kubeovnv1.SecurityGroup)	×
35	if !reflect.DeepEqual(oldSg.Spec, newSg.Spec) {	×
36	key := cache.MetaObjectToName(newSg).String()	×
37	klog.V(3).Infof("enqueue update securityGroup %s", key)	×
38	c.addOrUpdateSgQueue.Add(key)	×
39	}	×
40	}
41
42	func (c *Controller) enqueueDeleteSg(obj any) {	×
43	var sg *kubeovnv1.SecurityGroup	×
44	switch t := obj.(type) {	×
45	case *kubeovnv1.SecurityGroup:	×
46	sg = t	×
47	case cache.DeletedFinalStateUnknown:	×
48	s, ok := t.Obj.(*kubeovnv1.SecurityGroup)	×
49	if !ok {	×
50	klog.Warningf("unexpected object type: %T", t.Obj)	×
51	return	×
52	}	×
53	sg = s	×
54	default:	×
55	klog.Warningf("unexpected type: %T", obj)	×
56	return	×
57	}
58
59	key := cache.MetaObjectToName(sg).String()	×
60	klog.V(3).Infof("enqueue delete securityGroup %s", key)	×
61	c.delSgQueue.Add(key)	×
62	}
63
64	func (c *Controller) initDefaultDenyAllSecurityGroup() error {	×
65	pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)	×
66	if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{	×
67	"type": "security_group",	×
68	sgKey: util.DenyAllSecurityGroup,	×
69	}); err != nil {	×
70	klog.Errorf("create port group for sg %s: %v", util.DenyAllSecurityGroup, err)	×
71	return err	×
72	}	×
73
74	if err := c.OVNNbClient.CreateSgDenyAllACL(util.DenyAllSecurityGroup); err != nil {	×
75	klog.Errorf("create deny all acl for sg %s: %v", util.DenyAllSecurityGroup, err)	×
76	return err	×
77	}	×
78
79	c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)	×
80	return nil	×
81	}
82
83	func (c *Controller) syncSecurityGroup() error {	×
84	sgs, err := c.sgsLister.List(labels.Everything())	×
85	if err != nil {	×
86	klog.Errorf("failed to list security groups: %v", err)	×
87	return err	×
88	}	×
89	for _, sg := range sgs {	×
90	lost, err := c.OVNNbClient.SGLostACL(sg)	×
91	if err != nil {	×
92	err = fmt.Errorf("failed to check if security group %s lost acl: %w", sg.Name, err)	×
93	klog.Error(err)	×
94	return err	×
95	}	×
96	if lost {	×
97	if err := c.handleAddOrUpdateSg(sg.Name, true); err != nil {	×
98	klog.Errorf("failed to sync security group %s: %v", sg.Name, err)	×
99	}	×
100	}
101	}
102	return nil	×
103	}
104
105	// updateDenyAllSgPorts set lsp to deny which security_groups is not empty
106	func (c *Controller) updateDenyAllSgPorts() error {	×
107	// list all lsp which security_groups is not empty	×
108	lsps, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{sgsKey: ""})	×
109	if err != nil {	×
110	klog.Errorf("list logical switch ports with security_groups is not empty: %v", err)	×
111	return err	×
112	}	×
113
114	addPorts := make([]string, 0, len(lsps))	×
115	for _, lsp := range lsps {	×
116	/* skip lsp which security_group does not exist */	×
117	// sgs format: sg1/sg2/sg3	×
118	sgs := strings.Split(lsp.ExternalIDs[sgsKey], "/")	×
119	allNotExist, err := c.securityGroupAllNotExist(sgs)	×
120	if err != nil {	×
121	klog.Error(err)	×
122	return err	×
123	}	×
124
125	if allNotExist {	×
126	continue	×
127	}
128
129	addPorts = append(addPorts, lsp.Name)	×
130	}
131	pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)	×
132		×
133	klog.V(6).Infof("setting ports of port group %s to %v", pgName, addPorts)	×
134	if err = c.OVNNbClient.PortGroupSetPorts(pgName, addPorts); err != nil {	×
135	klog.Error(err)	×
136	return err	×
137	}	×
138
139	return nil	×
140	}
141
142	func (c *Controller) handleAddOrUpdateSg(key string, force bool) error {	×
143	c.sgKeyMutex.LockKey(key)	×
144	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
145	klog.Infof("handle add/update security group %s", key)	×
146		×
147	// set 'deny all' for port associated with security group	×
148	if key == util.DenyAllSecurityGroup {	×
149	if err := c.updateDenyAllSgPorts(); err != nil {	×
150	klog.Errorf("update sg deny all policy failed. %v", err)	×
151	return err	×
152	}	×
153	}
154
155	cachedSg, err := c.sgsLister.Get(key)	×
156	if err != nil {	×
157	if k8serrors.IsNotFound(err) {	×
158	return nil	×
159	}	×
160	klog.Error(err)	×
161	return err	×
162	}
163	sg := cachedSg.DeepCopy()	×
164		×
165	if err = c.validateSgRule(sg); err != nil {	×
166	klog.Error(err)	×
167	return err	×
168	}	×
169
170	pgName := ovs.GetSgPortGroupName(sg.Name)	×
171	if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{	×
172	"type": "security_group",	×
173	sgKey: sg.Name,	×
174	}); err != nil {	×
175	klog.Errorf("create port group for sg %s: %v", sg.Name, err)	×
176	return err	×
177	}	×
178
179	v4AsName := ovs.GetSgV4AssociatedName(sg.Name)	×
180	v6AsName := ovs.GetSgV6AssociatedName(sg.Name)	×
181	externalIDs := map[string]string{	×
182	sgKey: sg.Name,	×
183	}	×
184		×
185	if err = c.OVNNbClient.CreateAddressSet(v4AsName, externalIDs); err != nil {	×
186	klog.Errorf("create address set %s for sg %s: %v", v4AsName, key, err)	×
187	return err	×
188	}	×
189
190	if err = c.OVNNbClient.CreateAddressSet(v6AsName, externalIDs); err != nil {	×
191	klog.Errorf("create address set %s for sg %s: %v", v6AsName, key, err)	×
192	return err	×
193	}	×
194
195	var ingressNeedUpdate, egressNeedUpdate bool	×
196	var newIngressMd5, newEgressMd5 string	×
197	if force {	×
198	klog.Infof("force update sg %s", sg.Name)	×
199	ingressNeedUpdate = true	×
200	egressNeedUpdate = true	×
201	} else {	×
202	// check md5	×
203	newIngressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.IngressRules, 1))	×
204	if !sg.Status.IngressLastSyncSuccess \|\| newIngressMd5 != sg.Status.IngressMd5 {	×
205	klog.Infof("ingress need update, sg:%s", sg.Name)	×
206	ingressNeedUpdate = true	×
207	}	×
208	newEgressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.EgressRules, 1))	×
209	if !sg.Status.EgressLastSyncSuccess \|\| newEgressMd5 != sg.Status.EgressMd5 {	×
210	klog.Infof("egress need update, sg:%s", sg.Name)	×
211	egressNeedUpdate = true	×
212	}	×
213
214	// check allowSameGroupTraffic switch
215	if sg.Status.AllowSameGroupTraffic != sg.Spec.AllowSameGroupTraffic {	×
216	klog.Infof("both ingress && egress need update, sg:%s", sg.Name)	×
217	ingressNeedUpdate = true	×
218	egressNeedUpdate = true	×
219	}	×
220	}
221
222	// update sg rule
223	if ingressNeedUpdate {	×
224	if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionToLport); err != nil {	×
225	sg.Status.IngressLastSyncSuccess = false	×
226	c.patchSgStatus(sg)	×
227	klog.Error(err)	×
228	return err	×
229	}	×
230
231	if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionToLport); err != nil {	×
232	klog.Error(err)	×
233	return err	×
234	}	×
235	sg.Status.IngressMd5 = newIngressMd5	×
236	sg.Status.IngressLastSyncSuccess = true	×
237	c.patchSgStatus(sg)	×
238	}
239	if egressNeedUpdate {	×
240	if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionFromLport); err != nil {	×
241	sg.Status.IngressLastSyncSuccess = false	×
242	c.patchSgStatus(sg)	×
243	klog.Error(err)	×
244	return err	×
245	}	×
246
247	if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionFromLport); err != nil {	×
248	klog.Error(err)	×
249	return err	×
250	}	×
251
252	sg.Status.EgressMd5 = newEgressMd5	×
253	sg.Status.EgressLastSyncSuccess = true	×
254	c.patchSgStatus(sg)	×
255	}
256
257	// update status
258	sg.Status.PortGroup = ovs.GetSgPortGroupName(sg.Name)	×
259	sg.Status.AllowSameGroupTraffic = sg.Spec.AllowSameGroupTraffic	×
260	c.patchSgStatus(sg)	×
261	c.syncSgPortsQueue.Add(key)	×
262	return nil	×
263	}
264
265	func (c Controller) validateSgRule(sg kubeovnv1.SecurityGroup) error {	1✔
266	// check sg rules	1✔
267	allRules := append(sg.Spec.IngressRules, sg.Spec.EgressRules...)	1✔
268	if err := util.ValidateSecurityGroupTier(sg.Spec.Tier); err != nil {	1✔
NEW 269	return err	×
NEW 270	}	×
271
272	for _, rule := range allRules {	2✔
273	if rule.IPVersion != "ipv4" && rule.IPVersion != "ipv6" {	1✔
274	return errors.New("IPVersion should be 'ipv4' or 'ipv6'")	×
275	}	×
276
277	if rule.Priority < util.SecurityGroupPriorityMin \|\| rule.Priority > util.SecurityGroupPriorityMax {	1✔
NEW 278	return fmt.Errorf("priority '%d' is not in the range of %d to %d", rule.Priority, util.SecurityGroupPriorityMin, util.SecurityGroupPriorityMax)	×
NEW 279	}	×
280
281	if sg.Spec.Tier == util.SecurityGroupAPITierMaximum && rule.Policy == kubeovnv1.SgPolicyPass {	1✔
NEW 282	return fmt.Errorf("policy pass not valid when the security group tier is maximum [%d]", util.SecurityGroupAPITierMaximum)	×
UNCOV 283	}	×
284
285	switch rule.RemoteType {	1✔
286	case kubeovnv1.SgRemoteTypeAddress:	1✔
287	if strings.Contains(rule.RemoteAddress, "/") {	2✔
288	if err := util.CheckCidrs(rule.RemoteAddress); err != nil {	1✔
289	return fmt.Errorf("invalid CIDR '%s'", rule.RemoteAddress)	×
290	}	×
291	} else {	1✔
292	if !util.IsValidIP(rule.RemoteAddress) {	1✔
293	return fmt.Errorf("invalid ip address '%s'", rule.RemoteAddress)	×
294	}	×
295	}
296	case kubeovnv1.SgRemoteTypeSg:	×
297	_, err := c.sgsLister.Get(rule.RemoteSecurityGroup)	×
298	if err != nil {	×
299	return fmt.Errorf("failed to get remote sg '%s', %w", rule.RemoteSecurityGroup, err)	×
300	}	×
301	default:	×
302	return fmt.Errorf("not support sgRemoteType '%s'", rule.RemoteType)	×
303	}
304
305	if rule.LocalAddress != "" {	2✔
306	if strings.Contains(rule.LocalAddress, "/") {	2✔
307	if err := util.CheckCidrs(rule.LocalAddress); err != nil {	2✔
308	return fmt.Errorf("invalid CIDR '%s'", rule.LocalAddress)	1✔
309	}	1✔
310	} else {	1✔
311	if !util.IsValidIP(rule.LocalAddress) {	2✔
312	return fmt.Errorf("invalid ip address '%s'", rule.LocalAddress)	1✔
313	}	1✔
314	}
315	}
316
317	if rule.Protocol == kubeovnv1.SgProtocolTCP \|\| rule.Protocol == kubeovnv1.SgProtocolUDP {	2✔
318	if rule.PortRangeMin < 1 \|\| rule.PortRangeMin > 65535 \|\| rule.PortRangeMax < 1 \|\| rule.PortRangeMax > 65535 {	1✔
319	return errors.New("portRange is out of range")	×
320	}	×
321	if rule.PortRangeMin > rule.PortRangeMax {	1✔
322	return errors.New("portRange err, range Minimum value greater than maximum value")	×
323	}	×
324	if rule.LocalAddress != "" {	2✔
325	if rule.SourcePortRangeMin < 1 \|\| rule.SourcePortRangeMin > 65535 \|\| rule.SourcePortRangeMax < 1 \|\| rule.SourcePortRangeMax > 65535 {	2✔
326	return errors.New("sourcePortRange is out of range")	1✔
327	}	1✔
328	if rule.SourcePortRangeMin > rule.SourcePortRangeMax {	2✔
329	return errors.New("sourcePortRange err, range Minimum value greater than maximum value")	1✔
330	}	1✔
331	}
332	}
333	}
334	return nil	1✔
335	}
336
337	func (c Controller) patchSgStatus(sg kubeovnv1.SecurityGroup) {	×
338	bytes, err := sg.Status.Bytes()	×
339	if err != nil {	×
340	klog.Error(err)	×
341	return	×
342	}	×
343	if _, err = c.config.KubeOvnClient.KubeovnV1().SecurityGroups().Patch(context.Background(), sg.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {	×
344	klog.Error("patch security group status failed", err)	×
345	}	×
346	}
347
348	func (c *Controller) handleDeleteSg(key string) error {	×
349	c.sgKeyMutex.LockKey(key)	×
350	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
351	klog.Infof("handle delete security group %s", key)	×
352		×
353	if err := c.OVNNbClient.DeleteSecurityGroup(key); err != nil {	×
354	klog.Errorf("delete sg %s: %v", key, err)	×
355	return err	×
356	}	×
357
358	return nil	×
359	}
360
361	func (c *Controller) syncSgLogicalPort(key string) error {	×
362	c.sgKeyMutex.LockKey(key)	×
363	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
364	klog.Infof("sync lsp for security group %s", key)	×
365		×
366	sgPorts, err := c.OVNNbClient.ListLogicalSwitchPorts(false, map[string]string{"associated_sg_" + key: "true"}, nil)	×
367	if err != nil {	×
368	klog.Errorf("failed to find logical port, %v", err)	×
369	return err	×
370	}	×
371
372	var ports, v4s, v6s, addresses []string	×
373	for _, lsp := range sgPorts {	×
374	ports = append(ports, lsp.Name)	×
375	if len(lsp.PortSecurity) != 0 {	×
376	addresses = lsp.PortSecurity	×
377	} else {	×
378	addresses = lsp.Addresses	×
379	}	×
380	for _, as := range addresses {	×
381	fields := strings.Fields(as)	×
382	if len(fields) < 2 {	×
383	continue	×
384	}
385	for _, address := range fields[1:] {	×
386	if strings.Contains(address, ":") {	×
387	v6s = append(v6s, address)	×
388	} else {	×
389	v4s = append(v4s, address)	×
390	}	×
391	}
392	}
393	}
394
395	sg, err := c.sgsLister.Get(key)	×
396	if err != nil {	×
397	if k8serrors.IsNotFound(err) {	×
398	klog.Warningf("no security group %s", key)	×
399	// The security group is gone, trigger an update of the deny-all security group	×
400	// to re-evaluate which ports should be included.	×
401	c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)	×
402	return nil	×
403	}	×
404	klog.Errorf("failed to get security group %s: %v", key, err)	×
405	return err	×
406	}
407
408	if err = c.OVNNbClient.PortGroupSetPorts(sg.Status.PortGroup, ports); err != nil {	×
409	klog.Errorf("add ports to port group %s: %v", sg.Status.PortGroup, err)	×
410	return err	×
411	}	×
412
413	v4AsName := ovs.GetSgV4AssociatedName(key)	×
414	if err := c.OVNNbClient.AddressSetUpdateAddress(v4AsName, v4s...); err != nil {	×
415	klog.Errorf("set ips to address set %s: %v", v4AsName, err)	×
416	return err	×
417	}	×
418
419	v6AsName := ovs.GetSgV6AssociatedName(key)	×
420	if err := c.OVNNbClient.AddressSetUpdateAddress(v6AsName, v6s...); err != nil {	×
421	klog.Errorf("set ips to address set %s: %v", v6AsName, err)	×
422	return err	×
423	}	×
424
425	c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)	×
426	return nil	×
427	}
428
429	func (c Controller) getPortSg(port ovnnb.LogicalSwitchPort) ([]string, error) {	1✔
430	var sgList []string	1✔
431	for key, value := range port.ExternalIDs {	2✔
432	if strings.HasPrefix(key, "associated_sg_") && value == "true" {	2✔
433	sgName := strings.ReplaceAll(key, "associated_sg_", "")	1✔
434	sgList = append(sgList, sgName)	1✔
435	}	1✔
436	}
437	return sgList, nil	1✔
438	}
439
440	func (c *Controller) reconcilePortSg(portName, securityGroups string) error {	×
441	port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, false)	×
442	if err != nil {	×
443	klog.Errorf("failed to get logical switch port %s: %v", portName, err)	×
444	return err	×
445	}	×
446	oldSgList, err := c.getPortSg(port)	×
447	if err != nil {	×
448	klog.Errorf("get port sg failed, %v", err)	×
449	return err	×
450	}	×
451	klog.Infof("reconcile port sg, port='%s', oldSgList='%s', newSgList='%s'", portName, oldSgList, securityGroups)	×
452		×
453	newSgList := strings.Split(securityGroups, ",")	×
454	diffSgList := util.DiffStringSlice(oldSgList, newSgList)	×
455	for _, sgName := range diffSgList {	×
456	if sgName == "" {	×
457	continue	×
458	}
459	needAssociated := "false"	×
460	if slices.Contains(newSgList, sgName) {	×
461	needAssociated = "true"	×
462	}	×
463
464	if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"associated_sg_" + sgName: needAssociated}); err != nil {	×
465	klog.Errorf("set logical switch port %s external_ids: %v", portName, err)	×
466	return err	×
467	}	×
468	c.syncSgPortsQueue.Add(sgName)	×
469	}
470
471	if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"security_groups": strings.ReplaceAll(securityGroups, ",", "/")}); err != nil {	×
472	klog.Errorf("set logical switch port %s external_ids: %v", portName, err)	×
473	return err	×
474	}	×
475
476	return nil	×
477	}
478
479	// securityGroupAllNotExist return true if all sgs does not exist
480	func (c *Controller) securityGroupAllNotExist(sgs []string) (bool, error) {	1✔
481	if len(sgs) == 0 {	2✔
482	return true, nil	1✔
483	}	1✔
484
485	notExistsCount := 0	1✔
486	// sgs format: sg1/sg2/sg3	1✔
487	for _, sg := range sgs {	2✔
488	ok, err := c.OVNNbClient.PortGroupExists(ovs.GetSgPortGroupName(sg))	1✔
489	if err != nil {	1✔
490	klog.Error(err)	×
491	return true, err	×
492	}	×
493
494	if !ok {	2✔
495	notExistsCount++	1✔
496	}	1✔
497	}
498
499	return notExistsCount == len(sgs), nil	1✔
500	}

kubeovn / kube-ovn / 22989950013

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous