openstack
/
neutron
Code Issues Proposed changes

Permit ICMPv6 RAs only from known routers 

Currently ingress ICMPv6 RAs are permitted from any IPs by
default to allow VMs to accept ICMPv6 RA from provider network.
In this way, VM can accept RAs from attacker VM and configure
a network prefix specified by the attacher VM.

Remove permitting ICMPv6 RAs from any IPs and add security rule
to only permit ICMPv6 RA from:

1. If the port's subnet is configured with ipv6_ra_mode value
(i.e.value is slaac, dhcpv6-stateful, or dhcpv6-stateless), RA
is sending from dnsmasq controlled by OpenStack. In this case,
allow RA from the link local address of gateway port (if the
gateway port is created).

2. If the subnet's gateway port is not managed by OpenStack, allow
the ICMPv6 RA sent from the subnet gateway IP if it's a link local
address. The administrator needs to configure the gateway IP as
link local address in this case to make the RA rule work.

Change-Id: I1d5c7aaa8e4cf057204eb746c0faab2c70409a94
Closes-Bug: 1262759
This commit is contained in:
Xuhan Peng 2014-02-09 22:02:33 -05:00
parent efb2db91a7
commit b7b0c7dbcd
5 changed files with 389 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
neutron/agent/linux/iptables_firewall.py
View File

@ -241,8 +241,8 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
return ['-p udp -m udp --sport 67 --dport 68 -j DROP']
def _accept_inbound_icmpv6(self):
# Allow router advertisements, multicast listener
# and neighbor advertisement into the instance
# Allow multicast listener, neighbor solicitation and
# neighbor advertisement into the instance
icmpv6_rules = []
for icmp6_type in constants.ICMPV6_ALLOWED_TYPES:
icmpv6_rules += ['-p icmpv6 --icmpv6-type %s -j RETURN' %

6
neutron/common/constants.py
View File

@ -106,12 +106,14 @@ PROTO_NUM_UDP = 17
# Multicast Listener Query (130),
# Multicast Listener Report (131),
# Multicast Listener Done (132),
# Router Advertisement (134),
# Neighbor Solicitation (135),
# Neighbor Advertisement (136)
ICMPV6_ALLOWED_TYPES = [130, 131, 132, 134, 135, 136]
ICMPV6_ALLOWED_TYPES = [130, 131, 132, 135, 136]
ICMPV6_TYPE_RA = 134
DHCPV6_STATEFUL = 'dhcpv6-stateful'
DHCPV6_STATELESS = 'dhcpv6-stateless'
IPV6_SLAAC = 'slaac'
IPV6_MODES = [DHCPV6_STATEFUL, DHCPV6_STATELESS, IPV6_SLAAC]
IPV6_LLA_PREFIX = 'fe80::/64'

76
neutron/db/securitygroups_rpc_base.py
View File

@ -16,8 +16,10 @@
# under the License.
import netaddr
from sqlalchemy.orm import exc
from neutron.common import constants as q_const
from neutron.common import ipv6_utils as ipv6
from neutron.common import utils
from neutron.db import models_v2
from neutron.db import securitygroups_db as sg_db
@ -222,6 +224,67 @@ class SecurityGroupServerRpcCallbackMixin(object):
ips[port['network_id']].append(ip)
return ips
def _select_ra_ips_for_network_ids(self, context, network_ids):
"""Select IP addresses to allow sending router advertisement from.
If OpenStack dnsmasq sends RA, get link local address of
gateway and allow RA from this Link Local address.
The gateway port link local address will only be obtained
when router is created before VM instance is booted and
subnet is attached to router.
If OpenStack doesn't send RA, allow RA from gateway IP.
Currently, the gateway IP needs to be link local to be able
to send RA to VM.
"""
if not network_ids:
return {}
ips = {}
for network_id in network_ids:
ips[network_id] = set([])
query = context.session.query(models_v2.Subnet)
subnets = query.filter(models_v2.Subnet.network_id.in_(network_ids))
for subnet in subnets:
gateway_ip = subnet['gateway_ip']
if subnet['ip_version'] != 6 or not gateway_ip:
continue
# TODO(xuhanp): Figure out how to call the following code
# each time router is created or updated.
if not netaddr.IPAddress(gateway_ip).is_link_local():
if subnet['ipv6_ra_mode']:
gateway_ip = self._get_lla_gateway_ip_for_subnet(context,
subnet)
else:
# TODO(xuhanp):Figure out how to allow gateway IP from
# existing device to be global address and figure out the
# link local address by other method.
continue
if gateway_ip:
ips[subnet['network_id']].add(gateway_ip)
return ips
def _get_lla_gateway_ip_for_subnet(self, context, subnet):
query = context.session.query(models_v2.Port)
query = query.join(models_v2.IPAllocation)
query = query.filter(
models_v2.IPAllocation.subnet_id == subnet['id'])
query = query.filter(
models_v2.IPAllocation.ip_address == subnet['gateway_ip'])
query = query.filter(models_v2.Port.device_owner ==
q_const.DEVICE_OWNER_ROUTER_INTF)
try:
gateway_port = query.one()
except (exc.NoResultFound, exc.MultipleResultsFound):
LOG.warn(_('No valid gateway port on subnet %s is '
'found for IPv6 RA'), subnet['id'])
return
mac_address = gateway_port['mac_address']
lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
q_const.IPV6_LLA_PREFIX,
mac_address))
return lla_ip
def _convert_remote_group_id_to_ip_prefix(self, context, ports):
remote_group_ids = self._select_remote_group_ids(ports)
ips = self._select_ips_for_remote_group(context, remote_group_ids)
@ -276,17 +339,18 @@ class SecurityGroupServerRpcCallbackMixin(object):
ra_rule = {'direction': 'ingress',
'ethertype': q_const.IPv6,
'protocol': 'icmp'}
ra_rule['source_ip_prefix'] = "%s/%s" % (ra_ip,
IP_MASK[q_const.IPv6])
'protocol': 'icmp',
'source_ip_prefix': ra_ip,
'source_port_range_min': q_const.ICMPV6_TYPE_RA}
port['security_group_rules'].append(ra_rule)
def _apply_provider_rule(self, context, ports):
network_ids = self._select_network_ids(ports)
ips = self._select_dhcp_ips_for_network_ids(context, network_ids)
ips_dhcp = self._select_dhcp_ips_for_network_ids(context, network_ids)
ips_ra = self._select_ra_ips_for_network_ids(context, network_ids)
for port in ports.values():
self._add_ingress_ra_rule(port, ips)
self._add_ingress_dhcp_rule(port, ips)
self._add_ingress_ra_rule(port, ips_ra)
self._add_ingress_dhcp_rule(port, ips_dhcp)
def _security_group_rules_for_ports(self, context, ports):
rules_in_db = self._select_rules_for_ports(context, ports)

