diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupHandler.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupHandler.java
new file mode 100644
index 0000000..77b60d9
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupHandler.java
@@ -0,0 +1,1072 @@
+/*
+ * Copyright 2020-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.kubevirtnetworking.impl;
+
+import com.google.common.collect.Sets;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IPv4;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.Ip4Prefix;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.TpPort;
+import org.onlab.packet.VlanId;
+import org.onlab.util.Tools;
+import org.onosproject.cfg.ComponentConfigService;
+import org.onosproject.cfg.ConfigProperty;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.LeadershipService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.kubevirtnetworking.api.KubevirtFlowRuleService;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetwork;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetworkEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetworkListener;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetworkService;
+import org.onosproject.kubevirtnetworking.api.KubevirtPort;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortListener;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortService;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupListener;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+import org.onosproject.kubevirtnetworking.util.RulePopulatorUtil;
+import org.onosproject.kubevirtnode.api.KubevirtNode;
+import org.onosproject.kubevirtnode.api.KubevirtNodeEvent;
+import org.onosproject.kubevirtnode.api.KubevirtNodeListener;
+import org.onosproject.kubevirtnode.api.KubevirtNodeService;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.driver.DriverService;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.flow.criteria.ExtensionSelector;
+import org.onosproject.net.flow.instructions.ExtensionTreatment;
+import org.onosproject.store.service.StorageService;
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.util.Dictionary;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+import static java.lang.Thread.sleep;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.kubevirtnetworking.api.Constants.ERROR_TABLE;
+import static org.onosproject.kubevirtnetworking.api.Constants.KUBEVIRT_NETWORKING_APP_ID;
+import static org.onosproject.kubevirtnetworking.api.Constants.PRIORITY_ACL_INGRESS_RULE;
+import static org.onosproject.kubevirtnetworking.api.Constants.PRIORITY_ACL_RULE;
+import static org.onosproject.kubevirtnetworking.api.Constants.PRIORITY_CT_DROP_RULE;
+import static org.onosproject.kubevirtnetworking.api.Constants.PRIORITY_CT_HOOK_RULE;
+import static org.onosproject.kubevirtnetworking.api.Constants.PRIORITY_CT_RULE;
+import static org.onosproject.kubevirtnetworking.api.Constants.TENANT_ACL_CT_TABLE;
+import static org.onosproject.kubevirtnetworking.api.Constants.TENANT_ACL_EGRESS_TABLE;
+import static org.onosproject.kubevirtnetworking.api.Constants.TENANT_ACL_INGRESS_TABLE;
+import static org.onosproject.kubevirtnetworking.api.Constants.TENANT_ACL_RECIRC_TABLE;
+import static org.onosproject.kubevirtnetworking.api.Constants.TENANT_FORWARDING_TABLE;
+import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.GENEVE;
+import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.GRE;
+import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.VLAN;
+import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.VXLAN;
+import static org.onosproject.kubevirtnetworking.impl.OsgiPropertyConstants.USE_SECURITY_GROUP;
+import static org.onosproject.kubevirtnetworking.impl.OsgiPropertyConstants.USE_SECURITY_GROUP_DEFAULT;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.getPropertyValueAsBoolean;
+import static org.onosproject.kubevirtnetworking.util.RulePopulatorUtil.buildPortRangeMatches;
+import static org.onosproject.kubevirtnetworking.util.RulePopulatorUtil.computeCtMaskFlag;
+import static org.onosproject.kubevirtnetworking.util.RulePopulatorUtil.computeCtStateFlag;
+import static org.onosproject.kubevirtnetworking.util.RulePopulatorUtil.niciraConnTrackTreatmentBuilder;
+import static org.onosproject.kubevirtnode.api.KubevirtNode.Type.WORKER;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Populates flow rules to handle EdgeStack SecurityGroups.
+ */
+@Component(
+        immediate = true,
+        property = {
+                USE_SECURITY_GROUP + ":Boolean=" + USE_SECURITY_GROUP_DEFAULT
+        }
+)
+public class KubevirtSecurityGroupHandler {
+
+    private final Logger log = getLogger(getClass());
+
+    private static final int VM_IP_PREFIX = 32;
+
+    private static final String STR_NULL = "null";
+    private static final String PROTO_ICMP = "ICMP";
+    private static final String PROTO_ICMP_NUM = "1";
+    private static final String PROTO_TCP = "TCP";
+    private static final String PROTO_TCP_NUM = "6";
+    private static final String PROTO_UDP = "UDP";
+    private static final String PROTO_UDP_NUM = "17";
+    private static final String PROTO_SCTP = "SCTP";
+    private static final String PROTO_SCTP_NUM = "132";
+    private static final byte PROTOCOL_SCTP = (byte) 0x84;
+    private static final String PROTO_ANY = "ANY";
+    private static final String PROTO_ANY_NUM = "0";
+    private static final String ETHTYPE_IPV4 = "IPV4";
+    private static final String EGRESS = "EGRESS";
+    private static final String INGRESS = "INGRESS";
+    private static final IpPrefix IP_PREFIX_ANY = Ip4Prefix.valueOf("0.0.0.0/0");
+
+    private static final int ICMP_CODE_MIN = 0;
+    private static final int ICMP_CODE_MAX = 255;
+    private static final int ICMP_TYPE_MIN = 0;
+    private static final int ICMP_TYPE_MAX = 255;
+
+    private static final int CT_COMMIT = 0;
+    private static final int CT_NO_COMMIT = 1;
+    private static final short CT_NO_RECIRC = -1;
+
+    private static final int ACTION_NONE = 0;
+    private static final int ACTION_DROP = -1;
+
+    private static final long SLEEP_MS = 5000;
+
+    /** Apply EdgeStack security group rule for VM traffic. */
+    private boolean useSecurityGroup = USE_SECURITY_GROUP_DEFAULT;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected CoreService coreService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected MastershipService mastershipService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected DriverService driverService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected LeadershipService leadershipService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ClusterService clusterService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected StorageService storageService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected ComponentConfigService configService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected KubevirtNodeService nodeService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected KubevirtNetworkService networkService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected KubevirtPortService portService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected KubevirtFlowRuleService flowRuleService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY)
+    protected KubevirtSecurityGroupService securityGroupService;
+
+    private final KubevirtPortListener portListener =
+            new InternalKubevirtPortListener();
+    private final KubevirtSecurityGroupListener securityGroupListener =
+            new InternalSecurityGroupListener();
+    private final KubevirtNodeListener nodeListener =
+            new InternalNodeListener();
+    private final KubevirtNetworkListener networkListener =
+            new InternalNetworkListener();
+
+    private final ExecutorService eventExecutor = newSingleThreadExecutor(
+            groupedThreads(this.getClass().getSimpleName(), "event-handler"));
+
+    private ApplicationId appId;
+    private NodeId localNodeId;
+
+    @Activate
+    protected void activate() {
+        appId = coreService.registerApplication(KUBEVIRT_NETWORKING_APP_ID);
+        localNodeId = clusterService.getLocalNode().id();
+        securityGroupService.addListener(securityGroupListener);
+        portService.addListener(portListener);
+        networkService.addListener(networkListener);
+        configService.registerProperties(getClass());
+        nodeService.addListener(nodeListener);
+
+        log.info("Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        securityGroupService.removeListener(securityGroupListener);
+        portService.removeListener(portListener);
+        configService.unregisterProperties(getClass(), false);
+        nodeService.removeListener(nodeListener);
+        networkService.removeListener(networkListener);
+        eventExecutor.shutdown();
+
+        log.info("Stopped");
+    }
+
+    @Modified
+    protected void modified(ComponentContext context) {
+        Dictionary<?, ?> properties = context.getProperties();
+        Boolean flag;
+
+        flag = Tools.isPropertyEnabled(properties, USE_SECURITY_GROUP);
+        if (flag == null) {
+            log.info("useSecurityGroup is not configured, " +
+                    "using current value of {}", useSecurityGroup);
+        } else {
+            useSecurityGroup = flag;
+            log.info("Configured. useSecurityGroup is {}",
+                    useSecurityGroup ? "enabled" : "disabled");
+        }
+
+        securityGroupService.setSecurityGroupEnabled(useSecurityGroup);
+        resetSecurityGroupRules();
+    }
+
+    private boolean getUseSecurityGroupFlag() {
+        Set<ConfigProperty> properties =
+                configService.getProperties(getClass().getName());
+        return getPropertyValueAsBoolean(properties, USE_SECURITY_GROUP);
+    }
+
+    private void initializeConnTrackTable(DeviceId deviceId, boolean install) {
+
+        // table={ACL_INGRESS_TABLE(44)},ip,ct_state=-trk, actions=ct(table:{ACL_CT_TABLE(45)})
+        long ctState = computeCtStateFlag(false, false, false);
+        long ctMask = computeCtMaskFlag(true, false, false);
+        setConnTrackRule(deviceId, ctState, ctMask, CT_NO_COMMIT, (short) TENANT_ACL_CT_TABLE,
+                ACTION_NONE, PRIORITY_CT_HOOK_RULE, install);
+
+        //table={ACL_CT_TABLE(45)},ip,nw_dst=10.10.0.2,ct_state=+trk+est,action=goto_table:{NORMAL_TABLE(80)}
+        ctState = computeCtStateFlag(true, false, true);
+        ctMask = computeCtMaskFlag(true, false, true);
+        setConnTrackRule(deviceId, ctState, ctMask, CT_NO_COMMIT, CT_NO_RECIRC,
+                TENANT_FORWARDING_TABLE, PRIORITY_CT_RULE, install);
+
+        //table={ACL_CT_TABLE(45)},ip,nw_dst=10.10.0.2,ct_state=+trk+new,action=drop
+        ctState = computeCtStateFlag(true, true, false);
+        ctMask = computeCtMaskFlag(true, true, false);
+        setConnTrackRule(deviceId, ctState, ctMask, CT_NO_COMMIT, CT_NO_RECIRC,
+                ACTION_DROP, PRIORITY_CT_DROP_RULE, install);
+    }
+
+    private void initializeAclTable(DeviceId deviceId, boolean install) {
+
+        ExtensionTreatment ctTreatment =
+                niciraConnTrackTreatmentBuilder(driverService, deviceId)
+                        .commit(true)
+                        .build();
+
+        TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder();
+        sBuilder.matchEthType(Ethernet.TYPE_IPV4);
+
+        TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder();
+        tBuilder.extension(ctTreatment, deviceId)
+                .transition(TENANT_FORWARDING_TABLE);
+
+        flowRuleService.setRule(appId,
+                deviceId,
+                sBuilder.build(),
+                tBuilder.build(),
+                PRIORITY_ACL_INGRESS_RULE,
+                TENANT_ACL_RECIRC_TABLE,
+                install);
+    }
+
+    private void initializeEgressTable(DeviceId deviceId, boolean install) {
+        if (install) {
+            flowRuleService.setUpTableMissEntry(deviceId, TENANT_ACL_EGRESS_TABLE);
+        } else {
+            flowRuleService.connectTables(deviceId, TENANT_ACL_EGRESS_TABLE, TENANT_FORWARDING_TABLE);
+        }
+    }
+
+    private void initializeIngressTable(DeviceId deviceId, boolean install) {
+        if (install) {
+            flowRuleService.setUpTableMissEntry(deviceId, TENANT_ACL_INGRESS_TABLE);
+        } else {
+            flowRuleService.connectTables(deviceId, TENANT_ACL_INGRESS_TABLE, TENANT_FORWARDING_TABLE);
+        }
+    }
+
+    private void updateSecurityGroupRule(KubevirtPort port,
+                                         KubevirtSecurityGroupRule sgRule, boolean install) {
+
+        if (port == null || sgRule == null) {
+            return;
+        }
+
+        if (sgRule.remoteGroupId() != null && !sgRule.remoteGroupId().isEmpty()) {
+            getRemotePorts(port, sgRule.remoteGroupId())
+                    .forEach(rPort -> {
+                        populateSecurityGroupRule(sgRule, port,
+                                rPort.ipAddress().toIpPrefix(), install);
+                        populateSecurityGroupRule(sgRule, rPort,
+                                port.ipAddress().toIpPrefix(), install);
+
+                        KubevirtSecurityGroupRule rSgRule = sgRule.updateDirection(
+                                sgRule.direction().equalsIgnoreCase(EGRESS) ? INGRESS : EGRESS);
+                        populateSecurityGroupRule(rSgRule, port,
+                                rPort.ipAddress().toIpPrefix(), install);
+                        populateSecurityGroupRule(rSgRule, rPort,
+                                port.ipAddress().toIpPrefix(), install);
+                    });
+        } else {
+            populateSecurityGroupRule(sgRule, port,
+                    sgRule.remoteIpPrefix() == null ? IP_PREFIX_ANY :
+                            sgRule.remoteIpPrefix(), install);
+        }
+    }
+
+    private boolean checkProtocol(String protocol) {
+        if (protocol == null) {
+            log.debug("No protocol was specified, use default IP(v4/v6) protocol.");
+            return true;
+        } else {
+            String protocolUpper = protocol.toUpperCase();
+            if (protocolUpper.equals(PROTO_TCP) ||
+                    protocolUpper.equals(PROTO_UDP) ||
+                    protocolUpper.equals(PROTO_ICMP) ||
+                    protocolUpper.equals(PROTO_SCTP) ||
+                    protocolUpper.equals(PROTO_ANY) ||
+                    protocol.equals(PROTO_TCP_NUM) ||
+                    protocol.equals(PROTO_UDP_NUM) ||
+                    protocol.equals(PROTO_ICMP_NUM) ||
+                    protocol.equals(PROTO_SCTP_NUM) ||
+                    protocol.equals(PROTO_ANY_NUM)) {
+                return true;
+            } else {
+                log.error("Unsupported protocol {}, we only support " +
+                        "TCP/UDP/ICMP/SCTP protocols.", protocol);
+                return false;
+            }
+        }
+    }
+
+    private void populateSecurityGroupRule(KubevirtSecurityGroupRule sgRule,
+                                           KubevirtPort port,
+                                           IpPrefix remoteIp,
+                                           boolean install) {
+        if (!checkProtocol(sgRule.protocol())) {
+            return;
+        }
+
+        DeviceId deviceId = port.isTenant() ? port.tenantDeviceId() : port.deviceId();
+
+        Set<TrafficSelector> selectors = buildSelectors(sgRule,
+                Ip4Address.valueOf(port.ipAddress().toInetAddress()),
+                remoteIp, port.networkId());
+        if (selectors == null || selectors.isEmpty()) {
+            return;
+        }
+
+        // if the device is not available we do not perform any action
+        if (deviceId == null || !deviceService.isAvailable(deviceId)) {
+            return;
+        }
+
+        // XXX All egress traffic needs to go through connection tracking module,
+        // which might hurt its performance.
+        ExtensionTreatment ctTreatment =
+                niciraConnTrackTreatmentBuilder(driverService, deviceId)
+                        .commit(true)
+                        .build();
+
+        TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder();
+
+        int aclTable;
+        if (sgRule.direction().equalsIgnoreCase(EGRESS)) {
+            aclTable = TENANT_ACL_EGRESS_TABLE;
+            tBuilder.transition(TENANT_ACL_RECIRC_TABLE);
+        } else {
+            aclTable = TENANT_ACL_INGRESS_TABLE;
+            tBuilder.extension(ctTreatment, deviceId)
+                    .transition(TENANT_FORWARDING_TABLE);
+        }
+
+        int finalAclTable = aclTable;
+        selectors.forEach(selector -> {
+            flowRuleService.setRule(appId,
+                    deviceId,
+                    selector, tBuilder.build(),
+                    PRIORITY_ACL_RULE,
+                    finalAclTable,
+                    install);
+        });
+    }
+
+    /**
+     * Sets connection tracking rule using OVS extension commands.
+     * It is not so graceful, but I don't want to make it more general because
+     * it is going to be used only here.
+     * The following is the usage of the function.
+     *
+     * @param deviceId Device ID
+     * @param ctState ctState: please use RulePopulatorUtil.computeCtStateFlag()
+     *                to build the value
+     * @param ctMask crMask: please use RulePopulatorUtil.computeCtMaskFlag()
+     *               to build the value
+     * @param commit CT_COMMIT for commit action, CT_NO_COMMIT otherwise
+     * @param recircTable table number for recirculation after CT actions.
+     *                    CT_NO_RECIRC with no recirculation
+     * @param action Additional actions. ACTION_DROP, ACTION_NONE,
+     *               GOTO_XXX_TABLE are supported.
+     * @param priority priority value for the rule
+     * @param install true for insertion, false for removal
+     */
+    private void setConnTrackRule(DeviceId deviceId, long ctState, long ctMask,
+                                  int commit, short recircTable,
+                                  int action, int priority, boolean install) {
+
+        ExtensionSelector esCtSate = RulePopulatorUtil
+                .buildCtExtensionSelector(driverService, deviceId, ctState, ctMask);
+        TrafficSelector selector = DefaultTrafficSelector.builder()
+                .extension(esCtSate, deviceId)
+                .matchEthType(Ethernet.TYPE_IPV4)
+                .build();
+
+        TrafficTreatment.Builder tb = DefaultTrafficTreatment.builder();
+
+        if (commit == CT_COMMIT || recircTable > 0) {
+            RulePopulatorUtil.NiciraConnTrackTreatmentBuilder natTreatmentBuilder =
+                    niciraConnTrackTreatmentBuilder(driverService, deviceId);
+            natTreatmentBuilder.natAction(false);
+            natTreatmentBuilder.commit(commit == CT_COMMIT);
+            if (recircTable > 0) {
+                natTreatmentBuilder.table(recircTable);
+            }
+            tb.extension(natTreatmentBuilder.build(), deviceId);
+        } else if (action == ACTION_DROP) {
+            tb.drop();
+        }
+
+        if (action != ACTION_NONE && action != ACTION_DROP) {
+            tb.transition(action);
+        }
+
+        int tableType = ERROR_TABLE;
+        if (priority == PRIORITY_CT_RULE || priority == PRIORITY_CT_DROP_RULE) {
+            tableType = TENANT_ACL_CT_TABLE;
+        } else if (priority == PRIORITY_CT_HOOK_RULE) {
+            tableType = TENANT_ACL_INGRESS_TABLE;
+        } else {
+            log.error("Cannot an appropriate table for the conn track rule.");
+        }
+
+        flowRuleService.setRule(
+                appId,
+                deviceId,
+                selector,
+                tb.build(),
+                priority,
+                tableType,
+                install);
+    }
+
+    /**
+     * Returns a set of host IP addresses engaged with supplied security group ID.
+     * It only searches a VM in the same tenant boundary.
+     *
+     * @param srcPort edgestack port
+     * @param sgId security group id
+     * @return set of ip addresses
+     */
+    private Set<KubevirtPort> getRemotePorts(KubevirtPort srcPort, String sgId) {
+        return portService.ports().stream()
+                .filter(port -> !port.macAddress().equals(srcPort.macAddress()))
+                .filter(port -> port.securityGroups().contains(sgId))
+                .filter(port -> port.ipAddress() != null)
+                .collect(Collectors.toSet());
+    }
+
+    private Set<TrafficSelector> buildSelectors(KubevirtSecurityGroupRule sgRule,
+                                                Ip4Address vmIp,
+                                                IpPrefix remoteIp,
+                                                String netId) {
+        if (remoteIp != null && remoteIp.equals(IpPrefix.valueOf(vmIp, VM_IP_PREFIX))) {
+            // do nothing if the remote IP is my IP
+            return null;
+        }
+
+        Set<TrafficSelector> selectorSet = Sets.newHashSet();
+
+        if (sgRule.portRangeMax() != null && sgRule.portRangeMin() != null &&
+                sgRule.portRangeMin() < sgRule.portRangeMax()) {
+            Map<TpPort, TpPort> portRangeMatchMap =
+                    buildPortRangeMatches(sgRule.portRangeMin(),
+                            sgRule.portRangeMax());
+            portRangeMatchMap.forEach((key, value) -> {
+
+                TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder();
+                buildMatches(sBuilder, sgRule, vmIp, remoteIp, netId);
+
+                if (sgRule.protocol().equalsIgnoreCase(PROTO_TCP) ||
+                        sgRule.protocol().equals(PROTO_TCP_NUM)) {
+                    if (sgRule.direction().equalsIgnoreCase(EGRESS)) {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchTcpSrc(key);
+                        } else {
+                            sBuilder.matchTcpSrcMasked(key, value);
+                        }
+                    } else {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchTcpDst(key);
+                        } else {
+                            sBuilder.matchTcpDstMasked(key, value);
+                        }
+                    }
+                } else if (sgRule.protocol().equalsIgnoreCase(PROTO_UDP) ||
+                        sgRule.protocol().equals(PROTO_UDP_NUM)) {
+                    if (sgRule.direction().equalsIgnoreCase(EGRESS)) {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchUdpSrc(key);
+                        } else {
+                            sBuilder.matchUdpSrcMasked(key, value);
+                        }
+                    } else {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchUdpDst(key);
+                        } else {
+                            sBuilder.matchUdpDstMasked(key, value);
+                        }
+                    }
+                } else if (sgRule.protocol().equalsIgnoreCase(PROTO_SCTP) ||
+                        sgRule.protocol().equals(PROTO_SCTP_NUM)) {
+                    if (sgRule.direction().equalsIgnoreCase(EGRESS)) {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchSctpSrc(key);
+                        } else {
+                            sBuilder.matchSctpSrcMasked(key, value);
+                        }
+                    } else {
+                        if (value.toInt() == TpPort.MAX_PORT) {
+                            sBuilder.matchSctpDst(key);
+                        } else {
+                            sBuilder.matchSctpDstMasked(key, value);
+                        }
+                    }
+                }
+
+                selectorSet.add(sBuilder.build());
+            });
+        } else {
+
+            TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder();
+            buildMatches(sBuilder, sgRule, vmIp, remoteIp, netId);
+
+            selectorSet.add(sBuilder.build());
+        }
+
+        return selectorSet;
+    }
+
+    private void buildMatches(TrafficSelector.Builder sBuilder,
+                              KubevirtSecurityGroupRule sgRule, Ip4Address vmIp,
+                              IpPrefix remoteIp, String netId) {
+        buildTunnelId(sBuilder, netId);
+        buildMatchEthType(sBuilder, sgRule.etherType());
+        buildMatchDirection(sBuilder, sgRule.direction(), vmIp);
+        buildMatchProto(sBuilder, sgRule.protocol());
+        buildMatchPort(sBuilder, sgRule.protocol(), sgRule.direction(),
+                sgRule.portRangeMin() == null ? 0 : sgRule.portRangeMin(),
+                sgRule.portRangeMax() == null ? 0 : sgRule.portRangeMax());
+        buildMatchIcmp(sBuilder, sgRule.protocol(),
+                sgRule.portRangeMin(), sgRule.portRangeMax());
+        buildMatchRemoteIp(sBuilder, remoteIp, sgRule.direction());
+    }
+
+    private void buildTunnelId(TrafficSelector.Builder sBuilder, String netId) {
+        KubevirtNetwork network = networkService.network(netId);
+
+        if (network == null) {
+            log.warn("Network {} not found!", netId);
+            return;
+        }
+
+        String segId = network.segmentId();
+        Type netType = network.type();
+
+        if (netType == VLAN) {
+            sBuilder.matchVlanId(VlanId.vlanId(segId));
+        } else if (netType == VXLAN || netType == GRE || netType == GENEVE) {
+            // sBuilder.matchTunnelId(Long.parseLong(segId));
+            log.trace("{} typed match rules are installed for security group", netType);
+        } else {
+            log.debug("Cannot tag the VID as it is unsupported vnet type {}", netType);
+        }
+
+
+    }
+
+    private void buildMatchDirection(TrafficSelector.Builder sBuilder,
+                                     String direction,
+                                     Ip4Address vmIp) {
+        if (direction.equalsIgnoreCase(EGRESS)) {
+            sBuilder.matchIPSrc(IpPrefix.valueOf(vmIp, VM_IP_PREFIX));
+        } else {
+            sBuilder.matchIPDst(IpPrefix.valueOf(vmIp, VM_IP_PREFIX));
+        }
+    }
+
+    private void buildMatchEthType(TrafficSelector.Builder sBuilder, String etherType) {
+        // Either IpSrc or IpDst (or both) is set by default, and we need to set EthType as IPv4.
+        sBuilder.matchEthType(Ethernet.TYPE_IPV4);
+        if (etherType != null && !Objects.equals(etherType, STR_NULL) &&
+                !etherType.equalsIgnoreCase(ETHTYPE_IPV4)) {
+            log.debug("EthType {} is not supported yet in Security Group", etherType);
+        }
+    }
+
+    private void buildMatchRemoteIp(TrafficSelector.Builder sBuilder,
+                                    IpPrefix remoteIpPrefix, String direction) {
+        if (remoteIpPrefix != null &&
+                !remoteIpPrefix.getIp4Prefix().equals(IP_PREFIX_ANY)) {
+            if (direction.equalsIgnoreCase(EGRESS)) {
+                sBuilder.matchIPDst(remoteIpPrefix);
+            } else {
+                sBuilder.matchIPSrc(remoteIpPrefix);
+            }
+        }
+    }
+
+    private void buildMatchProto(TrafficSelector.Builder sBuilder, String protocol) {
+        if (protocol != null) {
+            switch (protocol.toUpperCase()) {
+                case PROTO_ICMP:
+                case PROTO_ICMP_NUM:
+                    sBuilder.matchIPProtocol(IPv4.PROTOCOL_ICMP);
+                    break;
+                case PROTO_TCP:
+                case PROTO_TCP_NUM:
+                    sBuilder.matchIPProtocol(IPv4.PROTOCOL_TCP);
+                    break;
+                case PROTO_UDP:
+                case PROTO_UDP_NUM:
+                    sBuilder.matchIPProtocol(IPv4.PROTOCOL_UDP);
+                    break;
+                case PROTO_SCTP:
+                case PROTO_SCTP_NUM:
+                    sBuilder.matchIPProtocol(PROTOCOL_SCTP);
+                    break;
+                default:
+                    break;
+            }
+        }
+    }
+
+    private void buildMatchPort(TrafficSelector.Builder sBuilder,
+                                String protocol, String direction,
+                                int portMin, int portMax) {
+        if (portMax > 0 && portMin == portMax) {
+            if (protocol.equalsIgnoreCase(PROTO_TCP) ||
+                    protocol.equals(PROTO_TCP_NUM)) {
+                if (direction.equalsIgnoreCase(EGRESS)) {
+                    sBuilder.matchTcpSrc(TpPort.tpPort(portMax));
+                } else {
+                    sBuilder.matchTcpDst(TpPort.tpPort(portMax));
+                }
+            } else if (protocol.equalsIgnoreCase(PROTO_UDP) ||
+                    protocol.equals(PROTO_UDP_NUM)) {
+                if (direction.equalsIgnoreCase(EGRESS)) {
+                    sBuilder.matchUdpSrc(TpPort.tpPort(portMax));
+                } else {
+                    sBuilder.matchUdpDst(TpPort.tpPort(portMax));
+                }
+            } else if (protocol.equalsIgnoreCase(PROTO_SCTP) ||
+                    protocol.equals(PROTO_SCTP_NUM)) {
+                if (direction.equalsIgnoreCase(EGRESS)) {
+                    sBuilder.matchSctpSrc(TpPort.tpPort(portMax));
+                } else {
+                    sBuilder.matchSctpDst(TpPort.tpPort(portMax));
+                }
+            }
+        }
+    }
+
+    private void buildMatchIcmp(TrafficSelector.Builder sBuilder,
+                                String protocol, Integer icmpCode, Integer icmpType) {
+        if (protocol != null) {
+            if (protocol.equalsIgnoreCase(PROTO_ICMP) ||
+                    protocol.equals(PROTO_ICMP_NUM)) {
+                if (icmpCode != null && icmpCode >= ICMP_CODE_MIN &&
+                        icmpCode <= ICMP_CODE_MAX) {
+                    sBuilder.matchIcmpCode(icmpCode.byteValue());
+                }
+                if (icmpType != null && icmpType >= ICMP_TYPE_MIN &&
+                        icmpType <= ICMP_TYPE_MAX) {
+                    sBuilder.matchIcmpType(icmpType.byteValue());
+                }
+            }
+        }
+    }
+
+    private void resetSecurityGroupRules() {
+
+        if (getUseSecurityGroupFlag()) {
+            nodeService.completeNodes(WORKER).forEach(node -> {
+                initializeEgressTable(node.intgBridge(), true);
+                initializeConnTrackTable(node.intgBridge(), true);
+                initializeAclTable(node.intgBridge(), true);
+                initializeIngressTable(node.intgBridge(), true);
+
+                for (KubevirtNetwork network : networkService.tenantNetworks()) {
+                    initializeEgressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeIngressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeConnTrackTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeAclTable(network.tenantDeviceId(node.hostname()), true);
+                }
+            });
+
+            securityGroupService.securityGroups().forEach(securityGroup ->
+                    securityGroup.rules().forEach(this::securityGroupRuleAdded));
+        } else {
+            nodeService.completeNodes(WORKER).forEach(node -> {
+                initializeEgressTable(node.intgBridge(), false);
+                initializeConnTrackTable(node.intgBridge(), false);
+                initializeAclTable(node.intgBridge(), false);
+                initializeIngressTable(node.intgBridge(), false);
+
+                for (KubevirtNetwork network : networkService.tenantNetworks()) {
+                    initializeEgressTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeIngressTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeConnTrackTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeAclTable(network.tenantDeviceId(node.hostname()), false);
+                }
+            });
+
+            securityGroupService.securityGroups().forEach(securityGroup ->
+                    securityGroup.rules().forEach(this::securityGroupRuleRemoved));
+        }
+
+        log.info("Reset security group info " +
+                (getUseSecurityGroupFlag() ? "with" : "without") + " Security Group");
+    }
+
+    private void securityGroupRuleAdded(KubevirtSecurityGroupRule sgRule) {
+        portService.ports().stream()
+                .filter(port -> port.securityGroups().contains(sgRule.securityGroupId()))
+                .forEach(port -> {
+                    updateSecurityGroupRule(port, sgRule, true);
+                    log.info("Applied security group rule {} to port {}",
+                            sgRule.id(), port.macAddress());
+                });
+    }
+
+    private void securityGroupRuleRemoved(KubevirtSecurityGroupRule sgRule) {
+        portService.ports().stream()
+                .filter(port -> port.securityGroups().contains(sgRule.securityGroupId()))
+                .forEach(port -> {
+                    updateSecurityGroupRule(port, sgRule, false);
+                    log.info("Removed security group rule {} from port {}",
+                            sgRule.id(), port.macAddress());
+                });
+    }
+
+    private class InternalKubevirtPortListener implements KubevirtPortListener {
+
+        @Override
+        public boolean isRelevant(KubevirtPortEvent event) {
+            return getUseSecurityGroupFlag();
+        }
+
+        private boolean isRelevantHelper(KubevirtPortEvent event) {
+            DeviceId deviceId = event.subject().deviceId();
+
+            if (deviceId == null) {
+                return false;
+            }
+
+            return mastershipService.isLocalMaster(deviceId);
+        }
+
+        @Override
+        public void event(KubevirtPortEvent event) {
+            log.debug("security group event received {}", event);
+
+            switch (event.type()) {
+                case KUBEVIRT_PORT_SECURITY_GROUP_ADDED:
+                    eventExecutor.execute(() -> processPortSgAdd(event));
+                    break;
+                case KUBEVIRT_PORT_SECURITY_GROUP_REMOVED:
+                    eventExecutor.execute(() -> processPortSgRemove(event));
+                    break;
+                case KUBEVIRT_PORT_REMOVED:
+                    eventExecutor.execute(() -> processPortRemove(event));
+                    break;
+                case KUBEVIRT_PORT_DEVICE_ADDED:
+                    eventExecutor.execute(() -> processPortDeviceAdded(event));
+                    break;
+                default:
+                    // do nothing for the other events
+                    break;
+            }
+        }
+
+        private void processPortSgAdd(KubevirtPortEvent event) {
+            if (!isRelevantHelper(event)) {
+                return;
+            }
+
+            if (event.securityGroupId() == null ||
+                securityGroupService.securityGroup(event.securityGroupId()) == null) {
+                return;
+            }
+
+            KubevirtPort port = event.subject();
+            KubevirtSecurityGroup sg = securityGroupService.securityGroup(event.securityGroupId());
+
+            sg.rules().forEach(sgRule -> {
+                updateSecurityGroupRule(port, sgRule, true);
+            });
+            log.info("Added security group {} to port {}",
+                    event.securityGroupId(), event.subject().macAddress());
+        }
+
+        private void processPortSgRemove(KubevirtPortEvent event) {
+            if (!isRelevantHelper(event)) {
+                return;
+            }
+
+            if (event.securityGroupId() == null ||
+                securityGroupService.securityGroup(event.securityGroupId()) == null) {
+                return;
+            }
+
+            KubevirtPort port = event.subject();
+            KubevirtSecurityGroup sg = securityGroupService.securityGroup(event.securityGroupId());
+
+            sg.rules().forEach(sgRule -> {
+                updateSecurityGroupRule(port, sgRule, false);
+            });
+            log.info("Removed security group {} from port {}",
+                    event.securityGroupId(), event.subject().macAddress());
+        }
+
+        private void processPortRemove(KubevirtPortEvent event) {
+            if (!isRelevantHelper(event)) {
+                return;
+            }
+
+            KubevirtPort port = event.subject();
+            for (String sgStr : port.securityGroups()) {
+                KubevirtSecurityGroup sg = securityGroupService.securityGroup(sgStr);
+                sg.rules().forEach(sgRule -> {
+                    updateSecurityGroupRule(port, sgRule, false);
+                });
+                log.info("Removed security group {} from port {}",
+                                        sgStr, event.subject().macAddress());
+            }
+        }
+
+        private void processPortDeviceAdded(KubevirtPortEvent event) {
+            if (!isRelevantHelper(event)) {
+                return;
+            }
+
+            for (String sgId : event.subject().securityGroups()) {
+                KubevirtSecurityGroup sg = securityGroupService.securityGroup(sgId);
+
+                sg.rules().forEach(sgRule -> {
+                    updateSecurityGroupRule(event.subject(), sgRule, true);
+                });
+                log.info("Added security group {} to port {}",
+                        event.securityGroupId(), event.subject().macAddress());
+            }
+        }
+    }
+
+    private class InternalNetworkListener implements KubevirtNetworkListener {
+
+        private boolean isRelevantHelper() {
+            return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+        }
+
+        @Override
+        public void event(KubevirtNetworkEvent event) {
+            switch (event.type()) {
+                case KUBEVIRT_NETWORK_CREATED:
+                    eventExecutor.execute(() -> processNetworkCreation(event.subject()));
+                    break;
+                case KUBEVIRT_NETWORK_REMOVED:
+                case KUBEVIRT_NETWORK_UPDATED:
+                default:
+                    // do thing
+                    break;
+            }
+        }
+
+        private void processNetworkCreation(KubevirtNetwork network) {
+            if (!isRelevantHelper()) {
+                return;
+            }
+
+            Set<KubevirtNode> nodes = nodeService.completeNodes(WORKER);
+
+            if (nodes.size() > 0) {
+                // now we wait 5s for all tenant bridges are created,
+                // FIXME: we need to fina a better way to wait all tenant bridges
+                // are created before installing default security group rules
+                try {
+                    sleep(SLEEP_MS);
+                } catch (InterruptedException e) {
+                    log.error("Failed to install security group default rules.");
+                }
+
+                for (KubevirtNode node : nodes) {
+                    initializeEgressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeIngressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeConnTrackTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeAclTable(network.tenantDeviceId(node.hostname()), true);
+                }
+            }
+        }
+    }
+
+    private class InternalSecurityGroupListener implements KubevirtSecurityGroupListener {
+
+        @Override
+        public boolean isRelevant(KubevirtSecurityGroupEvent event) {
+            return getUseSecurityGroupFlag();
+        }
+
+        private boolean isRelevantHelper() {
+            return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+        }
+
+        @Override
+        public void event(KubevirtSecurityGroupEvent event) {
+            switch (event.type()) {
+                case KUBEVIRT_SECURITY_GROUP_RULE_CREATED:
+                    eventExecutor.execute(() -> processSgRuleCreate(event));
+                    break;
+                case KUBEVIRT_SECURITY_GROUP_RULE_REMOVED:
+                    eventExecutor.execute(() -> processSgRuleRemove(event));
+                    break;
+                default:
+                    // do nothing
+                    break;
+            }
+        }
+
+        private void processSgRuleCreate(KubevirtSecurityGroupEvent event) {
+            if (!isRelevantHelper()) {
+                return;
+            }
+
+            KubevirtSecurityGroupRule sgRuleToAdd = event.rule();
+            securityGroupRuleAdded(sgRuleToAdd);
+            log.info("Applied new security group rule {} to ports", sgRuleToAdd.id());
+        }
+
+        private void processSgRuleRemove(KubevirtSecurityGroupEvent event) {
+            if (!isRelevantHelper()) {
+                return;
+            }
+
+            KubevirtSecurityGroupRule sgRuleToRemove = event.rule();
+            securityGroupRuleRemoved(sgRuleToRemove);
+            log.info("Removed security group rule {} from ports", sgRuleToRemove.id());
+        }
+    }
+
+    private class InternalNodeListener implements KubevirtNodeListener {
+
+        @Override
+        public boolean isRelevant(KubevirtNodeEvent event) {
+            return event.subject().type() == WORKER;
+        }
+
+        private boolean isRelevantHelper() {
+            return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+        }
+
+        @Override
+        public void event(KubevirtNodeEvent event) {
+            switch (event.type()) {
+                case KUBEVIRT_NODE_COMPLETE:
+                    eventExecutor.execute(() -> processNodeComplete(event.subject()));
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        private void processNodeComplete(KubevirtNode node) {
+            if (!isRelevantHelper()) {
+                return;
+            }
+
+            resetSecurityGroupRulesByNode(node);
+        }
+
+        private void resetSecurityGroupRulesByNode(KubevirtNode node) {
+            if (getUseSecurityGroupFlag()) {
+                initializeEgressTable(node.intgBridge(), true);
+                initializeConnTrackTable(node.intgBridge(), true);
+                initializeAclTable(node.intgBridge(), true);
+                initializeIngressTable(node.intgBridge(), true);
+
+                for (KubevirtNetwork network : networkService.tenantNetworks()) {
+                    initializeEgressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeIngressTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeConnTrackTable(network.tenantDeviceId(node.hostname()), true);
+                    initializeAclTable(network.tenantDeviceId(node.hostname()), true);
+                }
+
+                securityGroupService.securityGroups().forEach(securityGroup ->
+                        securityGroup.rules().forEach(
+                                KubevirtSecurityGroupHandler.this::securityGroupRuleAdded));
+            } else {
+                initializeEgressTable(node.intgBridge(), false);
+                initializeConnTrackTable(node.intgBridge(), false);
+                initializeAclTable(node.intgBridge(), false);
+                initializeIngressTable(node.intgBridge(), false);
+
+                for (KubevirtNetwork network : networkService.tenantNetworks()) {
+                    initializeEgressTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeIngressTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeConnTrackTable(network.tenantDeviceId(node.hostname()), false);
+                    initializeAclTable(network.tenantDeviceId(node.hostname()), false);
+                }
+
+                securityGroupService.securityGroups().forEach(securityGroup ->
+                        securityGroup.rules().forEach(
+                                KubevirtSecurityGroupHandler.this::securityGroupRuleRemoved));
+            }
+
+            log.info("Reset security group info " +
+                    (getUseSecurityGroupFlag() ? "with" : "without") + " Security Group");
+        }
+    }
+}
\ No newline at end of file