x/neutron-classifier
Code Issues Proposed changes

Add Database Models and OVOs for classifications

Browse Source 
- Introducing database models and oslo versioned objects for
   classification resources used by the common classification
   framework service plugin.

Change-Id: I41d5b399352b47d74000596e6518e199d36709a7
Co-Authored-By: David Shaughnessy <david.shaughnessy@intel.com>
This commit is contained in:
Thaynara Silva
2017-08-31 12:07:55 +00:00
committed by Nakul Dahiwade
parent e88886c05c
commit ce22d7eead
18 changed files with 878 additions and 769 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
neutron_classifier/common/constants.py
View File

@@ -14,31 +14,17 @@
# under the License.
CLASSIFIER_TYPES = ['ip_classifier', 'ipv4_classifier', 'ipv6_classifier',
'transport_classifier', 'ethernet_classifier',
'encapsulation_classifier', 'neutron_port_classifier']
from neutron_classifier.objects import classifications as cs
# Protocol names and numbers
PROTO_NAME_ICMP = 'icmp'
PROTO_NAME_ICMP_V6 = 'icmpv6'
PROTO_NAME_TCP = 'tcp'
PROTO_NAME_UDP = 'udp'
FIELDS_IP_V4 = cs.IPV4Classification.fields.keys()
FIELDS_IP_V6 = cs.IPV6Classification.fields.keys()
FIELDS_TCP = cs.TCPClassification.fields.keys()
FIELDS_UDP = cs.UDPClassification.fields.keys()
FIELDS_ETHERNET = cs.EthernetClassification.fields.keys()
# TODO(sc68cal) add more protocols`
PROTOCOLS = [PROTO_NAME_ICMP, PROTO_NAME_ICMP_V6,
PROTO_NAME_TCP, PROTO_NAME_UDP]
ENCAPSULATION_TYPES = ['vxlan', 'gre']
NEUTRON_SERVICES = ['neutron-fwaas', 'networking-sfc', 'security-group']
DIRECTIONS = ['INGRESS', 'EGRESS', 'BIDIRECTIONAL']
ETHERTYPE_IPV4 = 0x0800
ETHERTYPE_IPV6 = 0x86DD
IP_VERSION_4 = 4
IP_VERSION_6 = 6
SECURITYGROUP_ETHERTYPE_IPV4 = 'IPv4'
SECURITYGROUP_ETHERTYPE_IPV6 = 'IPv6'
SUPPORTED_FIELDS = {'ipv4': FIELDS_IP_V4,
'ipv6': FIELDS_IP_V6,
'tcp': FIELDS_TCP,
'udp': FIELDS_UDP,
'ethernet': FIELDS_ETHERNET}

55
neutron_classifier/common/utils.py Normal file
View File

@@ -0,0 +1,55 @@
# Copyright 2011, VMware, Inc.
# All Rights Reserved.
#
# 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.
#
# Borrowed from the Neutron code base, more utilities will be added/borrowed as
# and when needed.
import importlib
import os
import re
import sys
import neutron_classifier
_SEPARATOR_REGEX = re.compile(r'[/\\]+')
def import_modules_recursively(topdir):
'''Import and return all modules below the topdir directory.'''
topdir = _SEPARATOR_REGEX.sub('/', topdir)
modules = []
for root, dirs, files in os.walk(topdir):
for file_ in files:
if file_[-3:] != '.py':
continue
module = file_[:-3]
if module == '__init__':
continue
import_base = _SEPARATOR_REGEX.sub('.', root)
# NOTE(ihrachys): in Python3, or when we are not located in the
# directory containing neutron code, __file__ is absolute, so we
# should truncate it to exclude PYTHONPATH prefix
prefixlen = len(os.path.dirname(neutron_classifier.__file__))
import_base = 'neutron_classifier' + import_base[prefixlen:]
module = '.'.join([import_base, module])
if module not in sys.modules:
importlib.import_module(module)
modules.append(module)
return modules

17
neutron_classifier/db/__init__.py
View File

@@ -0,0 +1,17 @@
# 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.
from neutron.db.migration.models import head
def get_metadata():
return head.model_base.BASEV2.metadata

202
neutron_classifier/db/api.py
View File

