Merge "Enable IPv6 in network policy driver."
This commit is contained in:
Zuul
Gerrit Code Review
commit
af5e8af415
|
|
@ -138,9 +138,12 @@ class NetworkPolicyDriver(base.NetworkPolicyDriver):
|
|||
if worker_subnet_id:
|
||||
default_cidrs.append(utils.get_subnet_cidr(worker_subnet_id))
|
||||
for cidr in default_cidrs:
|
||||
ethertype = constants.IPv4
|
||||
if ipaddress.ip_network(cidr).version == constants.IP_VERSION_6:
|
||||
ethertype = constants.IPv6
|
||||
default_rule = {
|
||||
'security_group_rule': {
|
||||
'ethertype': 'IPv4',
|
||||
'ethertype': ethertype,
|
||||
'security_group_id': sg_id,
|
||||
'direction': 'ingress',
|
||||
'description': 'Kuryr-Kubernetes NetPolicy SG rule',
|
||||
|
|
@ -365,12 +368,15 @@ class NetworkPolicyDriver(base.NetworkPolicyDriver):
|
|||
crd_rules, sg_id, direction, port, pod_selector,
|
||||
policy_namespace)
|
||||
if allow_all:
|
||||
container_port = None
|
||||
for container_port, pods in matched_pods.items():
|
||||
sg_rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, container_port,
|
||||
protocol=port.get('protocol'),
|
||||
pods=pods)
|
||||
crd_rules.append(sg_rule)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
sg_rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, container_port,
|
||||
protocol=port.get('protocol'),
|
||||
ethertype=ethertype,
|
||||
pods=pods)
|
||||
crd_rules.append(sg_rule)
|
||||
if direction == 'egress':
|
||||
rules = self._create_svc_egress_sg_rule(
|
||||
sg_id, policy_namespace, port=container_port,
|
||||
|
|
@ -410,26 +416,29 @@ class NetworkPolicyDriver(base.NetworkPolicyDriver):
|
|||
sg_rule_body_list, pod_selector, policy_namespace,
|
||||
allow_all=True)
|
||||
else:
|
||||
sg_rule = (
|
||||
driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, port.get('port'),
|
||||
protocol=port.get('protocol')))
|
||||
sg_rule_body_list.append(sg_rule)
|
||||
if direction == 'egress':
|
||||
rule = self._create_svc_egress_sg_rule(
|
||||
sg_id, policy_namespace, port=port.get('port'),
|
||||
protocol=port.get('protocol'))
|
||||
sg_rule_body_list.extend(rule)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
sg_rule = (
|
||||
driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, port.get('port'),
|
||||
ethertype=ethertype,
|
||||
protocol=port.get('protocol')))
|
||||
sg_rule_body_list.append(sg_rule)
|
||||
if direction == 'egress':
|
||||
rule = self._create_svc_egress_sg_rule(
|
||||
sg_id, policy_namespace, port=port.get('port'),
|
||||
protocol=port.get('protocol'))
|
||||
sg_rule_body_list.extend(rule)
|
||||
|
||||
def _create_default_sg_rule(self, sg_id, direction, sg_rule_body_list):
|
||||
default_rule = {
|
||||
'security_group_rule': {
|
||||
'ethertype': 'IPv4',
|
||||
'security_group_id': sg_id,
|
||||
'direction': direction,
|
||||
'description': 'Kuryr-Kubernetes NetPolicy SG rule',
|
||||
}}
|
||||
sg_rule_body_list.append(default_rule)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
default_rule = {
|
||||
'security_group_rule': {
|
||||
'ethertype': ethertype,
|
||||
'security_group_id': sg_id,
|
||||
'direction': direction,
|
||||
'description': 'Kuryr-Kubernetes NetPolicy SG rule',
|
||||
}}
|
||||
sg_rule_body_list.append(default_rule)
|
||||
|
||||
def _parse_sg_rules(self, sg_rule_body_list, direction, policy, sg_id):
|
||||
"""Parse policy into security group rules.
|
||||
|
|
@ -478,9 +487,10 @@ class NetworkPolicyDriver(base.NetworkPolicyDriver):
|
|||
if rule_list[0] == {}:
|
||||
LOG.debug('Applying default all open policy from %s',
|
||||
policy['metadata']['selfLink'])
|
||||
rule = driver_utils.create_security_group_rule_body(sg_id,
|
||||
direction)
|
||||
sg_rule_body_list.append(rule)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, ethertype=ethertype)
|
||||
sg_rule_body_list.append(rule)
|
||||
|
||||
for rule_block in rule_list:
|
||||
LOG.debug('Parsing %(dir)s Rule %(rule)s', {'dir': direction,
|
||||
|
|
@ -546,15 +556,17 @@ class NetworkPolicyDriver(base.NetworkPolicyDriver):
|
|||
sg_id, policy_namespace, resource=resource)
|
||||
sg_rule_body_list.extend(rule)
|
||||
if allow_all:
|
||||
rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction,
|
||||
port_range_min=1,
|
||||
port_range_max=65535)
|
||||
if direction == 'egress':
|
||||
rule = self._create_svc_egress_sg_rule(
|
||||
sg_id, policy_namespace)
|
||||
sg_rule_body_list.extend(rule)
|
||||
sg_rule_body_list.append(rule)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction,
|
||||
port_range_min=1,
|
||||
port_range_max=65535,
|
||||
ethertype=ethertype)
|
||||
sg_rule_body_list.append(rule)
|
||||
if direction == 'egress':
|
||||
rule = self._create_svc_egress_sg_rule(
|
||||
sg_id, policy_namespace)
|
||||
sg_rule_body_list.extend(rule)
|
||||
else:
|
||||
LOG.debug('This network policy specifies no %(direction)s '
|
||||
'%(rule_direction)s and no ports: %(policy)s',
|
||||
|
|
|
|||
|
|
@ -12,6 +12,8 @@
|
|||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from oslo_config import cfg
|
||||
from oslo_log import log as logging
|
||||
|
||||
from kuryr_kubernetes import clients
|
||||
from kuryr_kubernetes import config
|
||||
|
|
@ -20,9 +22,6 @@ from kuryr_kubernetes.controller.drivers import base
|
|||
from kuryr_kubernetes.controller.drivers import utils as driver_utils
|
||||
from kuryr_kubernetes import exceptions
|
||||
|
||||
from oslo_config import cfg
|
||||
from oslo_log import log as logging
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
|
|
@ -172,12 +171,27 @@ def _create_sg_rule_on_text_port(sg_id, direction, port, rule_selected_pods,
|
|||
matched_pods, container_ports, allow_all,
|
||||
namespace, matched, crd_rules, sg_id, direction,
|
||||
port, rule_selected_pod)
|
||||
|
||||
_apply_sg_rules_on_matched_pods(matched_pods, sg_id, direction, namespace,
|
||||
port, crd_rules, allow_all)
|
||||
|
||||
return matched
|
||||
|
||||
|
||||
def _apply_sg_rules_on_matched_pods(matched_pods, sg_id, direction, namespace,
|
||||
port, crd_rules, allow_all=False):
|
||||
for container_port, pods in matched_pods.items():
|
||||
if allow_all:
|
||||
sg_rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, container_port,
|
||||
protocol=port.get('protocol'),
|
||||
pods=pods)
|
||||
for ethertype in (constants.IPv4, constants.IPv6):
|
||||
sg_rule = driver_utils.create_security_group_rule_body(
|
||||
sg_id, direction, container_port,
|
||||
protocol=port.get('protocol'),
|
||||
ethertype=ethertype,
|
||||
pods=pods)
|
||||
sgr_id = driver_utils.create_security_group_rule(sg_rule)
|
||||
sg_rule['security_group_rule']['id'] = sgr_id
|
||||
if sg_rule not in crd_rules:
|
||||
crd_rules.append(sg_rule)
|
||||
else:
|
||||
namespace_obj = driver_utils.get_namespace(namespace)
|
||||
if not namespace_obj:
|
||||
|
|
@ -190,11 +204,10 @@ def _create_sg_rule_on_text_port(sg_id, direction, port, rule_selected_pods,
|
|||
sg_id, direction, container_port,
|
||||
protocol=port.get('protocol'), cidr=namespace_cidr,
|
||||
pods=pods)
|
||||
sgr_id = driver_utils.create_security_group_rule(sg_rule)
|
||||
sg_rule['security_group_rule']['id'] = sgr_id
|
||||
if sg_rule not in crd_rules:
|
||||
crd_rules.append(sg_rule)
|
||||
return matched
|
||||
sgr_id = driver_utils.create_security_group_rule(sg_rule)
|
||||
sg_rule['security_group_rule']['id'] = sgr_id
|
||||
if sg_rule not in crd_rules:
|
||||
crd_rules.append(sg_rule)
|
||||
|
||||
|
||||
def _create_sg_rules(crd, pod, pod_selector, rule_block,
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
|
||||
import urllib
|
||||
|
||||
import netaddr
|
||||
from openstack import exceptions as os_exc
|
||||
from oslo_config import cfg
|
||||
from oslo_log import log
|
||||
|
|
@ -233,7 +234,7 @@ def patch_kuryrnetworkpolicy_crd(crd, i_rules, e_rules, pod_selector,
|
|||
|
||||
def create_security_group_rule_body(
|
||||
security_group_id, direction, port_range_min=None,
|
||||
port_range_max=None, protocol=None, ethertype='IPv4', cidr=None,
|
||||
port_range_max=None, protocol=None, ethertype=None, cidr=None,
|
||||
description="Kuryr-Kubernetes NetPolicy SG rule", namespace=None,
|
||||
pods=None):
|
||||
if not port_range_min:
|
||||
|
|
@ -243,6 +244,12 @@ def create_security_group_rule_body(
|
|||
port_range_max = port_range_min
|
||||
if not protocol:
|
||||
protocol = 'TCP'
|
||||
|
||||
if not ethertype:
|
||||
ethertype = 'IPv4'
|
||||
if cidr and netaddr.IPNetwork(cidr).version == 6:
|
||||
ethertype = 'IPv6'
|
||||
|
||||
security_group_rule_body = {
|
||||
'security_group_rule': {
|
||||
'ethertype': ethertype,
|
||||
|
|
|
|||
|
|
@ -240,7 +240,7 @@ class TestNetworkPolicyDriver(test_base.TestCase):
|
|||
self._driver.os_net.create_security_group.return_value = (
|
||||
munch.Munch({'id': mock.sentinel.id,
|
||||
'security_group_rules': []}))
|
||||
m_utils.get_subnet_cidr.return_value = {'cidr': mock.sentinel.cidr}
|
||||
m_utils.get_subnet_cidr.return_value = mock.sentinel.cidr
|
||||
m_parse.return_value = (self._i_rules, self._e_rules)
|
||||
self._driver.os_net.create_security_group_rule.return_value = (
|
||||
munch.Munch({'id': mock.sentinel.id}))
|
||||
|
|
@ -265,7 +265,7 @@ class TestNetworkPolicyDriver(test_base.TestCase):
|
|||
self._driver.os_net.create_security_group.return_value = (
|
||||
munch.Munch({'id': mock.sentinel.id,
|
||||
'security_group_rules': []}))
|
||||
m_utils.get_subnet_cidr.return_value = {'cidr': mock.sentinel.cidr}
|
||||
m_utils.get_subnet_cidr.return_value = mock.sentinel.cidr
|
||||
m_parse.return_value = (self._i_rules, self._e_rules)
|
||||
m_get_crd.side_effect = exceptions.K8sClientException
|
||||
self._driver.os_net.create_security_group_rule.return_value = (
|
||||
|
|
@ -292,7 +292,7 @@ class TestNetworkPolicyDriver(test_base.TestCase):
|
|||
self._driver.os_net.create_security_group.return_value = (
|
||||
munch.Munch({'id': mock.sentinel.id,
|
||||
'security_group_rules': []}))
|
||||
m_utils.get_subnet_cidr.return_value = {'cidr': mock.sentinel.cidr}
|
||||
m_utils.get_subnet_cidr.return_value = mock.sentinel.cidr
|
||||
m_parse.return_value = (self._i_rules, self._e_rules)
|
||||
m_add_crd.side_effect = exceptions.K8sClientException
|
||||
self._driver.os_net.create_security_group_rule.return_value = (
|
||||
|
|
@ -393,8 +393,10 @@ class TestNetworkPolicyDriver(test_base.TestCase):
|
|||
policy['spec']['egress'] = [{}]
|
||||
self._driver.parse_network_policy_rules(policy, self._sg_id)
|
||||
m_get_ns.assert_not_called()
|
||||
calls = [mock.call(self._sg_id, 'ingress'),
|
||||
mock.call(self._sg_id, 'egress')]
|
||||
calls = [mock.call(self._sg_id, 'ingress', ethertype='IPv4'),
|
||||
mock.call(self._sg_id, 'ingress', ethertype='IPv6'),
|
||||
mock.call(self._sg_id, 'egress', ethertype='IPv4'),
|
||||
mock.call(self._sg_id, 'egress', ethertype='IPv6')]
|
||||
m_create.assert_has_calls(calls)
|
||||
|
||||
@mock.patch.object(network_policy.NetworkPolicyDriver,
|
||||
|
|
@ -520,3 +522,248 @@ class TestNetworkPolicyDriver(test_base.TestCase):
|
|||
def test_release_network_policy_removed_crd(self, m_del_crd):
|
||||
self._driver.release_network_policy(None)
|
||||
m_del_crd.assert_not_called()
|
||||
|
||||
@mock.patch.object(network_policy.NetworkPolicyDriver,
|
||||
'_create_sg_rules_with_container_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_pods')
|
||||
def test__create_sg_rule_body_on_text_port_ingress(self,
|
||||
m_get_pods,
|
||||
m_get_ports,
|
||||
m_create_sgr_cont):
|
||||
pod = mock.sentinel.pod
|
||||
port = mock.sentinel.port
|
||||
container_ports = mock.sentinel.ports
|
||||
resources = [mock.sentinel.resource]
|
||||
crd_rules = mock.sentinel.crd_rules
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'ingress'
|
||||
|
||||
m_get_pods.return_value = {'items': [pod]}
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace)
|
||||
|
||||
m_get_pods.assert_called_with(pod_selector, namespace)
|
||||
m_get_ports.assert_called_with(pod, port)
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
@mock.patch.object(network_policy.NetworkPolicyDriver,
|
||||
'_create_sg_rules_with_container_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_pods')
|
||||
def test__create_sg_rule_body_on_text_port_ingress_all(self,
|
||||
m_get_pods,
|
||||
m_get_ports,
|
||||
m_create_sgr_cont,
|
||||
m_create_sgr):
|
||||
pod = mock.sentinel.pod
|
||||
port = mock.sentinel.port
|
||||
container_ports = mock.sentinel.ports
|
||||
resources = [mock.sentinel.resource]
|
||||
crd_rules = mock.sentinel.crd_rules
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'ingress'
|
||||
|
||||
m_get_pods.return_value = {'items': [pod]}
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace,
|
||||
allow_all=True)
|
||||
|
||||
m_get_pods.assert_called_with(pod_selector, namespace)
|
||||
m_get_ports.assert_called_with(pod, port)
|
||||
m_create_sgr.assert_not_called()
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_pods')
|
||||
def test__create_sg_rule_body_on_text_port_ingress_match(self,
|
||||
m_get_pods,
|
||||
m_get_ports,
|
||||
m_create_sgr):
|
||||
|
||||
def _create_sgr_cont(container_ports, allow_all, resource,
|
||||
matched_pods, crd_rules, sg_id, direction, port,
|
||||
pod_selector=None, policy_namespace=None):
|
||||
matched_pods[container_ports[0][1]] = 'foo'
|
||||
|
||||
pod = mock.sentinel.pod
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
container_ports = [("pod", mock.sentinel.container_port)]
|
||||
resources = [mock.sentinel.resource]
|
||||
crd_rules = []
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'ingress'
|
||||
self._driver._create_sg_rules_with_container_ports = _create_sgr_cont
|
||||
|
||||
m_get_pods.return_value = {'items': [pod]}
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace,
|
||||
allow_all=True)
|
||||
|
||||
m_get_pods.assert_called_with(pod_selector, namespace)
|
||||
m_get_ports.assert_called_with(pod, port)
|
||||
|
||||
calls = [mock.call(self._sg_id, direction, container_ports[0][1],
|
||||
protocol=port['protocol'], ethertype=e,
|
||||
pods='foo') for e in ('IPv4', 'IPv6')]
|
||||
|
||||
m_create_sgr.assert_has_calls(calls)
|
||||
self.assertEqual(len(crd_rules), 2)
|
||||
|
||||
@mock.patch.object(network_policy.NetworkPolicyDriver,
|
||||
'_create_sg_rules_with_container_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_pods')
|
||||
def test__create_sg_rule_body_on_text_port_egress(self,
|
||||
m_get_pods,
|
||||
m_get_ports,
|
||||
m_create_sgr_cont):
|
||||
pod = mock.sentinel.pod
|
||||
port = mock.sentinel.port
|
||||
container_ports = mock.sentinel.ports
|
||||
resources = [{'spec': 'foo'}]
|
||||
crd_rules = mock.sentinel.crd_rules
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'egress'
|
||||
|
||||
m_get_pods.return_value = {'items': [pod]}
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace)
|
||||
|
||||
m_get_ports.assert_called_with(resources[0], port)
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
@mock.patch.object(network_policy.NetworkPolicyDriver,
|
||||
'_create_sg_rules_with_container_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
def test__create_sg_rule_body_on_text_port_egress_all(self,
|
||||
m_get_ports,
|
||||
m_create_sgr_cont,
|
||||
m_create_sgr):
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
container_ports = mock.sentinel.ports
|
||||
resources = [{'spec': 'foo'}]
|
||||
crd_rules = []
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'egress'
|
||||
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace,
|
||||
allow_all=True)
|
||||
|
||||
m_get_ports.assert_called_with(resources[0], port)
|
||||
m_create_sgr.assert_called_once_with(self._sg_id, 'egress', None,
|
||||
cidr=mock.ANY, protocol='TCP')
|
||||
self.assertEqual(len(crd_rules), 1)
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_ports')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.get_pods')
|
||||
def test__create_sg_rule_body_on_text_port_egress_match(self,
|
||||
m_get_pods,
|
||||
m_get_ports,
|
||||
m_create_sgr):
|
||||
|
||||
def _create_sgr_cont(container_ports, allow_all, resource,
|
||||
matched_pods, crd_rules, sg_id, direction, port,
|
||||
pod_selector=None, policy_namespace=None):
|
||||
matched_pods[container_ports[0][1]] = 'foo'
|
||||
|
||||
pod = mock.sentinel.pod
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
container_ports = [("pod", mock.sentinel.container_port)]
|
||||
resources = [{'spec': 'foo'}]
|
||||
crd_rules = []
|
||||
pod_selector = {}
|
||||
namespace = mock.sentinel.namespace
|
||||
direction = 'egress'
|
||||
self._driver._create_sg_rules_with_container_ports = _create_sgr_cont
|
||||
|
||||
m_get_pods.return_value = {'items': [pod]}
|
||||
m_get_ports.return_value = container_ports
|
||||
|
||||
self._driver._create_sg_rule_body_on_text_port(self._sg_id,
|
||||
direction,
|
||||
port,
|
||||
resources,
|
||||
crd_rules,
|
||||
pod_selector,
|
||||
namespace,
|
||||
allow_all=True)
|
||||
|
||||
m_get_ports.assert_called_with(resources[0], port)
|
||||
|
||||
calls = [mock.call(self._sg_id, direction, container_ports[0][1],
|
||||
protocol=port['protocol'], ethertype=e,
|
||||
pods='foo') for e in ('IPv4', 'IPv6')]
|
||||
|
||||
m_create_sgr.assert_has_calls(calls)
|
||||
# NOTE(gryf): there are 3 rules created in case of egress direction,
|
||||
# since additional one is created for specific cidr in service subnet.
|
||||
self.assertEqual(len(crd_rules), 3)
|
||||
|
||||
def test__create_all_pods_sg_rules(self):
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
direction = 'ingress'
|
||||
rules = []
|
||||
|
||||
self._driver._create_all_pods_sg_rules(port, self._sg_id, direction,
|
||||
rules, '', None)
|
||||
self.assertEqual(len(rules), 2)
|
||||
|
||||
def test__create_default_sg_rule(self):
|
||||
for direction in ('ingress', 'egress'):
|
||||
rules = []
|
||||
|
||||
self._driver._create_default_sg_rule(self._sg_id, direction, rules)
|
||||
self.assertEqual(len(rules), 2)
|
||||
self.assertListEqual(rules, [{'security_group_rule': {
|
||||
'ethertype': e,
|
||||
'security_group_id': self._sg_id,
|
||||
'direction': direction,
|
||||
'description': 'Kuryr-Kubernetes NetPolicy SG rule'
|
||||
}} for e in ('IPv4', 'IPv6')])
|
||||
|
|
|
|||
|
|
@ -700,3 +700,94 @@ class TestNetworkPolicySecurityGroupsDriver(test_base.TestCase):
|
|||
|
||||
self.assertEqual(matched, matched_selector)
|
||||
self.assertEqual(rules, final_crd_rules)
|
||||
|
||||
|
||||
class TestNetworkPolicySecurityGroupsFunctions(test_base.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.kubernetes = self.useFixture(k_fix.MockK8sClient()).client
|
||||
self.npsg = network_policy_security_groups
|
||||
self.sg_id = mock.sentinel.sg_id
|
||||
|
||||
self.crd = {
|
||||
'spec': {
|
||||
'ingressSgRules': [],
|
||||
'networkpolicy_spec': {
|
||||
'ingress': [],
|
||||
'policyTypes': ['Ingress']
|
||||
}
|
||||
},
|
||||
'metadata': {'namespace': 'ns'}
|
||||
}
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
def test__apply_sg_rules_on_matched_pods_empty_match(self, m_create_sgrb,
|
||||
m_create_sgr):
|
||||
self.npsg._apply_sg_rules_on_matched_pods({}, self.sg_id, 'ingress',
|
||||
'ns', 'port', 'crd_rules')
|
||||
|
||||
m_create_sgrb.assert_not_called()
|
||||
m_create_sgr.assert_not_called()
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'get_namespace_subnet_cidr')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'get_namespace')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule_body')
|
||||
def test__apply_sg_rules_on_matched_pods_not_all(self, m_create_sgrb,
|
||||
m_create_sgr, m_get_ns,
|
||||
m_get_ns_sub_cidr):
|
||||
pod = mock.sentinel.pod
|
||||
ns = mock.sentinel.ns
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
matched_pods = {'container_port': [pod]}
|
||||
|
||||
m_get_ns.return_value = ns
|
||||
m_create_sgrb.return_value = {'security_group_rule': {}}
|
||||
crd_rules = []
|
||||
direction = 'ingress'
|
||||
|
||||
self.npsg._apply_sg_rules_on_matched_pods(matched_pods, self.sg_id,
|
||||
direction, 'ns', port,
|
||||
crd_rules)
|
||||
|
||||
m_get_ns_sub_cidr.assert_called_once_with(ns)
|
||||
m_create_sgrb.assert_called_once_with(self.sg_id, direction,
|
||||
'container_port',
|
||||
protocol=mock.ANY, cidr=mock.ANY,
|
||||
pods=[pod])
|
||||
m_create_sgr.assert_called_once()
|
||||
self.assertEqual(len(crd_rules), 1)
|
||||
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'get_namespace_subnet_cidr')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'get_namespace')
|
||||
@mock.patch('kuryr_kubernetes.controller.drivers.utils.'
|
||||
'create_security_group_rule')
|
||||
def test__apply_sg_rules_on_matched_pods_all(self, m_create_sgr, m_get_ns,
|
||||
m_get_ns_sub_cidr):
|
||||
pod = mock.sentinel.pod
|
||||
ns = mock.sentinel.ns
|
||||
port = {'protocol': 'TCP', 'port': 22}
|
||||
matched_pods = {'container_port': [pod]}
|
||||
|
||||
m_get_ns.return_value = ns
|
||||
crd_rules = []
|
||||
direction = 'ingress'
|
||||
|
||||
self.npsg._apply_sg_rules_on_matched_pods(matched_pods, self.sg_id,
|
||||
direction, 'ns', port,
|
||||
crd_rules, allow_all=True)
|
||||
|
||||
self.assertEqual(m_create_sgr.call_count, 2)
|
||||
self.assertEqual(len(crd_rules), 2)
|
||||
self.assertListEqual([r['security_group_rule']['ethertype']
|
||||
for r in crd_rules], ['IPv4', 'IPv6'])
|
||||
|
|
|
|||
Loading…
Reference in New Issue