| /* |
| * Copyright 2019-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.k8snetworking.impl; |
| |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import io.fabric8.kubernetes.api.model.LabelSelectorRequirement; |
| import io.fabric8.kubernetes.api.model.Namespace; |
| import io.fabric8.kubernetes.api.model.Pod; |
| import io.fabric8.kubernetes.api.model.Service; |
| |
| import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicy; |
| import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyEgressRule; |
| import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyIngressRule; |
| import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyPeer; |
| import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyPort; |
| import org.onlab.packet.IPv4; |
| import org.onlab.packet.IpAddress; |
| import org.onlab.packet.IpPrefix; |
| import org.onlab.packet.TpPort; |
| import org.onosproject.cfg.ComponentConfigService; |
| 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.k8snetworking.api.K8sFlowRuleService; |
| import org.onosproject.k8snetworking.api.K8sNamespaceEvent; |
| import org.onosproject.k8snetworking.api.K8sNamespaceListener; |
| import org.onosproject.k8snetworking.api.K8sNamespaceService; |
| import org.onosproject.k8snetworking.api.K8sNetworkPolicyEvent; |
| import org.onosproject.k8snetworking.api.K8sNetworkPolicyListener; |
| import org.onosproject.k8snetworking.api.K8sNetworkPolicyService; |
| import org.onosproject.k8snetworking.api.K8sNetworkService; |
| import org.onosproject.k8snetworking.api.K8sPodEvent; |
| import org.onosproject.k8snetworking.api.K8sPodListener; |
| import org.onosproject.k8snetworking.api.K8sPodService; |
| import org.onosproject.k8snetworking.api.K8sServiceEvent; |
| import org.onosproject.k8snetworking.api.K8sServiceListener; |
| import org.onosproject.k8snetworking.api.K8sServiceService; |
| import org.onosproject.k8snode.api.K8sNodeService; |
| 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.store.service.StorageService; |
| 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.Reference; |
| import org.osgi.service.component.annotations.ReferenceCardinality; |
| import org.slf4j.Logger; |
| |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import static java.util.concurrent.Executors.newSingleThreadExecutor; |
| import static org.onlab.packet.Ethernet.TYPE_IPV4; |
| import static org.onlab.util.Tools.groupedThreads; |
| import static org.onosproject.k8snetworking.api.Constants.ACL_EGRESS_BLACK_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.ACL_EGRESS_WHITE_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.ACL_INGRESS_BLACK_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.ACL_INGRESS_WHITE_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.ACL_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.DEFAULT_METADATA_MASK; |
| import static org.onosproject.k8snetworking.api.Constants.DEFAULT_NAMESPACE_HASH; |
| import static org.onosproject.k8snetworking.api.Constants.DEFAULT_SEGMENT_ID; |
| import static org.onosproject.k8snetworking.api.Constants.DEFAULT_SERVICE_IP_NONE; |
| import static org.onosproject.k8snetworking.api.Constants.GROUPING_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.K8S_NETWORKING_APP_ID; |
| import static org.onosproject.k8snetworking.api.Constants.NAMESPACE_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.PRIORITY_CIDR_RULE; |
| import static org.onosproject.k8snetworking.api.Constants.PRIORITY_NAMESPACE_RULE; |
| import static org.onosproject.k8snetworking.api.Constants.ROUTING_TABLE; |
| import static org.onosproject.k8snetworking.api.Constants.SHIFTED_IP_PREFIX; |
| import static org.onosproject.k8snetworking.impl.OsgiPropertyConstants.SERVICE_IP_CIDR_DEFAULT; |
| import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.namespaceHashByNamespace; |
| import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.namespaceHashByPodIp; |
| import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.namespaceHashByServiceIp; |
| import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.shiftIpDomain; |
| import static org.slf4j.LoggerFactory.getLogger; |
| |
| /** |
| * Handles the ACL by referring to the network policy defined through kubernetes. |
| */ |
| @Component(immediate = true) |
| public class K8sNetworkPolicyHandler { |
| |
| private final Logger log = getLogger(getClass()); |
| |
| private static final String DIRECTION_INGRESS = "ingress"; |
| private static final String DIRECTION_EGRESS = "egress"; |
| |
| private static final String PROTOCOL_TCP = "tcp"; |
| private static final String PROTOCOL_UDP = "udp"; |
| |
| private static final String KUBE_SYSTEM = "kube-system"; |
| |
| private static final int HOST_PREFIX = 32; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected CoreService coreService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected LeadershipService leadershipService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected ClusterService clusterService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected DriverService driverService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected DeviceService deviceService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected ComponentConfigService configService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected StorageService storageService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sNetworkService k8sNetworkService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sFlowRuleService k8sFlowRuleService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sNodeService k8sNodeService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sPodService k8sPodService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sServiceService k8sServiceService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sNetworkPolicyService k8sNetworkPolicyService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected K8sNamespaceService k8sNamespaceService; |
| |
| private final ExecutorService eventExecutor = newSingleThreadExecutor( |
| groupedThreads(this.getClass().getSimpleName(), "event-handler", log)); |
| private final InternalPodListener internalPodListener = |
| new InternalPodListener(); |
| private final InternalServiceListener internalServiceListener = |
| new InternalServiceListener(); |
| private final InternalNetworkPolicyListener internalNetworkPolicyListener = |
| new InternalNetworkPolicyListener(); |
| private final InternalNamespaceListener internalNamespaceListener = |
| new InternalNamespaceListener(); |
| |
| private ApplicationId appId; |
| private NodeId localNodeId; |
| |
| @Activate |
| protected void activate() { |
| appId = coreService.registerApplication(K8S_NETWORKING_APP_ID); |
| |
| localNodeId = clusterService.getLocalNode().id(); |
| leadershipService.runForLeadership(appId.name()); |
| k8sPodService.addListener(internalPodListener); |
| k8sServiceService.addListener(internalServiceListener); |
| k8sNetworkPolicyService.addListener(internalNetworkPolicyListener); |
| k8sNamespaceService.addListener(internalNamespaceListener); |
| |
| log.info("Started"); |
| } |
| |
| @Deactivate |
| protected void deactivate() { |
| leadershipService.withdraw(appId.name()); |
| k8sPodService.removeListener(internalPodListener); |
| k8sServiceService.removeListener(internalServiceListener); |
| k8sNetworkPolicyService.removeListener(internalNetworkPolicyListener); |
| k8sNamespaceService.removeListener(internalNamespaceListener); |
| eventExecutor.shutdown(); |
| |
| log.info("Stopped"); |
| } |
| |
| private void setBlockRulesByPolicy(NetworkPolicy policy, boolean install) { |
| final Map<String, List<String>> filter = Maps.newConcurrentMap(); |
| |
| k8sPodService.pods().forEach(pod -> |
| filter.putAll(getBlockRuleFilter(pod, policy))); |
| |
| setBlockRules(filter, install); |
| } |
| |
| private void setBlockRulesByPod(Pod pod, boolean install) { |
| final Map<String, List<String>> filter = Maps.newConcurrentMap(); |
| |
| k8sNetworkPolicyService.networkPolicies().forEach(policy -> |
| filter.putAll(getBlockRuleFilter(pod, policy))); |
| |
| setBlockRules(filter, install); |
| } |
| |
| private Map<String, List<String>> getBlockRuleFilter(Pod pod, NetworkPolicy policy) { |
| |
| // if the POD is not included in the namespace of the given policy, |
| // we do not block the POD |
| if (!pod.getMetadata().getNamespace().equals(policy.getMetadata().getNamespace())) { |
| return Maps.newConcurrentMap(); |
| } |
| |
| Map<String, String> labels = policy.getSpec().getPodSelector().getMatchLabels(); |
| Map<String, List<String>> filter = Maps.newConcurrentMap(); |
| String podIp = pod.getStatus().getPodIP(); |
| List<String> policyTypes = policy.getSpec().getPolicyTypes(); |
| |
| if (podIp != null && policyTypes != null) { |
| if (labels == null) { |
| filter.put(podIp, policyTypes); |
| } else { |
| pod.getMetadata().getLabels().forEach((k, v) -> { |
| if (labels.get(k) != null && labels.get(k).equals(v)) { |
| filter.put(podIp, policyTypes); |
| } |
| }); |
| } |
| } |
| |
| return filter; |
| } |
| |
| private void setBlockRules(Map<String, List<String>> filter, boolean install) { |
| filter.forEach((k, v) -> { |
| v.forEach(d -> { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder() |
| .matchEthType(TYPE_IPV4); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder(); |
| |
| Integer nsHash = namespaceHashByPodIp(k8sPodService, k8sNamespaceService, k); |
| if (nsHash != null) { |
| tBuilder.setTunnelId(nsHash); |
| } |
| |
| if (d.equalsIgnoreCase(DIRECTION_INGRESS)) { |
| sBuilder.matchIPDst(IpPrefix.valueOf(IpAddress.valueOf(k), HOST_PREFIX)); |
| tBuilder.transition(ACL_INGRESS_WHITE_TABLE); |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_TABLE, install); |
| } else if (d.equalsIgnoreCase(DIRECTION_EGRESS)) { |
| // original IP |
| sBuilder.matchIPSrc(IpPrefix.valueOf(IpAddress.valueOf(k), HOST_PREFIX)); |
| tBuilder.transition(ACL_EGRESS_WHITE_TABLE); |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_TABLE, install); |
| |
| |
| // shifted IP |
| sBuilder.matchIPSrc(IpPrefix.valueOf(IpAddress.valueOf( |
| shiftIpDomain(k, SHIFTED_IP_PREFIX)), HOST_PREFIX)); |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_TABLE, install); |
| } |
| }); |
| }); |
| } |
| |
| private void setDefaultAllowNamespaceRules(Namespace namespace, boolean install) { |
| |
| String ns = namespace.getMetadata().getName(); |
| if (KUBE_SYSTEM.equalsIgnoreCase(ns)) { |
| setAllowNamespaceRulesBase(0, namespace.hashCode(), |
| DIRECTION_INGRESS, install); |
| } |
| } |
| |
| private void setDefaultAllowServiceRules(boolean install) { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder() |
| .matchEthType(TYPE_IPV4) |
| .matchIPSrc(IpPrefix.valueOf(SERVICE_IP_CIDR_DEFAULT)); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder() |
| .setTunnelId(DEFAULT_SEGMENT_ID) |
| .transition(ROUTING_TABLE); |
| |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_INGRESS_WHITE_TABLE, install); |
| } |
| |
| private void setAllowNamespaceRulesBase(int tunnelId, int metadata, |
| String direction, boolean install) { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder(); |
| |
| if (tunnelId != 0) { |
| sBuilder.matchTunnelId(tunnelId); |
| } |
| |
| sBuilder.matchMetadata(metadata); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder() |
| .setTunnelId(DEFAULT_SEGMENT_ID) |
| .transition(ROUTING_TABLE); |
| |
| if (DIRECTION_INGRESS.equals(direction)) { |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_INGRESS_WHITE_TABLE, install); |
| } |
| } |
| |
| private void setAllowRulesByPolicy(NetworkPolicy policy, boolean install) { |
| Map<String, Map<String, List<NetworkPolicyPort>>> |
| white = Maps.newConcurrentMap(); |
| |
| int nsHash = namespaceHashByNamespace(k8sNamespaceService, |
| policy.getMetadata().getNamespace()); |
| |
| List<NetworkPolicyIngressRule> ingress = policy.getSpec().getIngress(); |
| if (ingress != null && ingress.size() == 1) { |
| NetworkPolicyIngressRule rule = ingress.get(0); |
| if (rule.getFrom().size() == 0 && rule.getPorts().size() == 0) { |
| setAllowAllRule(nsHash, DIRECTION_INGRESS, install); |
| } |
| } |
| |
| policy.getSpec().getIngress().forEach(i -> { |
| Map<String, List<NetworkPolicyPort>> direction = Maps.newConcurrentMap(); |
| direction.put(DIRECTION_INGRESS, i.getPorts()); |
| i.getFrom().forEach(peer -> { |
| |
| // IP block |
| if (peer.getIpBlock() != null) { |
| if (peer.getIpBlock().getExcept() != null && |
| peer.getIpBlock().getExcept().size() > 0) { |
| Map<String, List<NetworkPolicyPort>> |
| blkDirection = Maps.newConcurrentMap(); |
| |
| blkDirection.put(DIRECTION_INGRESS, i.getPorts()); |
| white.compute(peer.getIpBlock().getCidr(), (k, v) -> blkDirection); |
| |
| setBlackRules(peer.getIpBlock().getCidr(), DIRECTION_INGRESS, |
| peer.getIpBlock().getExcept(), install); |
| } else { |
| white.compute(peer.getIpBlock().getCidr(), (k, v) -> direction); |
| } |
| } |
| |
| // POD selector |
| Set<Pod> pods = podsFromPolicyPeer(peer, policy.getMetadata().getNamespace()); |
| |
| pods.stream() |
| .filter(pod -> pod.getStatus().getPodIP() != null) |
| .forEach(pod -> { |
| white.compute(shiftIpDomain(pod.getStatus().getPodIP(), |
| SHIFTED_IP_PREFIX) + "/" + HOST_PREFIX, (m, n) -> direction); |
| white.compute(pod.getStatus().getPodIP() + "/" + HOST_PREFIX, (m, n) -> direction); |
| }); |
| |
| // Namespace selector |
| setAllowNamespaceRules(nsHash, |
| namespacesByPolicyPeer(peer), DIRECTION_INGRESS, install); |
| }); |
| }); |
| |
| List<NetworkPolicyEgressRule> egress = policy.getSpec().getEgress(); |
| if (egress != null && egress.size() == 1) { |
| NetworkPolicyEgressRule rule = egress.get(0); |
| if (rule.getTo().size() == 0 && rule.getPorts().size() == 0) { |
| setAllowAllRule(nsHash, DIRECTION_EGRESS, install); |
| } |
| } |
| |
| policy.getSpec().getEgress().forEach(e -> { |
| Map<String, List<NetworkPolicyPort>> direction = Maps.newConcurrentMap(); |
| direction.put(DIRECTION_EGRESS, e.getPorts()); |
| e.getTo().forEach(peer -> { |
| |
| // IP block |
| if (peer.getIpBlock() != null) { |
| if (peer.getIpBlock().getExcept() != null && |
| peer.getIpBlock().getExcept().size() > 0) { |
| |
| Map<String, List<NetworkPolicyPort>> |
| blkDirection = Maps.newConcurrentMap(); |
| blkDirection.put(DIRECTION_EGRESS, e.getPorts()); |
| white.compute(peer.getIpBlock().getCidr(), (k, v) -> { |
| if (v != null) { |
| v.put(DIRECTION_EGRESS, e.getPorts()); |
| return v; |
| } else { |
| return blkDirection; |
| } |
| }); |
| |
| setBlackRules(peer.getIpBlock().getCidr(), DIRECTION_EGRESS, |
| peer.getIpBlock().getExcept(), install); |
| } else { |
| white.compute(peer.getIpBlock().getCidr(), (k, v) -> { |
| if (v != null) { |
| v.put(DIRECTION_EGRESS, e.getPorts()); |
| return v; |
| } else { |
| return direction; |
| } |
| }); |
| } |
| } |
| |
| // POD selector |
| Set<Pod> pods = podsFromPolicyPeer(peer, policy.getMetadata().getNamespace()); |
| |
| pods.stream() |
| .filter(pod -> pod.getStatus().getPodIP() != null) |
| .forEach(pod -> { |
| white.compute(shiftIpDomain(pod.getStatus().getPodIP(), |
| SHIFTED_IP_PREFIX) + "/" + HOST_PREFIX, (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| |
| white.compute(pod.getStatus().getPodIP() + "/" + HOST_PREFIX, |
| (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| }); |
| |
| // Namespace selector |
| setAllowNamespaceRules(nsHash, |
| namespacesByPolicyPeer(peer), DIRECTION_EGRESS, install); |
| }); |
| }); |
| |
| setAllowRules(namespaceHashByNamespace(k8sNamespaceService, |
| policy.getMetadata().getNamespace()), white, install); |
| setBlackToRouteRules(true); |
| } |
| |
| private Set<Pod> podsFromPolicyPeer(NetworkPolicyPeer peer, String namespace) { |
| Set<Pod> pods = Sets.newConcurrentHashSet(); |
| if (peer.getPodSelector() != null) { |
| Map<String, String> podLabels = peer.getPodSelector().getMatchLabels(); |
| List<LabelSelectorRequirement> matchExps = peer.getPodSelector().getMatchExpressions(); |
| |
| if (podLabels == null && matchExps.size() == 0) { |
| k8sPodService.pods().stream() |
| .filter(pod -> pod.getMetadata().getNamespace().equals( |
| namespace)) |
| .forEach(pods::add); |
| } else { |
| k8sPodService.pods().stream() |
| .filter(pod -> pod.getMetadata().getNamespace().equals( |
| namespace)) |
| .forEach(pod -> { |
| pod.getMetadata().getLabels().forEach((k, v) -> { |
| if (podLabels != null && podLabels.get(k) != null && |
| podLabels.get(k).equals(v)) { |
| pods.add(pod); |
| } |
| }); |
| }); |
| } |
| } |
| return pods; |
| } |
| |
| private void setAllowRulesByPod(Pod pod, boolean install) { |
| Map<String, Map<String, List<NetworkPolicyPort>>> |
| white = Maps.newConcurrentMap(); |
| AtomicReference<NetworkPolicy> selectedPolicy = new AtomicReference<>(); |
| k8sNetworkPolicyService.networkPolicies().stream() |
| .filter(policy -> policy.getMetadata().getNamespace().equals( |
| pod.getMetadata().getNamespace())) |
| .forEach(policy -> { |
| String podIp = pod.getStatus().getPodIP(); |
| |
| policy.getSpec().getIngress().forEach(i -> { |
| Map<String, List<NetworkPolicyPort>> |
| direction = Maps.newConcurrentMap(); |
| direction.put(DIRECTION_INGRESS, i.getPorts()); |
| i.getFrom().forEach(peer -> { |
| if (peer.getPodSelector() != null) { |
| Map<String, String> podLabels = peer.getPodSelector().getMatchLabels(); |
| List<LabelSelectorRequirement> matchExps = peer.getPodSelector().getMatchExpressions(); |
| |
| if (podLabels == null && matchExps.size() == 0 && podIp != null) { |
| white.compute(shiftIpDomain(podIp, SHIFTED_IP_PREFIX) + |
| "/" + HOST_PREFIX, (m, n) -> direction); |
| white.compute(podIp + "/" + |
| HOST_PREFIX, (m, n) -> direction); |
| |
| selectedPolicy.set(policy); |
| } else { |
| pod.getMetadata().getLabels().forEach((k, v) -> { |
| if (podLabels != null && podLabels.get(k) != null && |
| podLabels.get(k).equals(v) && podIp != null) { |
| white.compute(shiftIpDomain(podIp, SHIFTED_IP_PREFIX) + |
| "/" + HOST_PREFIX, (m, n) -> direction); |
| white.compute(podIp + "/" + |
| HOST_PREFIX, (m, n) -> direction); |
| |
| selectedPolicy.set(policy); |
| } |
| }); |
| } |
| } |
| }); |
| }); |
| }); |
| |
| k8sNetworkPolicyService.networkPolicies().stream() |
| .filter(policy -> policy.getMetadata().getNamespace().equals( |
| pod.getMetadata().getNamespace())) |
| .forEach(policy -> { |
| String podIp = pod.getStatus().getPodIP(); |
| |
| policy.getSpec().getEgress().forEach(e -> { |
| Map<String, List<NetworkPolicyPort>> direction = Maps.newConcurrentMap(); |
| direction.put(DIRECTION_EGRESS, e.getPorts()); |
| e.getTo().forEach(peer -> { |
| if (peer.getPodSelector() != null) { |
| Map<String, String> podLabels = peer.getPodSelector().getMatchLabels(); |
| List<LabelSelectorRequirement> matchExps = peer.getPodSelector().getMatchExpressions(); |
| |
| if (podLabels == null && matchExps.size() == 0 && podIp != null) { |
| white.compute(shiftIpDomain(podIp, SHIFTED_IP_PREFIX) + |
| "/" + HOST_PREFIX, (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| |
| white.compute(podIp + "/" + |
| HOST_PREFIX, (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| |
| selectedPolicy.set(policy); |
| } else { |
| pod.getMetadata().getLabels().forEach((k, v) -> { |
| if (podLabels != null && podLabels.get(k) != null && |
| podLabels.get(k).equals(v) && podIp != null) { |
| white.compute(shiftIpDomain(podIp, SHIFTED_IP_PREFIX) + |
| "/" + HOST_PREFIX, (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| |
| white.compute(podIp + "/" + |
| HOST_PREFIX, (m, n) -> { |
| if (n != null) { |
| n.put(DIRECTION_EGRESS, e.getPorts()); |
| return n; |
| } else { |
| return direction; |
| } |
| }); |
| |
| selectedPolicy.set(policy); |
| } |
| }); |
| } |
| } |
| }); |
| }); |
| }); |
| |
| int nsHash = selectedPolicy.get() != null ? namespaceHashByNamespace(k8sNamespaceService, |
| selectedPolicy.get().getMetadata().getNamespace()) : DEFAULT_NAMESPACE_HASH; |
| |
| setAllowRules(nsHash, white, install); |
| setBlackToRouteRules(true); |
| } |
| |
| private Set<Namespace> namespacesByLabels(Map<String, String> labels) { |
| Set<Namespace> nsSet = Sets.newConcurrentHashSet(); |
| k8sNamespaceService.namespaces().forEach(ns -> { |
| if (ns != null && ns.getMetadata() != null && |
| ns.getMetadata().getLabels() != null && labels != null) { |
| ns.getMetadata().getLabels().forEach((k, v) -> { |
| if (labels.get(k) != null && labels.get(k).equals(v)) { |
| nsSet.add(ns); |
| } |
| }); |
| } |
| }); |
| |
| return nsSet; |
| } |
| |
| private Set<Namespace> namespacesByPolicyPeer(NetworkPolicyPeer peer) { |
| if (peer.getNamespaceSelector() != null) { |
| Map<String, String> labels = peer.getNamespaceSelector().getMatchLabels(); |
| if (labels == null || labels.size() == 0) { |
| // if none of match labels are specified, it means the |
| // target PODs are from any namespaces |
| return k8sNamespaceService.namespaces(); |
| } else { |
| return namespacesByLabels(labels); |
| } |
| } |
| |
| return Sets.newConcurrentHashSet(); |
| } |
| |
| private void setAllowNamespaceRules(int tunnelId, Set<Namespace> nsSet, |
| String direction, boolean install) { |
| |
| nsSet.forEach(ns -> { |
| setAllowNamespaceRulesBase(tunnelId, ns.hashCode(), direction, install); |
| }); |
| } |
| |
| private void setAllowAllRule(int nsHash, String direction, boolean install) { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder() |
| .matchTunnelId(nsHash); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder() |
| .setTunnelId(DEFAULT_SEGMENT_ID) |
| .transition(ROUTING_TABLE); |
| |
| int table = 0; |
| |
| if (DIRECTION_INGRESS.equalsIgnoreCase(direction)) { |
| table = ACL_INGRESS_WHITE_TABLE; |
| } |
| |
| setPolicyRulesBase(sBuilder, tBuilder, table, install); |
| } |
| |
| private void setAllowRules(int namespaceHash, |
| Map<String, Map<String, List<NetworkPolicyPort>>> white, |
| boolean install) { |
| white.forEach((k, v) -> { |
| v.forEach((pk, pv) -> { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder() |
| .matchTunnelId(namespaceHash) |
| .matchEthType(TYPE_IPV4); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder(); |
| tBuilder.setTunnelId(DEFAULT_SEGMENT_ID); |
| |
| if (pk.equalsIgnoreCase(DIRECTION_INGRESS)) { |
| sBuilder.matchIPSrc(IpPrefix.valueOf(k)); |
| tBuilder.writeMetadata(k.hashCode(), DEFAULT_METADATA_MASK) |
| .transition(ACL_INGRESS_BLACK_TABLE); |
| |
| if (pv.size() == 0) { |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_INGRESS_WHITE_TABLE, install); |
| } else { |
| pv.forEach(p -> { |
| if (p.getProtocol().equalsIgnoreCase(PROTOCOL_TCP)) { |
| sBuilder.matchIPProtocol(IPv4.PROTOCOL_TCP); |
| |
| if (p.getPort() != null && |
| p.getPort().getIntVal() != null) { |
| sBuilder.matchTcpDst(TpPort.tpPort(p.getPort().getIntVal())); |
| } |
| } |
| if (p.getProtocol().equalsIgnoreCase(PROTOCOL_UDP)) { |
| sBuilder.matchIPProtocol(IPv4.PROTOCOL_UDP); |
| |
| if (p.getPort() != null && |
| p.getPort().getIntVal() != null) { |
| sBuilder.matchUdpDst(TpPort.tpPort(p.getPort().getIntVal())); |
| } |
| } |
| |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_INGRESS_WHITE_TABLE, install); |
| }); |
| } |
| } else if (pk.equalsIgnoreCase(DIRECTION_EGRESS)) { |
| sBuilder.matchIPDst(IpPrefix.valueOf(k)); |
| tBuilder.writeMetadata(k.hashCode(), DEFAULT_METADATA_MASK) |
| .transition(ACL_EGRESS_BLACK_TABLE); |
| |
| if (pv.size() == 0) { |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_EGRESS_WHITE_TABLE, install); |
| } else { |
| pv.forEach(p -> { |
| if (p.getProtocol().equalsIgnoreCase(PROTOCOL_TCP)) { |
| sBuilder.matchIPProtocol(IPv4.PROTOCOL_TCP); |
| |
| if (p.getPort() != null && |
| p.getPort().getIntVal() != null) { |
| sBuilder.matchTcpSrc(TpPort.tpPort(p.getPort().getIntVal())); |
| } |
| } |
| if (p.getProtocol().equalsIgnoreCase(PROTOCOL_UDP)) { |
| sBuilder.matchIPProtocol(IPv4.PROTOCOL_UDP); |
| |
| if (p.getPort() != null && |
| p.getPort().getIntVal() != null) { |
| sBuilder.matchUdpSrc(TpPort.tpPort(p.getPort().getIntVal())); |
| } |
| } |
| setPolicyRulesBase(sBuilder, tBuilder, ACL_EGRESS_WHITE_TABLE, install); |
| }); |
| } |
| |
| } else { |
| log.error("In correct direction has been specified at network policy."); |
| } |
| }); |
| }); |
| } |
| |
| private void setPolicyRulesBase(TrafficSelector.Builder sBuilder, |
| TrafficTreatment.Builder tBuilder, |
| int table, |
| boolean install) { |
| k8sNodeService.completeNodes().forEach(n -> { |
| k8sFlowRuleService.setRule( |
| appId, |
| n.intgBridge(), |
| sBuilder.build(), |
| tBuilder.build(), |
| PRIORITY_CIDR_RULE, |
| table, |
| install |
| ); |
| }); |
| } |
| |
| private void setBlackRules(String whiteIpCidr, String direction, |
| List<String> except, boolean install) { |
| k8sNodeService.completeNodes().forEach(n -> { |
| except.forEach(blkIp -> { |
| TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder() |
| .matchEthType(TYPE_IPV4) |
| .matchMetadata(whiteIpCidr.hashCode()); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder() |
| .drop(); |
| int table = 0; |
| if (direction.equalsIgnoreCase(DIRECTION_INGRESS)) { |
| sBuilder.matchIPSrc(IpPrefix.valueOf(blkIp)); |
| table = ACL_INGRESS_BLACK_TABLE; |
| } |
| if (direction.equalsIgnoreCase(DIRECTION_EGRESS)) { |
| sBuilder.matchIPDst(IpPrefix.valueOf(blkIp)); |
| table = ACL_EGRESS_BLACK_TABLE; |
| } |
| |
| setPolicyRulesBase(sBuilder, tBuilder, table, install); |
| }); |
| }); |
| } |
| |
| private void setBlackToRouteRules(boolean install) { |
| |
| k8sNodeService.completeNodes().forEach(n -> { |
| ImmutableSet.of(ACL_INGRESS_BLACK_TABLE, ACL_EGRESS_BLACK_TABLE).forEach(t -> { |
| k8sFlowRuleService.setRule( |
| appId, |
| n.intgBridge(), |
| DefaultTrafficSelector.builder().build(), |
| DefaultTrafficTreatment.builder() |
| .transition(ROUTING_TABLE).build(), |
| 0, |
| t, |
| install |
| ); |
| }); |
| }); |
| } |
| |
| private void setNamespaceRulesByPod(Pod pod, boolean install) { |
| String podIp = pod.getStatus().getPodIP(); |
| |
| if (podIp == null) { |
| return; |
| } |
| |
| Integer nsHash = namespaceHashByPodIp(k8sPodService, k8sNamespaceService, podIp); |
| |
| // in uninstall case, we will have null nsHash value |
| if (install && nsHash == null) { |
| return; |
| } |
| |
| setNamespaceRulesBase(podIp, nsHash, install); |
| } |
| |
| private void setNamespaceRulesByService(Service service, boolean install) { |
| String clusterIp = service.getSpec().getClusterIP(); |
| |
| if (clusterIp == null) { |
| return; |
| } |
| |
| // one the creating of new service, the cluster IP will be configured as None |
| // in this case, we will NOT install any namespace related rules |
| if (DEFAULT_SERVICE_IP_NONE.equalsIgnoreCase(clusterIp)) { |
| return; |
| } |
| |
| setNamespaceRulesBase(clusterIp, namespaceHashByServiceIp(k8sServiceService, |
| k8sNamespaceService, clusterIp), install); |
| } |
| |
| private void setNamespaceRulesBase(String ip, Integer nsHash, boolean install) { |
| |
| k8sNodeService.completeNodes().forEach(n -> { |
| TrafficSelector.Builder origBuilder = DefaultTrafficSelector.builder() |
| .matchEthType(TYPE_IPV4) |
| .matchIPSrc(IpPrefix.valueOf(IpAddress.valueOf(ip), HOST_PREFIX)); |
| TrafficSelector.Builder convBuilder = DefaultTrafficSelector.builder() |
| .matchEthType(TYPE_IPV4) |
| .matchIPSrc(IpPrefix.valueOf(IpAddress.valueOf( |
| shiftIpDomain(ip, SHIFTED_IP_PREFIX)), HOST_PREFIX)); |
| TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder(); |
| |
| if (install) { |
| tBuilder.writeMetadata(nsHash, DEFAULT_METADATA_MASK) |
| .transition(GROUPING_TABLE); |
| } |
| |
| k8sFlowRuleService.setRule( |
| appId, |
| n.intgBridge(), |
| origBuilder.build(), |
| tBuilder.build(), |
| PRIORITY_NAMESPACE_RULE, |
| NAMESPACE_TABLE, |
| install |
| ); |
| |
| k8sFlowRuleService.setRule( |
| appId, |
| n.intgBridge(), |
| convBuilder.build(), |
| tBuilder.build(), |
| PRIORITY_NAMESPACE_RULE, |
| NAMESPACE_TABLE, |
| install |
| ); |
| }); |
| } |
| |
| private class InternalServiceListener implements K8sServiceListener { |
| |
| private boolean isRelevantHelper() { |
| return Objects.equals(localNodeId, leadershipService.getLeader(appId.name())); |
| } |
| |
| @Override |
| public void event(K8sServiceEvent event) { |
| Service service = event.subject(); |
| switch (event.type()) { |
| case K8S_SERVICE_CREATED: |
| case K8S_SERVICE_UPDATED: |
| eventExecutor.execute(() -> processServiceCreation(service)); |
| break; |
| case K8S_SERVICE_REMOVED: |
| eventExecutor.execute(() -> processServiceRemoval(service)); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| private void processServiceCreation(Service service) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setNamespaceRulesByService(service, true); |
| } |
| |
| private void processServiceRemoval(Service service) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setNamespaceRulesByService(service, false); |
| } |
| } |
| |
| private class InternalPodListener implements K8sPodListener { |
| |
| private boolean isRelevantHelper() { |
| return Objects.equals(localNodeId, leadershipService.getLeader(appId.name())); |
| } |
| |
| @Override |
| public void event(K8sPodEvent event) { |
| Pod pod = event.subject(); |
| switch (event.type()) { |
| case K8S_POD_CREATED: |
| case K8S_POD_UPDATED: |
| eventExecutor.execute(() -> processPodCreation(pod)); |
| break; |
| case K8S_POD_REMOVED: |
| eventExecutor.execute(() -> processPodRemoval(pod)); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| private void processPodCreation(Pod pod) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setBlockRulesByPod(pod, true); |
| setAllowRulesByPod(pod, true); |
| setNamespaceRulesByPod(pod, true); |
| } |
| |
| private void processPodRemoval(Pod pod) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setBlockRulesByPod(pod, false); |
| setAllowRulesByPod(pod, false); |
| setNamespaceRulesByPod(pod, false); |
| } |
| } |
| |
| private class InternalNetworkPolicyListener implements K8sNetworkPolicyListener { |
| |
| private boolean isRelevantHelper() { |
| return Objects.equals(localNodeId, leadershipService.getLeader(appId.name())); |
| } |
| |
| @Override |
| public void event(K8sNetworkPolicyEvent event) { |
| NetworkPolicy policy = event.subject(); |
| switch (event.type()) { |
| case K8S_NETWORK_POLICY_CREATED: |
| case K8S_NETWORK_POLICY_UPDATED: |
| eventExecutor.execute(() -> processNetworkPolicyCreation(policy)); |
| break; |
| case K8S_NETWORK_POLICY_REMOVED: |
| eventExecutor.execute(() -> processNetworkPolicyRemoval(policy)); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| private void processNetworkPolicyCreation(NetworkPolicy policy) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setBlockRulesByPolicy(policy, true); |
| setAllowRulesByPolicy(policy, true); |
| } |
| |
| private void processNetworkPolicyRemoval(NetworkPolicy policy) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setBlockRulesByPolicy(policy, false); |
| setAllowRulesByPolicy(policy, false); |
| } |
| } |
| |
| private class InternalNamespaceListener implements K8sNamespaceListener { |
| |
| private boolean isRelevantHelper() { |
| return Objects.equals(localNodeId, leadershipService.getLeader(appId.name())); |
| } |
| |
| @Override |
| public void event(K8sNamespaceEvent event) { |
| Namespace ns = event.subject(); |
| switch (event.type()) { |
| case K8S_NAMESPACE_CREATED: |
| eventExecutor.execute(() -> processNamespaceCreation(ns)); |
| break; |
| case K8S_NAMESPACE_REMOVED: |
| eventExecutor.execute(() -> processNamespaceRemoval(ns)); |
| break; |
| default: |
| break; |
| } |
| } |
| |
| private void processNamespaceCreation(Namespace namespace) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setDefaultAllowNamespaceRules(namespace, true); |
| setDefaultAllowServiceRules(true); |
| } |
| |
| private void processNamespaceRemoval(Namespace namespace) { |
| if (!isRelevantHelper()) { |
| return; |
| } |
| |
| setDefaultAllowNamespaceRules(namespace, false); |
| } |
| } |
| } |