@@ -1,202 +0,0 @@
# Copyright (c) 2015 Mirantis, Inc.
# Copyright (c) 2015 Huawei Technologies India Pvt Ltd.
#
# 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.
from neutron_classifier.common import constants
from neutron_classifier.db import models
from neutron_classifier.db import validators
def security_group_ethertype_to_ethertype_value(ethertype):
if ethertype == constants.SECURITYGROUP_ETHERTYPE_IPV6:
return constants.ETHERTYPE_IPV6
else:
return constants.ETHERTYPE_IPV4
def ethertype_value_to_security_group_ethertype(ethertype):
if ethertype == constants.ETHERTYPE_IPV6:
return constants.SECURITYGROUP_ETHERTYPE_IPV6
else:
return constants.SECURITYGROUP_ETHERTYPE_IPV4
def get_classifier_group(context, classifier_group_id):
return context.session.query(models.ClassifierGroup).get(
classifier_group_id)
def create_classifier_chain(classifier_group, classifiers,
incremeting_sequence=False):
if incremeting_sequence:
seq = 0
for classifier in classifiers:
ce = models.ClassifierChainEntry(classifier_group=classifier_group,
classifier=classifier)
if incremeting_sequence:
ce.sequence = seq
classifier_group.classifier_chain.append(ce)
def convert_security_group_to_classifier(context, security_group):
cgroup = models.ClassifierGroup()
cgroup.service = 'security-group'
for rule in security_group['security_group_rules']:
convert_security_group_rule_to_classifier(context, rule, cgroup)
context.session.add(cgroup)
context.session.commit()
return cgroup
def convert_security_group_rule_to_classifier(context, sgr, group):
cl1 = cl2 = cl3 = cl4 = cl5 = None
# Ethertype
if validators.is_ethernetclassifier_valid(sgr, validators.SG_RULE_TYPE):
cl1 = models.EthernetClassifier()
cl1.ethertype = security_group_ethertype_to_ethertype_value(
sgr['ethertype'])
# protocol
if validators.is_protocolclassifier_valid(sgr, validators.SG_RULE_TYPE):
if cl1 and cl1.ethertype == constants.ETHERTYPE_IPV6:
cl2 = models.Ipv6Classifier()
cl2.next_header = sgr['protocol']
else:
cl2 = models.Ipv4Classifier()
cl2.protocol = sgr['protocol']
# remote ip
if validators.is_ipclassifier_valid(sgr, validators.SG_RULE_TYPE):
cl3 = models.IpClassifier()
cl3.source_ip_prefix = sgr['remote_ip_prefix']
# Ports
if validators.is_transportclassifier_valid(sgr, validators.SG_RULE_TYPE):
cl4 = models.TransportClassifier(
destination_port_range_min=sgr['port_range_min'],
destination_port_range_max=sgr['port_range_max'])
# Direction
if validators.is_directionclassifier_valid(sgr, validators.SG_RULE_TYPE):
cl5 = models.DirectionClassifier(direction=sgr['direction'])
classifiers = [cl1, cl2, cl3, cl4, cl5]
create_classifier_chain(group, classifiers)
def convert_classifier_group_to_security_group(context, classifier_group_id):
sg_dict = {}
cg = get_classifier_group(context, classifier_group_id)
for classifier in [link.classifier for link in cg.classifier_chain]:
classifier_type = type(classifier)
if classifier_type is models.TransportClassifier:
sg_dict['port_range_min'] = classifier.destination_port_range_min
sg_dict['port_range_max'] = classifier.destination_port_range_max
continue
if classifier_type is models.IpClassifier:
sg_dict['remote_ip_prefix'] = classifier.source_ip_prefix
continue
if classifier_type is models.DirectionClassifier:
sg_dict['direction'] = classifier.direction
continue
if classifier_type is models.EthernetClassifier:
sg_dict['ethertype'] = ethertype_value_to_security_group_ethertype(
classifier.ethertype)
continue
if classifier_type is models.Ipv4Classifier:
sg_dict['protocol'] = classifier.protocol
continue
if classifier_type is models.Ipv6Classifier:
sg_dict['protocol'] = classifier.next_header
continue
return sg_dict
def convert_firewall_policy_to_classifier(context, firewall):
cgroup = models.ClassifierGroup()
cgroup.service = 'neutron-fwaas'
for rule in firewall['firewall_rules']:
convert_firewall_rule_to_classifier(context, rule, cgroup)
context.session.add(cgroup)
context.session.commit()
return cgroup
def convert_firewall_rule_to_classifier(context, fwr, group):
cl1 = cl2 = cl3 = cl4 = None
# ip_version
if validators.is_ethernetclassifier_valid(fwr, validators.FW_RULE_TYPE):
cl1 = models.EthernetClassifier()
cl1.ethertype = fwr['ip_version']
# protocol
if validators.is_protocolclassifier_valid(fwr, validators.FW_RULE_TYPE):
if cl1.ethertype == constants.IP_VERSION_6:
cl2 = models.Ipv6Classifier()
cl2.next_header = fwr['protocol']
else:
cl2 = models.Ipv4Classifier()
cl2.protocol = fwr['protocol']
# Source and destination ip
if validators.is_ipclassifier_valid(fwr, validators.FW_RULE_TYPE):
cl3 = models.IpClassifier()
cl3.source_ip_prefix = fwr['source_ip_address']
cl3.destination_ip_prefix = fwr['destination_ip_address']
# Ports
if validators.is_transportclassifier_valid(fwr, validators.FW_RULE_TYPE):
cl4 = models.TransportClassifier(
source_port_range_min=fwr['source_port_range_min'],
source_port_range_max=fwr['source_port_range_max'],
destination_port_range_min=fwr['destination_port_range_min'],
destination_port_range_max=fwr['destination_port_range_max'])
classifiers = [cl1, cl2, cl3, cl4]
create_classifier_chain(group, classifiers)
def convert_classifier_to_firewall(context, classifier_group_id):
fw_rule = {}
cg = get_classifier_group(context, classifier_group_id)
for classifier in [link.classifier for link in cg.classifier_chain]:
classifier_type = type(classifier)
if classifier_type is models.EthernetClassifier:
fw_rule['ip_version'] = classifier.ethertype
continue
if classifier_type is models.Ipv4Classifier:
fw_rule['protocol'] = classifier.protocol
continue
if classifier_type is models.Ipv6Classifier:
fw_rule['protocol'] = classifier.next_header
continue
if classifier_type is models.TransportClassifier:
fw_rule['source_port_range_min'] = classifier.source_port_range_min
fw_rule['source_port_range_max'] = classifier.source_port_range_max
fw_rule['destination_port_range_min'] = \
classifier.destination_port_range_min
fw_rule['destination_port_range_max'] = \
classifier.destination_port_range_max
continue
if classifier_type is models.IpClassifier:
fw_rule['source_ip_address'] = classifier.source_ip_prefix
fw_rule['destination_ip_address'] = \
classifier.destination_ip_prefix
continue
return fw_rule

274
neutron_classifier/db/models.py
View File