24
neutron/tests/unit/oneconvergence/test_security_group.py
View File

@ -75,6 +75,18 @@ class TestOneConvergenceSGServerRpcCallBack(
def test_security_group_rules_for_devices_ipv6_source_group(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_gateway_global(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_gateway_lla(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_no_gateway_port(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_rule_for_device_ipv6_multi_router_interfaces(self):
self.skipTest("NVSD Plugin does not support IPV6.")
class TestOneConvergenceSGServerRpcCallBackXML(
OneConvergenceSecurityGroupsTestCase,
@ -88,6 +100,18 @@ class TestOneConvergenceSGServerRpcCallBackXML(
def test_security_group_rules_for_devices_ipv6_source_group(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_gateway_global(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_gateway_lla(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_ra_rules_for_devices_ipv6_no_gateway_port(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_security_group_rule_for_device_ipv6_multi_router_interfaces(self):
self.skipTest("NVSD Plugin does not support IPV6.")
class TestOneConvergenceSecurityGroups(OneConvergenceSecurityGroupsTestCase,
test_sg.TestSecurityGroups,

313
neutron/tests/unit/test_security_groups_rpc.py
View File

@ -30,6 +30,7 @@ from neutron.agent.linux import iptables_manager
from neutron.agent import rpc as agent_rpc
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.common import constants as const
from neutron.common import ipv6_utils as ipv6
from neutron import context
from neutron.db import securitygroups_rpc_base as sg_db_rpc
from neutron.extensions import allowedaddresspairs as addr_pair
@ -38,7 +39,14 @@ from neutron.manager import NeutronManager
from neutron.openstack.common.rpc import proxy
from neutron.tests import base
from neutron.tests.unit import test_extension_security_group as test_sg
from neutron.tests.unit import test_iptables_firewall as test_fw
FAKE_PREFIX = {const.IPv4: '10.0.0.0/24',
const.IPv6: '2001:0db8::/64'}
FAKE_IP = {const.IPv4: '10.0.0.1',
const.IPv6: 'fe80::1',
'IPv6_GLOBAL': '2001:0db8::1',
'IPv6_LLA': 'fe80::123'}
class FakeSGCallback(sg_db_rpc.SecurityGroupServerRpcCallbackMixin):
@ -61,7 +69,7 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
self.rpc = FakeSGCallback()
def test_security_group_rules_for_devices_ipv4_ingress(self):
fake_prefix = test_fw.FAKE_PREFIX[const.IPv4]
fake_prefix = FAKE_PREFIX[const.IPv4]
with self.network() as n:
with nested(self.subnet(n),
self.security_group()) as (subnet_v4,
@ -119,7 +127,7 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
if ('allowed-address-pairs'
not in plugin_obj.supported_extension_aliases):
self.skipTest("Test depeneds on allowed-address-pairs extension")
fake_prefix = test_fw.FAKE_PREFIX['IPv4']
fake_prefix = FAKE_PREFIX['IPv4']
with self.network() as n:
with nested(self.subnet(n),
self.security_group()) as (subnet_v4,
@ -178,7 +186,7 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
self._delete('ports', port_id1)
def test_security_group_rules_for_devices_ipv4_egress(self):
fake_prefix = test_fw.FAKE_PREFIX[const.IPv4]
fake_prefix = FAKE_PREFIX[const.IPv4]
with self.network() as n:
with nested(self.subnet(n),
self.security_group()) as (subnet_v4,
@ -291,9 +299,11 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
self._delete('ports', port_id2)
def test_security_group_rules_for_devices_ipv6_ingress(self):
fake_prefix = test_fw.FAKE_PREFIX[const.IPv6]
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6),
self.security_group()) as (subnet_v6,
@ -341,18 +351,271 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
{'direction': 'ingress',
'protocol': const.PROTO_NAME_UDP,
'ethertype': const.IPv6,
'port_range_max': 23, 'security_group_id': sg1_id,
'port_range_max': 23,
'security_group_id': sg1_id,
'port_range_min': 23,
'source_ip_prefix': fake_prefix},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': fake_gateway,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
def test_security_group_ra_rules_for_devices_ipv6_gateway_global(self):
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP['IPv6_GLOBAL']
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6,
ipv6_ra_mode=const.IPV6_SLAAC),
self.security_group()) as (subnet_v6,
sg1):
sg1_id = sg1['security_group']['id']
rule1 = self._build_security_group_rule(
sg1_id,
'ingress', const.PROTO_NAME_TCP, '22',
'22',
ethertype=const.IPv6)
rules = {
'security_group_rules': [rule1['security_group_rule']]}
self._make_security_group_rule(self.fmt, rules)
# Create gateway port
gateway_res = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
device_owner='network:router_interface')
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
const.IPV6_LLA_PREFIX,
gateway_mac))
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id])
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
ctx = context.get_admin_context()
ports_rpc = self.rpc.security_group_rules_for_devices(
ctx, devices=devices)
port_rpc = ports_rpc[port_id1]
expected = [{'direction': 'egress', 'ethertype': const.IPv4,
'security_group_id': sg1_id},
{'direction': 'egress', 'ethertype': const.IPv6,
'security_group_id': sg1_id},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_TCP,
'ethertype': const.IPv6,
'port_range_max': 22,
'security_group_id': sg1_id,
'port_range_min': 22},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': gateway_lla_ip,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
# Note(xuhanp): remove gateway port's fixed_ips or gateway port
# deletion will be prevented.
data = {'port': {'fixed_ips': []}}
req = self.new_update_request('ports', data, gateway_port_id)
self.deserialize(self.fmt, req.get_response(self.api))
self._delete('ports', gateway_port_id)
def test_security_group_rule_for_device_ipv6_multi_router_interfaces(self):
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP['IPv6_GLOBAL']
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6,
ipv6_ra_mode=const.IPV6_SLAAC),
self.security_group()) as (subnet_v6,
sg1):
sg1_id = sg1['security_group']['id']
rule1 = self._build_security_group_rule(
sg1_id,
'ingress', const.PROTO_NAME_TCP, '22',
'22',
ethertype=const.IPv6)
rules = {
'security_group_rules': [rule1['security_group_rule']]}
self._make_security_group_rule(self.fmt, rules)
# Create gateway port
gateway_res = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
device_owner='network:router_interface')
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
const.IPV6_LLA_PREFIX,
gateway_mac))
# Create another router interface port
interface_res = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
device_owner='network:router_interface')
interface_port_id = interface_res['port']['id']
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id])
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
ctx = context.get_admin_context()
ports_rpc = self.rpc.security_group_rules_for_devices(
ctx, devices=devices)
port_rpc = ports_rpc[port_id1]
expected = [{'direction': 'egress', 'ethertype': const.IPv4,
'security_group_id': sg1_id},
{'direction': 'egress', 'ethertype': const.IPv6,
'security_group_id': sg1_id},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_TCP,
'ethertype': const.IPv6,
'port_range_max': 22,
'security_group_id': sg1_id,
'port_range_min': 22},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': gateway_lla_ip,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
data = {'port': {'fixed_ips': []}}
req = self.new_update_request('ports', data, gateway_port_id)
self.deserialize(self.fmt, req.get_response(self.api))
req = self.new_update_request('ports', data, interface_port_id)
self.deserialize(self.fmt, req.get_response(self.api))
self._delete('ports', gateway_port_id)
self._delete('ports', interface_port_id)
def test_security_group_ra_rules_for_devices_ipv6_gateway_lla(self):
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP['IPv6_LLA']
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6,
ipv6_ra_mode=const.IPV6_SLAAC),
self.security_group()) as (subnet_v6,
sg1):
sg1_id = sg1['security_group']['id']
rule1 = self._build_security_group_rule(
sg1_id,
'ingress', const.PROTO_NAME_TCP, '22',
'22',
ethertype=const.IPv6)
rules = {
'security_group_rules': [rule1['security_group_rule']]}
self._make_security_group_rule(self.fmt, rules)
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id])
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
ctx = context.get_admin_context()
ports_rpc = self.rpc.security_group_rules_for_devices(
ctx, devices=devices)
port_rpc = ports_rpc[port_id1]
expected = [{'direction': 'egress', 'ethertype': const.IPv4,
'security_group_id': sg1_id},
{'direction': 'egress', 'ethertype': const.IPv6,
'security_group_id': sg1_id},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_TCP,
'ethertype': const.IPv6,
'port_range_max': 22,
'security_group_id': sg1_id,
'port_range_min': 22},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': fake_gateway,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
def test_security_group_ra_rules_for_devices_ipv6_no_gateway_port(self):
fake_prefix = FAKE_PREFIX[const.IPv6]
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=None,
cidr=fake_prefix,
ip_version=6,
ipv6_ra_mode=const.IPV6_SLAAC),
self.security_group()) as (subnet_v6,
sg1):
sg1_id = sg1['security_group']['id']
rule1 = self._build_security_group_rule(
sg1_id,
'ingress', const.PROTO_NAME_TCP, '22',
'22',
ethertype=const.IPv6)
rules = {
'security_group_rules': [rule1['security_group_rule']]}
self._make_security_group_rule(self.fmt, rules)
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id])
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
ctx = context.get_admin_context()
ports_rpc = self.rpc.security_group_rules_for_devices(
ctx, devices=devices)
port_rpc = ports_rpc[port_id1]
expected = [{'direction': 'egress', 'ethertype': const.IPv4,
'security_group_id': sg1_id},
{'direction': 'egress', 'ethertype': const.IPv6,
'security_group_id': sg1_id},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_TCP,
'ethertype': const.IPv6,
'port_range_max': 22,
'security_group_id': sg1_id,
'port_range_min': 22},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
def test_security_group_rules_for_devices_ipv6_egress(self):
fake_prefix = test_fw.FAKE_PREFIX[const.IPv6]
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6),
self.security_group()) as (subnet_v6,
@ -371,15 +634,12 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
rules = {
'security_group_rules': [rule1['security_group_rule'],
rule2['security_group_rule']]}
res = self._create_security_group_rule(self.fmt, rules)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, webob.exc.HTTPCreated.code)
self._make_security_group_rule(self.fmt, rules)
res1 = self._create_port(
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id])
ports_rest1 = self.deserialize(self.fmt, res1)
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
@ -405,15 +665,22 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
'security_group_id': sg1_id,
'port_range_min': 23,
'dest_ip_prefix': fake_prefix},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': fake_gateway,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
self._delete('ports', port_id1)
def test_security_group_rules_for_devices_ipv6_source_group(self):
fake_prefix = test_fw.FAKE_PREFIX[const.IPv6]
fake_prefix = FAKE_PREFIX[const.IPv6]
fake_gateway = FAKE_IP[const.IPv6]
with self.network() as n:
with nested(self.subnet(n,
gateway_ip=fake_gateway,
cidr=fake_prefix,
ip_version=6),
self.security_group(),
@ -430,25 +697,21 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
remote_group_id=sg2['security_group']['id'])
rules = {
'security_group_rules': [rule1['security_group_rule']]}
res = self._create_security_group_rule(self.fmt, rules)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, webob.exc.HTTPCreated.code)
self._make_security_group_rule(self.fmt, rules)
res1 = self._create_port(
ports_rest1 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg1_id,
sg2_id])
ports_rest1 = self.deserialize(self.fmt, res1)
port_id1 = ports_rest1['port']['id']
self.rpc.devices = {port_id1: ports_rest1['port']}
devices = [port_id1, 'no_exist_device']
res2 = self._create_port(
ports_rest2 = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
security_groups=[sg2_id])
ports_rest2 = self.deserialize(self.fmt, res2)
port_id2 = ports_rest2['port']['id']
ctx = context.get_admin_context()
@ -464,12 +727,17 @@ class SGServerRpcCallBackMixinTestCase(test_sg.SecurityGroupDBTestCase):
{'direction': 'egress', 'ethertype': const.IPv6,
'security_group_id': sg2_id},
{'direction': 'ingress',
'source_ip_prefix': 'fe80::3/128',
'source_ip_prefix': '2001:db8::2/128',
'protocol': const.PROTO_NAME_TCP,
'ethertype': const.IPv6,
'port_range_max': 25, 'port_range_min': 24,
'remote_group_id': sg2_id,
'security_group_id': sg1_id},
{'direction': 'ingress',
'protocol': const.PROTO_NAME_ICMP,
'ethertype': const.IPv6,
'source_ip_prefix': fake_gateway,
'source_port_range_min': const.ICMPV6_TYPE_RA},
]
self.assertEqual(port_rpc['security_group_rules'],
expected)
@ -1304,7 +1572,6 @@ IPTABLES_FILTER_V6_1 = """# Generated by iptables_manager
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 130 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 131 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 132 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 134 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 135 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 136 -j RETURN
[0:0] -A %(bn)s-i_port1 -m state --state INVALID -j DROP
@ -1355,7 +1622,6 @@ IPTABLES_FILTER_V6_2 = """# Generated by iptables_manager
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 130 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 131 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 132 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 134 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 135 -j RETURN
[0:0] -A %(bn)s-i_port1 -p icmpv6 --icmpv6-type 136 -j RETURN
[0:0] -A %(bn)s-i_port1 -m state --state INVALID -j DROP
@ -1378,7 +1644,6 @@ IPTABLES_FILTER_V6_2 = """# Generated by iptables_manager
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 130 -j RETURN
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 131 -j RETURN
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 132 -j RETURN
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 134 -j RETURN
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 135 -j RETURN
[0:0] -A %(bn)s-i_port2 -p icmpv6 --icmpv6-type 136 -j RETURN
[0:0] -A %(bn)s-i_port2 -m state --state INVALID -j DROP