FLOATINGIP_STATUS_ERROR = 'ERROR'
DEVICE_OWNER_COMPUTE_PREFIX = "compute:"
-
-DEVICE_OWNER_ROUTER_HA_INTF = "network:router_ha_interface"
-DEVICE_OWNER_ROUTER_INTF = "network:router_interface"
-DEVICE_OWNER_ROUTER_GW = "network:router_gateway"
-DEVICE_OWNER_FLOATINGIP = "network:floatingip"
-DEVICE_OWNER_DHCP = "network:dhcp"
-DEVICE_OWNER_DVR_INTERFACE = "network:router_interface_distributed"
-DEVICE_OWNER_AGENT_GW = "network:floatingip_agent_gateway"
-DEVICE_OWNER_ROUTER_SNAT = "network:router_centralized_snat"
+DEVICE_OWNER_NETWORK_PREFIX = "network:"
+
+DEVICE_OWNER_ROUTER_HA_INTF = (DEVICE_OWNER_NETWORK_PREFIX +
+ "router_ha_interface")
+DEVICE_OWNER_ROUTER_INTF = DEVICE_OWNER_NETWORK_PREFIX + "router_interface"
+DEVICE_OWNER_ROUTER_GW = DEVICE_OWNER_NETWORK_PREFIX + "router_gateway"
+DEVICE_OWNER_FLOATINGIP = DEVICE_OWNER_NETWORK_PREFIX + "floatingip"
+DEVICE_OWNER_DHCP = DEVICE_OWNER_NETWORK_PREFIX + "dhcp"
+DEVICE_OWNER_DVR_INTERFACE = (DEVICE_OWNER_NETWORK_PREFIX +
+ "router_interface_distributed")
+DEVICE_OWNER_AGENT_GW = (DEVICE_OWNER_NETWORK_PREFIX +
+ "floatingip_agent_gateway")
+DEVICE_OWNER_ROUTER_SNAT = (DEVICE_OWNER_NETWORK_PREFIX +
+ "router_centralized_snat")
DEVICE_OWNER_LOADBALANCER = "neutron:LOADBALANCER"
DEVICE_OWNER_LOADBALANCERV2 = "neutron:LOADBALANCERV2"
-DEVICE_OWNER_PREFIXES = ["network:", "neutron:"]
+DEVICE_OWNER_PREFIXES = [DEVICE_OWNER_NETWORK_PREFIX, "neutron:"]
# Collection used to identify devices owned by router interfaces.
# DEVICE_OWNER_ROUTER_HA_INTF is a special case and so is not included.
Trust is currently based on the device_owner field starting with 'network:'
since we restrict who can use that in the default policy.json file.
"""
- return port['device_owner'].startswith('network:')
+ return port['device_owner'].startswith(n_const.DEVICE_OWNER_NETWORK_PREFIX)
class DelayedStringRenderer(object):
context.session.delete(subnetpool)
def _check_mac_addr_update(self, context, port, new_mac, device_owner):
- if (device_owner and device_owner.startswith('network:')):
+ if (device_owner and
+ device_owner.startswith(constants.DEVICE_OWNER_NETWORK_PREFIX)):
raise n_exc.UnsupportedPortDeviceOwner(
op=_("mac address update"), port_id=id,
device_owner=device_owner)
LOG.info(_LI("Skipping ARP spoofing rules for port '%s' because "
"it has port security disabled"), vif.port_name)
return
- if port_details['device_owner'].startswith('network:'):
+ if port_details['device_owner'].startswith(
+ n_const.DEVICE_OWNER_NETWORK_PREFIX):
LOG.debug("Skipping ARP spoofing rules for network owned port "
"'%s'.", vif.port_name)
return
from tempest_lib.common.utils import data_utils
from tempest_lib import exceptions as lib_exc
+from neutron.common import constants
from neutron.tests.api import base
from neutron.tests.tempest import config
from neutron.tests.tempest import test
body = self.client.list_ports()
ports = body['ports']
for port in ports:
- if (port['device_owner'].startswith('network:router_interface')
+ if (port['device_owner'].startswith(
+ constants.DEVICE_OWNER_ROUTER_INTF)
and port['device_id'] in [r['id'] for r in self.routers]):
self.client.remove_router_interface_with_port_id(
port['device_id'], port['id']
subnet_id = _uuid()
return {'admin_state_up': True,
'device_id': _uuid(),
- 'device_owner': 'network:router_ha_interface',
+ 'device_owner': l3_constants.DEVICE_OWNER_ROUTER_HA_INTF,
'fixed_ips': [{'ip_address': ip,
'prefixlen': 18,
'subnet_id': subnet_id}],
# License for the specific language governing permissions and limitations
# under the License.
+from neutron.common import constants
from neutron.plugins.ml2.drivers.linuxbridge.agent import arp_protect
-
from neutron.tests.common import machine_fixtures
from neutron.tests.common import net_helpers
from neutron.tests.functional import base as functional_base
self._add_arp_protection(self.source, ['1.1.1.1'])
no_arping(self.observer.namespace, self.source.ip)
self._add_arp_protection(self.source, ['1.1.1.1'],
- {'device_owner': 'network:router_gateway'})
+ {'device_owner':
+ constants.DEVICE_OWNER_ROUTER_GW})
arping(self.observer.namespace, self.source.ip)
def test_arp_protection_dead_reference_removal(self):
'gateway_ip': float_subnet['gateway_ip'],
'id': fixed_ip['subnet_id']}],
'network_id': external_gw_port['network_id'],
- 'device_owner': 'network:floatingip_agent_gateway',
+ 'device_owner': l3_constants.DEVICE_OWNER_AGENT_GW,
'mac_address': 'fa:16:3e:80:8d:89',
'binding:host_id': self.agent.conf.host,
'fixed_ips': [{'subnet_id': fixed_ip['subnet_id'],
'gateway_ip': snat_subnet['gateway_ip'],
'id': fixed_ip['subnet_id']}],
'network_id': port['network_id'],
- 'device_owner': 'network:router_centralized_snat',
+ 'device_owner': l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:89',
'fixed_ips': [{'subnet_id': fixed_ip['subnet_id'],
'ip_address': snat_ip,
# block first and then disable port security to make sure old rules
# are cleared
self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'])
- self._setup_arp_spoof_for_port(self.dst_p.name, ['192.168.0.3'],
- device_owner='network:router_gateway')
+ self._setup_arp_spoof_for_port(
+ self.dst_p.name, ['192.168.0.3'],
+ device_owner=n_const.DEVICE_OWNER_ROUTER_GW)
self.src_p.addr.add('%s/24' % self.src_addr)
self.dst_p.addr.add('%s/24' % self.dst_addr)
net_helpers.assert_ping(self.src_namespace, self.dst_addr, count=2)
fake_dhcp_port = dhcp.DictModel(dict(id='12345678-1234-aaaa-123456789022',
device_id='dhcp-12345678-1234-aaaa-123456789022',
- device_owner='network:dhcp',
+ device_owner=const.DEVICE_OWNER_DHCP,
allocation_pools=fake_subnet1_allocation_pools,
mac_address='aa:bb:cc:dd:ee:22',
network_id='12345678-1234-5678-1234567890ab',
'gateway_ip': '152.2.0.1',
'id': subnet_id_1}],
'network_id': _uuid(),
- 'device_owner': 'network:router_centralized_snat',
+ 'device_owner':
+ l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_1,
'ip_address': '152.2.0.13',
'gateway_ip': '152.10.0.1',
'id': subnet_id_2}],
'network_id': _uuid(),
- 'device_owner': 'network:router_centralized_snat',
+ 'device_owner':
+ l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_2,
'ip_address': '152.10.0.13',
'gateway_ip': '152.2.0.1',
'id': subnet_id_1}],
'network_id': _uuid(),
- 'device_owner': 'network:router_centralized_snat',
+ 'device_owner':
+ l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_1,
'ip_address': '152.2.0.13',
'gateway_ip': '152.10.0.1',
'id': subnet_id_2}],
'network_id': _uuid(),
- 'device_owner': 'network:router_centralized_snat',
+ 'device_owner':
+ l3_constants.DEVICE_OWNER_ROUTER_SNAT,
'mac_address': 'fa:16:3e:80:8d:80',
'fixed_ips': [{'subnet_id': subnet_id_2,
'ip_address': '152.10.0.13',
ports = ri.router.get(l3_constants.INTERFACE_KEY, [])
subnet_id = l3_test_common.get_subnet_id(ports[0])
test_ports = [{'mac_address': '00:11:22:33:44:55',
- 'device_owner': 'network:dhcp',
+ 'device_owner': l3_constants.DEVICE_OWNER_DHCP,
'fixed_ips': [{'ip_address': '1.2.3.4',
'prefixlen': 24,
'subnet_id': subnet_id}]}]
'gateway_ip': '20.0.0.1'}],
'id': _uuid(),
'binding:host_id': 'myhost',
- 'device_owner': 'network:floatingip_agent_gateway',
+ 'device_owner': l3_constants.DEVICE_OWNER_AGENT_GW,
'network_id': fake_network_id,
'mac_address': 'ca:fe:de:ad:be:ef'}]
)
class FakeDhcpPort(object):
id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa'
admin_state_up = True
- device_owner = 'network:dhcp'
+ device_owner = constants.DEVICE_OWNER_DHCP
fixed_ips = [FakeIPAllocation('192.168.0.1',
'dddddddd-dddd-dddd-dddd-dddddddddddd')]
mac_address = '00:00:80:aa:bb:ee'
class FakeReservedPort(object):
admin_state_up = True
- device_owner = 'network:dhcp'
+ device_owner = constants.DEVICE_OWNER_DHCP
fixed_ips = [FakeIPAllocation('192.168.0.6',
'dddddddd-dddd-dddd-dddd-dddddddddddd')]
mac_address = '00:00:80:aa:bb:ee'
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
- device_owner='network:router_interface')
+ device_owner=const.DEVICE_OWNER_ROUTER_INTF)
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id'],
'ip_address': fake_gateway}],
- device_owner='network:router_interface')
+ device_owner=const.DEVICE_OWNER_ROUTER_INTF)
gateway_mac = gateway_res['port']['mac_address']
gateway_port_id = gateway_res['port']['id']
gateway_lla_ip = str(ipv6.get_ipv6_addr_by_EUI64(
interface_res = self._make_port(
self.fmt, n['network']['id'],
fixed_ips=[{'subnet_id': subnet_v6['subnet']['id']}],
- device_owner='network:router_interface')
+ device_owner=const.DEVICE_OWNER_ROUTER_INTF)
interface_port_id = interface_res['port']['id']
ports_rest1 = self._make_port(
with self.network() as net, self.network() as net1:
with self.subnet(network=net, cidr='10.0.0.0/24') as subnet,\
self.subnet(network=net1, cidr='10.0.1.0/24') as subnet1:
- with self.port(subnet=subnet, device_owner='network:dhcp'),\
+ with self.port(subnet=subnet,
+ device_owner=constants.DEVICE_OWNER_DHCP),\
self.port(subnet=subnet1):
# check that user allocations on another network don't
# affect _subnet_get_user_allocation method
port)
def test_port_without_device_id_no_notify(self):
- port = models_v2.Port(id='port-uuid', device_owner="network:dhcp",
+ port = models_v2.Port(id='port-uuid',
+ device_owner=n_const.DEVICE_OWNER_DHCP,
status=n_const.PORT_STATUS_ACTIVE)
self._record_port_status_changed_helper(n_const.PORT_STATUS_ACTIVE,
sql_attr.NO_VALUE,
def test_non_compute_instances_no_notify(self):
device_id = '32102d7b-1cf4-404d-b50a-97aae1f55f87'
port = models_v2.Port(id='port-uuid', device_id=device_id,
- device_owner="network:dhcp",
+ device_owner=n_const.DEVICE_OWNER_DHCP,
status=n_const.PORT_STATUS_ACTIVE)
self._record_port_status_changed_helper(n_const.PORT_STATUS_ACTIVE,
sql_attr.NO_VALUE,
def test_arp_spoofing_network_port(self):
int_br = mock.create_autospec(self.agent.int_br)
self.agent.setup_arp_spoofing_protection(
- int_br, FakeVif(), {'device_owner': 'network:router_interface'})
+ int_br, FakeVif(),
+ {'device_owner': n_const.DEVICE_OWNER_ROUTER_INTF})
self.assertTrue(int_br.delete_arp_spoofing_protection.called)
self.assertFalse(int_br.install_arp_spoofing_protection.called)
network['network']['tenant_id'],
'name': 'port1',
'device_owner':
- 'network:router_interface_distributed',
+ constants.DEVICE_OWNER_DVR_INTERFACE,
'admin_state_up': 1,
'fixed_ips':
[{'subnet_id': subnet_id}]}}
{
'id': 'dvr_port1',
'device_id': 'r1',
- 'device_owner': 'network:router_interface_distributed',
+ 'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
{
'id': 'dvr_port2',
'device_id': 'r2',
- 'device_owner': 'network:router_interface_distributed',
+ 'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
dvr_port = {
'id': 'dvr_port1',
'device_id': 'r1',
- 'device_owner': 'network:router_interface_distributed',
+ 'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
dvr_port = {
'id': 'dvr_port1',
'device_id': 'r1',
- 'device_owner': 'network:router_interface_distributed',
+ 'device_owner': constants.DEVICE_OWNER_DVR_INTERFACE,
'fixed_ips': [
{
'subnet_id': '80947d4a-fbc8-484b-9f92-623a6bfcf3e0',
oslo_policy.PolicyNotAuthorized)
def test_create_port_device_owner_regex(self):
- blocked_values = ('network:', 'network:abdef', 'network:dhcp',
- 'network:router_interface')
+ blocked_values = (const.DEVICE_OWNER_NETWORK_PREFIX,
+ 'network:abdef',
+ const.DEVICE_OWNER_DHCP,
+ const.DEVICE_OWNER_ROUTER_INTF)
for val in blocked_values:
self._test_advsvc_action_on_attr(
'create', 'port', 'device_owner', val,