@@ -1,4 +1,5 @@
# Copyright (c) 2015 Mirantis, Inc.
# Copyright 2017 Intel Corporation.
#
# 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
@@ -12,161 +13,186 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_classifier.common import constants
from oslo_utils import uuidutils
from neutron_lib.db import model_base
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.orderinglist import ordering_list
from sqlalchemy import orm
Base = declarative_base()
# Stolen from neutron/db/model_base.py
class HasTenant(object):
"""Tenant mixin, add to subclasses that have a tenant."""
tenant_id = sa.Column(sa.String(255), index=True)
# Stolen from neutron/db/model_base.py
class HasId(object):
"""id mixin, add to subclasses that have an id."""
id = sa.Column(sa.String(36),
primary_key=True,
default=uuidutils.generate_uuid)
class Classifier(Base, HasId):
__tablename__ = 'classifiers'
classifier_type = sa.Column(sa.String)
__mapper_args__ = {'polymorphic_on': classifier_type}
class ClassifierGroup(Base, HasTenant, HasId):
__tablename__ = 'classifier_groups'
# Service plugin models
class ClassificationGroup(model_base.BASEV2, model_base.HasId,
model_base.HasProject):
__tablename__ = 'classification_groups'
name = sa.Column(sa.String(255))
description = sa.Column(sa.String(255))
classifier_chain = orm.relationship(
'ClassifierChainEntry',
backref=orm.backref('classifier_chains', cascade='all, delete'),
order_by='ClassifierChainEntry.sequence',
collection_class=ordering_list('sequence', count_from=1))
service = sa.Column(sa.Enum(*constants.NEUTRON_SERVICES), index=True)
shared = sa.Column(sa.Boolean(), default=False)
operator = sa.Column(sa.Enum('AND', 'OR'), default='AND')
class ClassifierChainEntry(Base, HasId):
__tablename__ = 'classifier_chains'
classifier_group_id = sa.Column(sa.String(36),
sa.ForeignKey('classifier_groups.id',
ondelete="CASCADE"))
classifier_id = sa.Column(sa.String(36),
sa.ForeignKey('classifiers.id',
ondelete="CASCADE"))
classifier = orm.relationship(Classifier)
sequence = sa.Column(sa.Integer)
classifier_group = orm.relationship(ClassifierGroup)
def __init__(self, classifier_group=None, classifier=None, sequence=None):
super(ClassifierChainEntry, self).__init__()
self.classifier = classifier
self.classifier_group = classifier_group
self.sequence = sequence
class CGToClassificationMapping(model_base.BASEV2):
__tablename__ = 'classification_group_to_classification_mappings'
container_cg_id = sa.Column(sa.String(36),
sa.ForeignKey('classification_groups.id'),
primary_key=True)
stored_classification_id = sa.Column(sa.String(36),
sa.ForeignKey('classifications.id'),
primary_key=True)
class DirectionClassifier(Classifier):
__tablename__ = 'direction_classifiers'
__mapper_args__ = {'polymorphic_identity': 'directionclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
class CGToClassificationGroupMapping(model_base.BASEV2):
__tablename__ = 'classification_group_to_cg_mappings'
container_cg_id = sa.Column(sa.String(36),
sa.ForeignKey('classification_groups.id'),
primary_key=True)
stored_cg_id = sa.Column(sa.String(36),
sa.ForeignKey('classification_groups.id'),
primary_key=True)
class ClassificationBase(Base, model_base.HasId, model_base.HasProject,
model_base.BASEV2):
__tablename__ = 'classifications'
c_type = sa.Column(sa.String(36))
__mapper_args__ = {'polymorphic_on': c_type}
name = sa.Column(sa.String(255))
description = sa.Column(sa.String(255))
shared = sa.Column(sa.Boolean())
negated = sa.Column(sa.Boolean())
class IPV4Classification(ClassificationBase):
__tablename__ = 'ipv4_classifications'
__mapper_args__ = {'polymorphic_identity': 'ipv4'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifications.id'),
primary_key=True)
direction = sa.Column(sa.Enum(*constants.DIRECTIONS))
def __init__(self, direction=None):
super(DirectionClassifier, self).__init__()
self.direction = direction
dscp = sa.Column(sa.Integer())
dscp_mask = sa.Column(sa.Integer())
ecn = sa.Column(sa.Enum("0", "1", "2", "3"))
length_min = sa.Column(sa.Integer())
length_max = sa.Column(sa.Integer())
flags = sa.Column(sa.Integer())
flags_mask = sa.Column(sa.Integer())
ttl_min = sa.Column(sa.SmallInteger())
ttl_max = sa.Column(sa.SmallInteger())
protocol = sa.Column(sa.Integer())
src_addr = sa.Column(sa.String(19))
dst_addr = sa.Column(sa.String(19))
class EncapsulationClassifier(Classifier):
__tablename__ = 'encapsulation_classifiers'
__mapper_args__ = {'polymorphic_identity': 'encapsulationclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
class IPV6Classification(ClassificationBase):
__tablename__ = 'ipv6_classifications'
__mapper_args__ = {'polymorphic_identity': 'ipv6'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifications.id'),
primary_key=True)
encapsulation_type = sa.Column(sa.Enum(*constants.ENCAPSULATION_TYPES))
encapsulation_id = sa.Column(sa.String(255))
dscp = sa.Column(sa.Integer())
dscp_mask = sa.Column(sa.Integer())
ecn = sa.Column(sa.Enum("0", "1", "2", "3"))
length_min = sa.Column(sa.Integer())
length_max = sa.Column(sa.Integer())
next_header = sa.Column(sa.Integer())
hops_min = sa.Column(sa.SmallInteger())
hops_max = sa.Column(sa.SmallInteger())
src_addr = sa.Column(sa.String(49))
dst_addr = sa.Column(sa.String(49))
class EthernetClassifier(Classifier):
__tablename__ = 'ethernet_classifiers'
__mapper_args__ = {'polymorphic_identity': 'ethernetclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
class EthernetClassification(ClassificationBase):
__tablename__ = 'ethernet_classifications'
__mapper_args__ = {'polymorphic_identity': 'ethernet'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifications.id'),
primary_key=True)
ethertype = sa.Column(sa.Integer)
source_mac = sa.Column(sa.String(255))
destination_mac = sa.Column(sa.String(255))
ethertype = sa.Column(sa.Integer())
src_addr = sa.Column(sa.String(17))
dst_addr = sa.Column(sa.String(17))
class IpClassifier(Classifier):
__tablename__ = 'ip_classifiers'
__mapper_args__ = {'polymorphic_identity': 'ipclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
class UDPClassification(ClassificationBase):
__tablename__ = 'udp_classifications'
__mapper_args__ = {'polymorphic_identity': 'udp'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifications.id'),
primary_key=True)
source_ip_prefix = sa.Column(sa.String(255))
destination_ip_prefix = sa.Column(sa.String(255))
src_port_min = sa.Column(sa.Integer)
src_port_max = sa.Column(sa.Integer)
dst_port_min = sa.Column(sa.Integer)
dst_port_max = sa.Column(sa.Integer)
length_min = sa.Column(sa.Integer())
length_max = sa.Column(sa.Integer())
class Ipv4Classifier(Classifier):
__tablename__ = 'ipv4_classifiers'
__mapper_args__ = {'polymorphic_identity': 'ipv4classifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
class TCPClassification(ClassificationBase):
__tablename__ = 'tcp_classifications'
__mapper_args__ = {'polymorphic_identity': 'tcp'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifications.id'),
primary_key=True)
dscp_tag = sa.Column(sa.String(255))
protocol = sa.column(sa.Enum(*constants.PROTOCOLS))
dscp_mask = sa.Column(sa.String(255))
src_port_min = sa.Column(sa.Integer)
src_port_max = sa.Column(sa.Integer)
dst_port_min = sa.Column(sa.Integer)
dst_port_max = sa.Column(sa.Integer)
flags = sa.Column(sa.Integer())
flags_mask = sa.Column(sa.Integer())
window_min = sa.Column(sa.Integer())
window_max = sa.Column(sa.Integer())
class Ipv6Classifier(Classifier):
__tablename__ = 'ipv6_classifiers'
__mapper_args__ = {'polymorphic_identity': 'ipv6classifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
primary_key=True)
next_header = sa.Column(sa.Enum(*constants.PROTOCOLS))
traffic_class = sa.Column(sa.String(255))
flow_label = sa.Column(sa.String(255))
def _read_classification_groups(plugin, context, id=None):
class_group = plugin._get_collection(context, ClassificationGroup,
_generate_dict_from_cg_db)
cg_m_c = plugin._get_collection(context, CGToClassificationMapping,
_generate_dict_from_cgmapping_db)
cg_m_cg = plugin._get_collection(context, CGToClassificationGroupMapping,
_generate_dict_from_cgmapping_db)
id_class = None
for cg in class_group:
class_ids = []
group_ids = []
if id and cg['id'] != id:
continue
for mapping in cg_m_c:
if cg['id'] == mapping['container_cg_id']:
class_ids.append(mapping['stored_classification_id'])
for mapping in cg_m_cg:
if cg['id'] == mapping['container_cg_id']:
group_ids.append(mapping['stored_cg_id'])
cg['classifications'] = ','.join(str(x) for x in class_ids)
cg['classification_groups'] = ','.join(str(x) for x in group_ids)
id_class = cg
if id:
return id_class
else:
return class_group
class NeutronPortClassifier(Classifier):
__tablename__ = 'neutron_port_classifiers'
__mapper_args__ = {'polymorphic_identity': 'neutronportclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
primary_key=True)
logical_source_port = sa.Column(sa.String(255))
logical_destination_port = sa.Column(sa.String(255))
def _generate_dict_from_cg_db(model, fields=None):
resp = {}
resp['id'] = model.id
resp['name'] = model.name
resp['description'] = model.description
resp['project_id'] = model.project_id
resp['classifications'] = ''
resp['classification_groups'] = ''
resp['shared'] = model.shared
resp['operator'] = model.operator
return resp
class TransportClassifier(Classifier):
__tablename__ = 'transport_classifiers'
__mapper_args__ = {'polymorphic_identity': 'transportclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
primary_key=True)
source_port_range_max = sa.Column(sa.Integer)
source_port_range_min = sa.Column(sa.Integer)
destination_port_range_max = sa.Column(sa.Integer)
destination_port_range_min = sa.Column(sa.Integer)
def _generate_dict_from_cgmapping_db(model, fields=None):
resp = {}
def __init__(self, source_port_range_min=None,
source_port_range_max=None,
destination_port_range_min=None,
destination_port_range_max=None):
super(TransportClassifier, self).__init__()
self.destination_port_range_min = destination_port_range_min
self.destination_port_range_max = destination_port_range_max
self.source_port_range_min = source_port_range_min
self.source_port_range_max = source_port_range_max
resp['container_cg_id'] = model.container_cg_id
if 'stored_cg_id' in model:
resp['stored_cg_id'] = model.stored_cg_id
else:
resp['stored_classification_id'] = model.stored_classification_id
return resp
class VlanClassifier(Classifier):
__tablename__ = 'vlan_classifiers'
__mapper_args__ = {'polymorphic_identity': 'vlanclassifier'}
id = sa.Column(sa.String(36), sa.ForeignKey('classifiers.id'),
primary_key=True)
vlan_priority = sa.Column(sa.Integer)
RESOURCE_MODELS = {'ipv4_classification': IPV4Classification,
'ipv6_classification': IPV6Classification,
'tcp_classification': TCPClassification,
'udp_classification': UDPClassification,
'ethernet_classification': EthernetClassification}

26
neutron_classifier/db/rbac_db_models.py Normal file
View File

@@ -0,0 +1,26 @@
# Copyright 2017 Intel Corporation.
#
# 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.
from neutron.db import rbac_db_models
from neutron_lib.db import model_base
class ClassificationGroupRBAC(rbac_db_models.RBACColumns, model_base.BASEV2):
"""RBAC table for classification groups."""
object_id = rbac_db_models._object_id_column('classification_groups.id')
object_type = 'classification_group'
def get_valid_actions(self):
return (rbac_db_models.ACCESS_SHARED, rbac_db_models.rbac_db_models, )

140
neutron_classifier/db/validators.py
View File

@@ -1,140 +0,0 @@
# Copyright (c) 2016 Huawei Technologies India Pvt Ltd.
#
# 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.
from neutron_classifier.common import constants as const
from neutron_classifier.common import exceptions as exc
import netaddr
SG_RULE_TYPE = 1
FW_RULE_TYPE = 2
def get_attr_value(dict, key):
return dict.get(key, None)
def _validate_fwr_protocol_parameters(fwr, protocol):
"""Check if given port values and protocol is valid."""
if protocol in (const.PROTO_NAME_ICMP, const.PROTO_NAME_ICMP_V6):
source_port_range_min = get_attr_value(fwr, 'source_port_range_min')
source_port_range_max = get_attr_value(fwr, 'source_port_range_max')
destination_port_range_min = get_attr_value(
fwr, 'destination_port_range_min')
destination_port_range_max = get_attr_value(
fwr, 'destination_port_range_max')
if (source_port_range_min or source_port_range_max or
destination_port_range_min or destination_port_range_max):
raise exc.InvalidICMPParameter(param="Source, destination port")
def _validate_sg_ethertype_and_protocol(rule, protocol):
"""Check if given ethertype and protocol is valid."""
eth_value = get_attr_value(rule, 'ethertype')
if protocol == const.PROTO_NAME_ICMP_V6:
if eth_value == const.SECURITYGROUP_ETHERTYPE_IPV4:
raise exc.EthertypeConflictWithProtocol(ethertype=eth_value,
protocol=protocol)
def validate_port_range(min_port, max_port):
"""Check that port_range is valid."""
port_range = '%s:%s' % (min_port, max_port)
if(min_port is None and max_port is None):
return
if (int(min_port) <= 0 or int(max_port) <= 0):
raise exc.InvalidPortRange(port_range=port_range)
if int(min_port) > int(max_port):
raise exc.InvalidPortRange(port_range=port_range)
def is_ethernetclassifier_valid(rule, type):
"""Check ethertype or ip_version in rule dict."""
if type == SG_RULE_TYPE:
attr_type = 'ethertype'
attr_list = [const.SECURITYGROUP_ETHERTYPE_IPV4,
const.SECURITYGROUP_ETHERTYPE_IPV6]
else:
attr_type = 'ip_version'
attr_list = [const.IP_VERSION_4, const.IP_VERSION_6]
eth_value = get_attr_value(rule, attr_type)
if not eth_value:
return False
elif eth_value not in attr_list:
raise exc.InvalidEthernetClassifier(eth_type=attr_type)
return True
def is_protocolclassifier_valid(rule, type):
"""Check protocol in rule dict and validate dependent params"""
protocol = get_attr_value(rule, 'protocol')
if not protocol:
return False
if type == SG_RULE_TYPE:
_validate_sg_ethertype_and_protocol(rule, protocol)
else:
_validate_fwr_protocol_parameters(rule, protocol)
return True
def is_ipclassifier_valid(rule, type):
"""validate the ip address received in rule dict"""
src_ip_version = dst_ip_version = None
src_ip_address = dst_ip_address = None
if type == SG_RULE_TYPE:
dst_ip_address = get_attr_value(rule, 'remote_ip_prefix')
attr_type = 'ethertype'
else:
src_ip_address = get_attr_value(rule, 'source_ip_address')
dst_ip_address = get_attr_value(rule, 'destination_ip_address')
attr_type = 'ip_version'
if src_ip_address:
src_ip_version = netaddr.IPNetwork(src_ip_address).version
if dst_ip_address:
dst_ip_version = netaddr.IPNetwork(dst_ip_address).version
rule_ip_version = get_attr_value(rule, attr_type)
if type == SG_RULE_TYPE:
if rule_ip_version != "IPv%d" % dst_ip_version:
raise exc.IpAddressConflict()
elif ((src_ip_version and src_ip_version != rule_ip_version) or
(dst_ip_version and dst_ip_version != rule_ip_version)):
raise exc.IpAddressConflict()
return True
def is_directionclassifier_valid(rule, type):
"""Check direction param in rule dict"""
direction = get_attr_value(rule, 'direction')
if not direction:
return False
return True
def is_transportclassifier_valid(rule, type):
"""Verify port range values"""
if type == SG_RULE_TYPE:
port_range_min = get_attr_value(rule, 'port_range_min')
port_range_max = get_attr_value(rule, 'port_range_max')
validate_port_range(port_range_min, port_range_max)
else:
source_port_range_min = get_attr_value(rule, 'source_port_range_min')
source_port_range_max = get_attr_value(rule, 'source_port_range_max')
destination_port_range_min = get_attr_value(
rule, 'destination_port_range_min')
destination_port_range_max = get_attr_value(
rule, 'destination_port_range_max')
validate_port_range(source_port_range_min, source_port_range_max)
validate_port_range(destination_port_range_min,
destination_port_range_max)
return True

17
neutron_classifier/objects/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
# 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.
def register_objects():
# local import to avoid circular import failure
__import__('neutron_classifier.objects.classifications')
__import__('neutron_classifier.objects.classification_type')

36
neutron_classifier/objects/classification_type.py Normal file
View File

@@ -0,0 +1,36 @@
# Copyright (c) 2017 Intel Corporation.
#
# 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.
from neutron.objects import base
from neutron_classifier.common import constants
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
@obj_base.VersionedObjectRegistry.register
class ClassificationType(base.NeutronObject):
VERSION = '1.0'
fields = {
'type': obj_fields.StringField(),
'supported_parameters': obj_fields.ListOfStringsField(),
}
@classmethod
def get_object(cls, classification_type, **kwargs):
parameters = constants.SUPPORTED_FIELDS[classification_type]
return cls(type=classification_type, supported_parameters=parameters)

273
neutron_classifier/objects/classifications.py Normal file
View File

@@ -0,0 +1,273 @@
# Copyright 2017 Intel Corporation.
#
# 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.
import abc
import six
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
from neutron.db import api as db_api
from neutron.objects import base
from neutron.objects import common_types
from neutron.objects import rbac_db
from neutron_classifier.db import models
from neutron_classifier.db.rbac_db_models import ClassificationGroupRBAC
@obj_base.VersionedObjectRegistry.register
class ClassificationGroup(rbac_db.NeutronRbacObject):
# Version 1.0: Initial version
VERSION = '1.0'
# required by RbacNeutronMetaclass
rbac_db_model = ClassificationGroupRBAC
db_model = models.ClassificationGroup
fields = {
'id': common_types.UUIDField(),
'name': obj_fields.StringField(),
'description': obj_fields.StringField(),
'project_id': obj_fields.StringField(),
'shared': obj_fields.BooleanField(default=False),
'operator': obj_fields.EnumField(['AND', 'OR'], default='AND'),
}
fields_no_update = ['id', 'project_id']
@classmethod
def get_object(cls, context, **kwargs):
# We want to get the policy regardless of its tenant id. We'll make
# sure the tenant has permission to access the policy later on.
admin_context = context.elevated()
with db_api.autonested_transaction(admin_context.session):
obj = super(ClassificationGroup, cls).get_object(admin_context,
**kwargs)
if not obj or not cls.is_accessible(context, obj):
return
return obj
@obj_base.VersionedObjectRegistry.register
class CGToClassificationMapping(base.NeutronDbObject):
VERSION = '1.0'
rbac_db_model = ClassificationGroupRBAC
db_model = models.CGToClassificationMapping
fields = {
'container_cg_id': obj_fields.ObjectField('ClassificationGroup',
subclasses=True),
'store_classification_id': obj_fields.ObjectField('ClassificationBase',
subclasses=True),
}
@obj_base.VersionedObjectRegistry.register
class CGToClassificationGroupMapping(base.NeutronDbObject):
VERSION = '1.0'
rbac_db_model = ClassificationGroupRBAC
db_model = models.CGToClassificationGroupMapping
fields = {
'container_cg_id': obj_fields.ObjectField('ClassificationGroup',
subclasses=True),
'stored_cg_id': obj_fields.ObjectField('ClassificationGroup',
subclasses=True),
}
@six.add_metaclass(abc.ABCMeta)
class ClassificationBase(base.NeutronDbObject):
VERSION = '1.0'
db_model = models.ClassificationBase
fields = {
'id': common_types.UUIDField(),
'name': obj_fields.StringField(),
'description': obj_fields.StringField(),
'project_id': obj_fields.StringField(),
'shared': obj_fields.BooleanField(default=False),
'c_type': obj_fields.StringField(),
'negated': obj_fields.BooleanField(default=False),
}
fields_no_update = ['id', 'c_type']
@classmethod
def get_objects(cls, context, _pager=None, validate_filters=True,
**kwargs):
with db_api.autonested_transaction(context.session):
objects = super(ClassificationBase,
cls).get_objects(context, _pager,
validate_filters,
**kwargs)
return objects
@obj_base.VersionedObjectRegistry.register
class IPV4Classification(ClassificationBase):
VERSION = '1.0'
db_model = models.IPV4Classification
fields = {
'dscp': obj_fields.IntegerField(nullable=True),
'dscp_mask': obj_fields.IntegerField(nullable=True),
'ecn': obj_fields.EnumField(valid_values=["0", "1", "2", "3"],
nullable=True),
'length_min': obj_fields.IntegerField(nullable=True),
'length_max': obj_fields.IntegerField(nullable=True),
'flags': obj_fields.IntegerField(nullable=True),
'flags_mask': obj_fields.IntegerField(nullable=True),
'ttl_min': obj_fields.IntegerField(nullable=True),
'ttl_max': obj_fields.IntegerField(nullable=True),
'protocol': obj_fields.IntegerField(nullable=True),
'src_addr': obj_fields.StringField(nullable=True),
'dst_addr': obj_fields.StringField(nullable=True),
}
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(ClassificationBase, self).create()
@classmethod
def get_object(cls, context, **kwargs):
with db_api.autonested_transaction(context.session):
obj = super(IPV4Classification,
cls).get_object(context, c_type='ipv4',
**kwargs)
return obj
@obj_base.VersionedObjectRegistry.register
class IPV6Classification(ClassificationBase):
VERSION = '1.0'
db_model = models.IPV6Classification
fields = {
'dscp': obj_fields.IntegerField(nullable=True),
'dscp_mask': obj_fields.IntegerField(nullable=True),
'ecn': obj_fields.EnumField(valid_values=["0", "1", "2", "3"],
nullable=True),
'length_min': obj_fields.IntegerField(nullable=True),
'length_max': obj_fields.IntegerField(nullable=True),
'next_header': obj_fields.IntegerField(nullable=True),
'hops_min': obj_fields.IntegerField(nullable=True),
'hops_max': obj_fields.IntegerField(nullable=True),
'src_addr': obj_fields.StringField(nullable=True),
'dst_addr': obj_fields.StringField(nullable=True),
}
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(ClassificationBase, self).create()
@classmethod
def get_object(cls, context, **kwargs):
with db_api.autonested_transaction(context.session):
obj = super(IPV6Classification,
cls).get_object(context, c_type='ipv6',
**kwargs)
return obj
@obj_base.VersionedObjectRegistry.register
class EthernetClassification(ClassificationBase):
VERSION = '1.0'
db_model = models.EthernetClassification
fields = {
'ethertype': obj_fields.IntegerField(nullable=True),
'src_addr': obj_fields.StringField(nullable=True),
'dst_addr': obj_fields.StringField(nullable=True),
}
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(ClassificationBase, self).create()
@classmethod
def get_object(cls, context, **kwargs):
with db_api.autonested_transaction(context.session):
obj = super(EthernetClassification,
cls).get_object(context, c_type='ethernet',
**kwargs)
return obj
@obj_base.VersionedObjectRegistry.register
class UDPClassification(ClassificationBase):
VERSION = '1.0'
db_model = models.UDPClassification
fields = {
'src_port_min': obj_fields.IntegerField(nullable=True),
'src_port_max': obj_fields.IntegerField(nullable=True),
'dst_port_min': obj_fields.IntegerField(nullable=True),
'dst_port_max': obj_fields.IntegerField(nullable=True),
'length_min': obj_fields.IntegerField(nullable=True),
'length_max': obj_fields.IntegerField(nullable=True),
}
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(ClassificationBase, self).create()
@classmethod
def get_object(cls, context, **kwargs):
with db_api.autonested_transaction(context.session):
obj = super(UDPClassification,
cls).get_object(context, c_type='udp',
**kwargs)
return obj
@obj_base.VersionedObjectRegistry.register
class TCPClassification(ClassificationBase):
VERSION = '1.0'
db_model = models.TCPClassification
fields = {
'src_port_min': obj_fields.IntegerField(nullable=True),
'src_port_max': obj_fields.IntegerField(nullable=True),
'dst_port_min': obj_fields.IntegerField(nullable=True),
'dst_port_max': obj_fields.IntegerField(nullable=True),
'flags': obj_fields.IntegerField(nullable=True),
'flags_mask': obj_fields.IntegerField(nullable=True),
'window_min': obj_fields.IntegerField(nullable=True),
'window_max': obj_fields.IntegerField(nullable=True),
}
def create(self):
with db_api.autonested_transaction(self.obj_context.session):
super(ClassificationBase, self).create()
@classmethod
def get_object(cls, context, **kwargs):
with db_api.autonested_transaction(context.session):
obj = super(TCPClassification,
cls).get_object(context, c_type='tcp',
**kwargs)
return obj
CLASS_MAP = {'ipv4': IPV4Classification,
'ethernet': EthernetClassification,
'ipv6': IPV6Classification,
'udp': UDPClassification,
'tcp': TCPClassification}

39
neutron_classifier/tests/base.py
View File

@@ -1,20 +1,35 @@
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# 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
# 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
# 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.
# 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.
from oslotest import base
import mock
from neutron.api.rpc.callbacks import resource_manager
from neutron.tests import base
class TestCase(base.BaseTestCase):
"""Test case base class for all tests."""
class BaseClassificationTestCase(base.BaseTestCase):
def setUp(self):
super(BaseClassificationTestCase, self).setUp()
with mock.patch.object(
resource_manager.ResourceCallbacksManager, '_singleton',
new_callable=mock.PropertyMock(return_value=False)):
self.consumer_manager = resource_manager.\
ConsumerResourceCallbacksManager()
self.producer_manager = resource_manager.\
ProducerResourceCallbacksManager()
for manager in (self.consumer_manager, self.producer_manager):
manager.clear()

2
neutron_classifier/tests/functional/test_placeholder.py
View File

@@ -1,7 +1,7 @@
from neutron_classifier.tests import base
class PlaceholderTest(base.TestCase):
class PlaceholderTest(base.BaseClassificationTestCase):
def test_noop(self):
pass

0
neutron_classifier/tests/unit/db/__init__.py Normal file
View File

190
neutron_classifier/tests/unit/db/test_models.py Normal file
View File

@@ -0,0 +1,190 @@
# Copyright 2017 Intel Corporation.
#
# 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.
import copy
from neutron_classifier.db import models
from neutron_classifier.tests import base
from neutron_lib import context
from oslo_utils import uuidutils
class TestDatabaseModels(base.BaseClassificationTestCase):
class _MockServicePlugin(object):
def __init__(self):
self.cg_list = []
standard_group = {'name': "Test Group",
'description': "Description of test group",
'project_id': uuidutils.generate_uuid(),
'classifications': '',
'classification_groups': '',
'shared': True,
'operator': 'AND'}
for n in range(5):
self.cg_list.append(copy.copy(standard_group))
self.cg_list[-1]['id'] = uuidutils.generate_uuid()
self.cg_list[-1]['name'] = "Test Group " + str(n)
self.cg_to_c_list = [{'container_cg_id': self.cg_list[0]['id'],
'stored_classification_id':
uuidutils.generate_uuid()},
{'container_cg_id': self.cg_list[0]['id'],
'stored_classification_id':
uuidutils.generate_uuid()},
{'container_cg_id': self.cg_list[0]['id'],
'stored_classification_id':
uuidutils.generate_uuid()},
{'container_cg_id': self.cg_list[1]['id'],
'stored_classification_id':
uuidutils.generate_uuid()},
{'container_cg_id': self.cg_list[2]['id'],
'stored_classification_id':
uuidutils.generate_uuid()},
{'container_cg_id': self.cg_list[3]['id'],
'stored_classification_id':
uuidutils.generate_uuid()}]
self.cg_to_cg_list = [{'container_cg_id': self.cg_list[0]['id'],
'stored_cg_id': self.cg_list[1]['id']},
{'container_cg_id': self.cg_list[0]['id'],
'stored_cg_id': self.cg_list[2]['id']},
{'container_cg_id': self.cg_list[0]['id'],
'stored_cg_id': self.cg_list[3]['id']},
{'container_cg_id': self.cg_list[4]['id'],
'stored_cg_id': self.cg_list[1]['id']},
{'container_cg_id': self.cg_list[4]['id'],
'stored_cg_id': self.cg_list[2]['id']},
{'container_cg_id': self.cg_list[4]['id'],
'stored_cg_id': self.cg_list[3]['id']}]
def _create_fake_model(self, model):
model1 = model()
if model == models.ClassificationGroup:
model1.id = self.cg_list[0]['id']
model1.name = self.cg_list[0]['name']
model1.description = self.cg_list[0]['description']
model1.project_id = self.cg_list[0]['project_id']
model1.classifications = ''
model1.classification_groups = ''
model1.shared = self.cg_list[0]['shared']
model1.operator = self.cg_list[0]['operator']
elif model == models.CGToClassificationMapping:
model1.container_cg_id = self.cg_to_c_list[0][
'container_cg_id']
model1.stored_classification_id = self.cg_to_c_list[0][
'stored_classification_id']
elif model == models.CGToClassificationGroupMapping:
model1.container_cg_id = self.cg_to_cg_list[0][
'container_cg_id']
model1.stored_cg_id = self.cg_to_cg_list[0]['stored_cg_id']
return model1
def _get_collection(self, context, model, function):
if model == models.ClassificationGroup:
return copy.deepcopy(self.cg_list)
elif model == models.CGToClassificationMapping:
return copy.deepcopy(self.cg_to_c_list)
elif model == models.CGToClassificationGroupMapping:
return copy.deepcopy(self.cg_to_cg_list)
def setUp(self):
super(TestDatabaseModels, self).setUp()
self.ctxt = context.Context('fake_user', 'fake_tenant')
self.mock_plugin = self._MockServicePlugin()
def test_read_classification_groups_without_id(self):
ret = models._read_classification_groups(self.mock_plugin, self.ctxt)
cg_ids = []
for cg in ret:
cg_ids.append(cg['id'])
if cg['name'] == 'Test Group 0':
class_group_0 = cg
if cg['name'] == 'Test Group 4':
class_group_4 = cg
self.assertIn(self.mock_plugin.cg_list[0]['id'], cg_ids)
self.assertIn(self.mock_plugin.cg_list[1]['id'], cg_ids)
self.assertIn(self.mock_plugin.cg_list[2]['id'], cg_ids)
self.assertIn(self.mock_plugin.cg_list[3]['id'], cg_ids)
self.assertIn(self.mock_plugin.cg_list[4]['id'], cg_ids)
self.assertIn(self.mock_plugin.cg_list[1]['id'],
class_group_0['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[2]['id'],
class_group_0['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[3]['id'],
class_group_0['classification_groups'])
self.assertNotIn(self.mock_plugin.cg_list[4]['id'],
class_group_0['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[1]['id'],
class_group_4['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[2]['id'],
class_group_4['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[3]['id'],
class_group_4['classification_groups'])
self.assertNotIn(self.mock_plugin.cg_list[0]['id'],
class_group_4['classification_groups'])
def test_read_classification_groups_with_id(self):
funct = models._read_classification_groups
ret = funct(self.mock_plugin, self.ctxt,
self.mock_plugin.cg_list[0]['id'])
self.assertEqual(ret['name'], self.mock_plugin.cg_list[0]['name'])
self.assertEqual(ret['id'], self.mock_plugin.cg_list[0]['id'])
self.assertIn(self.mock_plugin.cg_list[1]['id'],
ret['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[2]['id'],
ret['classification_groups'])
self.assertIn(self.mock_plugin.cg_list[3]['id'],
ret['classification_groups'])
self.assertNotIn(self.mock_plugin.cg_list[4]['id'],
ret['classification_groups'])
self.assertNotEqual(self.mock_plugin.cg_list[0]['classifications'],
ret['classifications'])
def test_generate_dict_from_cg_db(self):
model = self.mock_plugin._create_fake_model(models.ClassificationGroup)
ret = models._generate_dict_from_cg_db(model)
self.assertEqual(ret['name'], model.name)
self.assertEqual(ret['id'], model.id)
self.assertEqual(ret['description'], model.description)
self.assertEqual(ret['project_id'], model.project_id)
self.assertEqual(ret['classifications'], model.classifications)
self.assertEqual(ret['classification_groups'],
model.classification_groups)
self.assertEqual(ret['shared'], model.shared)
self.assertEqual(ret['operator'], model.operator)
def test_generate_dict_from_cgmapping_db(self):
model1 = self.mock_plugin._create_fake_model(
models.CGToClassificationMapping)
model2 = self.mock_plugin._create_fake_model(
models.CGToClassificationGroupMapping)
ret1 = models._generate_dict_from_cgmapping_db(model1)
ret2 = models._generate_dict_from_cgmapping_db(model2)
self.assertEqual(ret1['container_cg_id'], model1.container_cg_id)
self.assertEqual(ret1['stored_classification_id'],
model1.stored_classification_id)
self.assertEqual(ret2['container_cg_id'], model2.container_cg_id)
self.assertEqual(ret2['stored_cg_id'], model2.stored_cg_id)

0
neutron_classifier/tests/unit/objects/__init__.py Normal file
View File

74
neutron_classifier/tests/unit/objects/test_objects.py Normal file
View File

@@ -0,0 +1,74 @@
# Copyright 2017 Intel Corporation.
#
# 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.
# NOTE(davidsha) This file is largely a copy of the test_object.py file
# from Neutron
import os
import pprint
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fixture
from neutron import objects as n_obj
from neutron_classifier import objects
from neutron_classifier.tests import base as test_base
# NOTE: The hashes in this list should only be changed if they come with a
# corresponding version bump in the affected objects. Please keep the list in
# alphabetic order.
# This list also includes VersionedObjects from Neutron that are registered
# through dependencies.
object_data = {
'ClassificationGroup': '1.0-e621ff663f76bb494072872222f5fe72',
'CGToClassificationGroupMapping': '1.0-47db29213742e0d60f690daed48ab602',
'CGToClassificationMapping': '1.0-ef48876668c8df6603f105cf7eab16bb',
'ClassificationType': '1.0-699af4d3a1ec14a7d24fefb19c44c641',
'EthernetClassification': '1.0-267f03162a6e011197b663ee34e6cb0b',
'IPV4Classification': '1.0-d4f25a09ceaad9ec817dcebb3b5c4e78',
'IPV6Classification': '1.0-1051e98146a016522d516fe1bec49079',
'TCPClassification': '1.0-1c8a4bb3b2dcdebe8913adc00788c704',
'UDPClassification': '1.0-e55c7b58b9e2c7587cf9a0113225586b'}
class TestObjectVersions(test_base.BaseClassificationTestCase):
def setUp(self):
super(TestObjectVersions, self).setUp()
# NOTE(davidsha): Neutron Classifier OvO's need to be seeded,
# There also appears to be some versioned objects leaking in from
# Neutron from dependencies.
# Because of this I've included all Neutron OvO's and added them
# to the local object_data variable.
# This dependency will prevent upgrades to a neutron OvO from breaking
# this test if they were stored statically here.
objects.register_objects()
n_obj.register_objects()
def test_versions(self):
checker = fixture.ObjectVersionChecker(
obj_base.VersionedObjectRegistry.obj_classes())
fingerprints = checker.get_hashes()
if os.getenv('GENERATE_HASHES'):
with open('object_hashes.txt', 'w') as hashes_file:
hashes_file.write(pprint.pformat(fingerprints))
expected, actual = checker.test_hashes(object_data)
self.assertEqual(expected, actual,
'Some objects have changed; please make sure the '
'versions have been bumped, and then update their '
'hashes in the object_data map in this test module.')

265
neutron_classifier/tests/unit/test_db_api.py
View File

@@ -1,265 +0,0 @@
# Copyright (c) 2015 Mirantis, Inc.
# Copyright (c) 2015 Huawei Technologies India Pvt Ltd.
#
# 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.
import copy as cp
from neutron_classifier.db import api
from neutron_classifier.db import models
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
from oslo_utils import uuidutils
from oslotest import base
FAKE_SG_RULE_V6 = {'direction': 'INGRESS', 'protocol': 'tcp', 'ethertype':
'IPv6', 'tenant_id': 'fake_tenant', 'port_range_min': 80,
'port_range_max': 80, 'remote_ip_prefix':
'fddf:cb3b:bc4::/48', }
FAKE_SG_RULE_V4 = {'direction': 'INGRESS', 'protocol': 'tcp', 'ethertype':
'IPv4', 'tenant_id': 'fake_tenant', 'port_range_min': 80,
'port_range_max': 80, 'remote_ip_prefix': '10.0.0.0/8', }
FAKE_SG_V6 = {'name': 'fake security group', 'tenant_id':
uuidutils.generate_uuid(), 'description': 'this is fake',
'security_group_rules': [FAKE_SG_RULE_V6]}
FAKE_FW_RULE_V4 = {'ip_version': 4, 'protocol': 'udp',
'source_port_range_min': 1, 'source_port_range_max': 80,
'destination_port_range_min': 1,
'destination_port_range_max': 80,
'source_ip_address': '20.1.1.1/24',
'destination_ip_address': '30.1.1.1/24',
'position': 1, 'action': 'ALLOW', 'enabled': True,
'tenant_id': 'fake_tenant', }
FAKE_FW_RULE_V6 = {'ip_version': 6, 'protocol': 'udp',
'source_port_range_min': 1, 'source_port_range_max': 80,
'destination_port_range_min': 1,
'destination_port_range_max': 80,
'source_ip_address': 'fddf:cb3b:bc4::/48',
'destination_ip_address': 'fddf:cb3b:b33f::/48',
'position': 1, 'action': 'ALLOW', 'enabled': True,
'tenant_id': 'fake_tenant', }
FAKE_FW_V4 = {'name': 'fake firewall policy',
'tenant_id': uuidutils.generate_uuid(),
'description': 'this is fake',
'firewall_rules': [FAKE_FW_RULE_V4]}
FAKE_FW_V6 = {'name': 'fake firewall policy',
'tenant_id': uuidutils.generate_uuid(),
'description': 'this is fake',
'firewall_rules': [FAKE_FW_RULE_V6]}
class ClassifierTestContext(object):
"Classifier Database Context."
engine = None
session = None
def __init__(self):
self.engine = sa.create_engine('sqlite:///:memory:', echo=True)
self.session = sessionmaker(bind=self.engine)()
class DbApiTestCase(base.BaseTestCase):
def setUp(self):
super(DbApiTestCase, self).setUp()
self.context = ClassifierTestContext()
models.Base.metadata.create_all(self.context.engine)
def _create_classifier_group(self, service):
cg = models.ClassifierGroup()
cg.tenant_id = uuidutils.generate_uuid()
cg.name = 'test classifier'
cg.description = 'ensure all data inserted correctly'
cg.service = service
return cg
def test_create_classifier_chain(self):
cg = self._create_classifier_group('neutron-fwaas')
ipc = models.IpClassifier()
ipc.destination_ip_prefix = 'fd70:fbb6:449e::/48'
ipc.source_ip_prefix = 'fddf:cb3b:bc4::/48'
api.create_classifier_chain(cg, [ipc])
self.assertGreater(len(cg.classifier_chain), 0)
def _test_convert_security_group_rule_to_classifier(self,
security_group_rule):
# TODO(sc68cal) make this not call session.commit directly
cg = self._create_classifier_group('security-group')
api.convert_security_group_rule_to_classifier(self.context,
security_group_rule, cg)
# Save to the database
self.context.session.add(cg)
self.context.session.commit()
# Refresh the classifier group from the DB
cg = api.get_classifier_group(self.context, cg.id)
self.assertGreater(len(cg.classifier_chain), 0)
def test_convert_security_group_rule_v4_to_classifier(self):
self._test_convert_security_group_rule_to_classifier(FAKE_SG_RULE_V4)
def test_convert_security_group_rule_v6_to_classifier(self):
self._test_convert_security_group_rule_to_classifier(FAKE_SG_RULE_V6)
def test_convert_security_group_to_classifier_chain(self):
result = api.convert_security_group_to_classifier(self.context,
FAKE_SG_V6)
self.assertIsNotNone(result)
def test_convert_classifier_chain_to_security_group(self):
classifier_id = api.convert_security_group_to_classifier(
self.context, FAKE_SG_V6).id
result = api.convert_classifier_group_to_security_group(self.context,
classifier_id)
result['tenant_id'] = FAKE_SG_RULE_V6['tenant_id']
self.assertEqual(FAKE_SG_RULE_V6, result)
def _test_convert_sg_rule_to_classifier_exception(self, sg_rule):
try:
self._test_convert_security_group_rule_to_classifier(sg_rule)
except Exception:
pass
def test_convert_sg_rule_to_classifier_with_no_ethertype(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
del FAKE_SG_RULE['ethertype']
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# test case for invalid ip-version
def test_convert_sg_rule_to_classifier_with_invalid_ethertype(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
FAKE_SG_RULE['ethertype'] = 'IPvx'
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# test case for protocol none
def test_convert_sg_rule_to_classifier_with_None_protocol(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
del FAKE_SG_RULE['protocol']
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# can not allow icmpv6 protocol with IPv4 version
def test_convert_sg_rule_to_classifier_with_icmpv6_protocol(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
FAKE_SG_RULE['protocol'] = 'icmpv6'
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# ip-version is 4 and remote ip as v6 address
def test_convert_sg_rule_to_classifier_with_invalid_remote_ipv6(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
FAKE_SG_RULE['remote_ip_prefix'] = 'fddf:cb3b:bc4::/48'
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# ip-version is 6 and remote ip as v4 address
def test_convert_sg_rule_to_classifier_with_invalid_dest_ipv4(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V6)
FAKE_SG_RULE['remote_ip_prefix'] = '1.2.3.4/24'
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# invalid port-range
def test_convert_sg_rule_to_classifier_with_invalid_port_range(self):
FAKE_SG_RULE = cp.copy(FAKE_SG_RULE_V4)
FAKE_SG_RULE['port_range_min'] = 200
FAKE_SG_RULE['port_range_max'] = 10
self._test_convert_sg_rule_to_classifier_exception(FAKE_SG_RULE)
# Firewall testcases
def _test_convert_firewall_rule_to_classifier(self, fw_rule):
cg = self._create_classifier_group('neutron-fwaas')
api.convert_firewall_rule_to_classifier(self.context, fw_rule, cg)
# Save to the database
self.context.session.add(cg)
self.context.session.commit()
# Refresh the classifier group from the DB
cg = api.get_classifier_group(self.context, cg.id)
self.assertGreater(len(cg.classifier_chain), 0)
def test_convert_firewall_rule_v4_to_classifier(self):
self._test_convert_firewall_rule_to_classifier(FAKE_FW_RULE_V4)
def test_convert_firewall_rule_v6_to_classifier(self):
self._test_convert_firewall_rule_to_classifier(FAKE_FW_RULE_V6)
def test_convert_firewall_policy_v4_to_classifier_chain(self):
result = api.convert_firewall_policy_to_classifier(self.context,
FAKE_FW_V4)
self.assertIsNotNone(result)
def test_convert_firewall_policy_v6_to_classifier_chain(self):
result = api.convert_firewall_policy_to_classifier(self.context,
FAKE_FW_V6)
self.assertIsNotNone(result)
def test_convert_classifier_chain_to_firewall(self):
classifier_id = api.convert_firewall_policy_to_classifier(
self.context, FAKE_FW_V6).id
result = api.convert_classifier_to_firewall(self.context,
classifier_id)
result['tenant_id'] = FAKE_FW_RULE_V6['tenant_id']
result['position'] = FAKE_FW_RULE_V6['position']
result['action'] = FAKE_FW_RULE_V6['action']
result['enabled'] = FAKE_FW_RULE_V6['enabled']
self.assertEqual(FAKE_FW_RULE_V6, result)
def _test_convert_firewall_rule_to_classifier_exception(self, fw_rule):
try:
self._test_convert_firewall_rule_to_classifier(fw_rule)
except Exception:
pass
# test case for invalid ip-version
def test_convert_firewall_rule_to_classifier_with_invalid_ip_version(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V4)
FAKE_FW_RULE['ip_version'] = 5
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)
# test case for protocol none
def test_convert_firewall_rule_to_classifier_with_None_protocol(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V4)
del FAKE_FW_RULE['protocol']
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)
# icmp protocol with valid port range
def test_convert_firewall_rule_to_classifier_with_icmp_protocol(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V4)
FAKE_FW_RULE['protocol'] = 'icmp'
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)
# ip-version is 4 and source ip as v6 address
def test_convert_firewall_rule_to_classifier_with_invalid_source_ip(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V4)
FAKE_FW_RULE['source_ip_address'] = 'fddf:cb3b:bc4::/48'
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)
# ip-version is 6 and dest ip as v4 address
def test_convert_firewall_rule_to_classifier_with_invalid_dest_ip(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V6)
FAKE_FW_RULE['destination_ip_address'] = '1.2.3.4/24'
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)
# invalid port-range
def test_convert_firewall_rule_to_classifier_with_invalid_port_range(self):
FAKE_FW_RULE = cp.copy(FAKE_FW_RULE_V4)
FAKE_FW_RULE['source_port_range_min'] = 200
FAKE_FW_RULE['source_port_range_max'] = 10
FAKE_FW_RULE['destination_port_range_min'] = 100
FAKE_FW_RULE['destination_port_range_max'] = 10
self._test_convert_firewall_rule_to_classifier_exception(FAKE_FW_RULE)

1
test-requirements.txt
View File

@@ -6,6 +6,7 @@ oslosphinx>=4.7.0 # Apache-2.0
oslotest>=1.10.0 # Apache-2.0
os-testr>=1.0.0 # Apache-2.0
testrepository>=0.0.18 # Apache-2.0/BSD
testresources>=0.2.4 # Apache-2.0/BSD
testscenarios>=0.4 # Apache-2.0/BSD
testtools>=1.4.0 # MIT
pylint==1.4.5 # GPLv2