build.match._match_ipv4_10() - Code Metrics - Inspection of "Add support for non-strict exclusion in the consis..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#117)

by Carlos

created 2021-04-06 17:33 UTC

build.match._match_ipv4_10() F

↳ Parent: build.match

Complexity

Conditions

Size

Total Lines	51
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	563.3102

Importance

Changes

Metric	Value
cc	24
eloc	50
nop	3
dl	0
loc	51
ccs	1
cts	46
cp	0.0217
crap	563.3102
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

"""Switch match."""

import ipaddress

from pyof.v0x01.common.flow_match import FlowWildCards

IPV4_ETH_TYPE = 2048


def format_request(request):
    """Format user request to match function format."""
    args = {}
    for key, value in request.items():
        args[key] = value
    return args


def do_match(flow_to_install, version, stored_flow_dict):
    """Match a packet against this flow."""
    if version == 0x01:
        return match10(stored_flow_dict, flow_to_install)
    elif version == 0x04:
        return match13_no_strict(flow_to_install, stored_flow_dict)
    raise NotImplementedError(f'Unsupported OpenFlow version {version}')


def _get_match_fields(flow_dict):
    """Generate match fields."""
    match_fields = {}
    if 'match' in flow_dict:
        for key, value in flow_dict['match'].items():
            match_fields[key] = value
    return match_fields


# pylint: disable=too-many-return-statements, too-many-statements, R0912
def _match_ipv4_10(match_fields, args, wildcards):
    """Match IPV4 fields against packet with Flow (OF1.0)."""
    if not match_fields['eth_type'] == IPV4_ETH_TYPE:
        return False
    flow_ip_int = int(ipaddress.IPv4Address(match_fields.get('ipv4_src')))
    if flow_ip_int != 0:
        mask = (wildcards
                & FlowWildCards.OFPFW_NW_SRC_MASK) >> \
                FlowWildCards.OFPFW_NW_SRC_SHIFT
        if mask > 32:
            mask = 32
        if mask != 32 and 'ipv4_src' not in args:
            return False
        mask = (0xffffffff << mask) & 0xffffffff
        ip_int = int(ipaddress.IPv4Address(args.get('ipv4_src')))
        if ip_int & mask != flow_ip_int & mask:
            return False
    flow_ip_int = int(ipaddress.IPv4Address(match_fields['ipv4_dst']))
    if flow_ip_int != 0:
        mask = (wildcards
                & FlowWildCards.OFPFW_NW_DST_MASK) >> \
                FlowWildCards.OFPFW_NW_DST_SHIFT
        if mask > 32:
            mask = 32
        if mask != 32 and 'ipv4_dst' not in args:
            return False
        mask = (0xffffffff << mask) & 0xffffffff
        ip_int = int(ipaddress.IPv4Address(args.get('ipv4_dst')))
        if ip_int & mask != flow_ip_int & mask:
            return False
    if not wildcards & FlowWildCards.OFPFW_NW_TOS:
        if 'ip_tos' not in args:
            return False
        if match_fields.get('ip_tos') != int(args.get('ip_tos')):
            return False
    if not wildcards & FlowWildCards.OFPFW_NW_PROTO:
        if 'ip_proto' not in args:
            return False
        if match_fields.get('ip_proto') != int(args.get('ip_proto')):
            return False
    if not wildcards & FlowWildCards.OFPFW_TP_SRC:
        if 'tp_src' not in args:
            return False
        if match_fields.get('tcp_src') != int(args.get('tp_src')):
            return False
    if not wildcards & FlowWildCards.OFPFW_TP_DST:
        if 'tp_dst' not in args:
            return False
        if match_fields.get('tcp_dst') != int(args.get('tp_dst')):
            return False
    return True


# pylint: disable=too-many-return-statements, too-many-statements, R0912
def match10(flow_dict, args):
    """Match a packet against this flow (OF1.0)."""
    match_fields = _get_match_fields(flow_dict)
    wildcards = match_fields.get('wildcards')
    if not wildcards & FlowWildCards.OFPFW_IN_PORT:
        if 'in_port' not in args:
            return False
        if match_fields.get('in_port') != int(args.get('in_port')):
            return False
    if not wildcards & FlowWildCards.OFPFW_DL_VLAN_PCP:
        if 'vlan_pcp' not in args:
            return False
        if match_fields.get('vlan_pcp') != int(args.get('vlan_pcp')):
            return False
    if not wildcards & FlowWildCards.OFPFW_DL_VLAN:
        if 'vlan_vid' not in args:
            return False
        if match_fields.get('vlan_vid') != args.get('vlan_vid')[-1]:
            return False
    if not wildcards & FlowWildCards.OFPFW_DL_SRC:
        if 'eth_src' not in args:
            return False
        if match_fields.get('eth_src') != args.get('eth_src'):
            return False
    if not wildcards & FlowWildCards.OFPFW_DL_DST:
        if 'eth_dst' not in args:
            return False
        if match_fields.get('eth_dst') != args.get('eth_dst'):
            return False
    if not wildcards & FlowWildCards.OFPFW_DL_TYPE:
        if 'eth_type' not in args:
            return False
        if match_fields.get('eth_type') != int(args.get('eth_type')):
            return False
    if not _match_ipv4_10(match_fields, args, wildcards):
        return False
    return flow_dict


def match13_no_strict(flow_to_install, stored_flow_dict):
    """Match a packet againts the stored flow (OF 1.3).

    Return the flow if any fields match, otherwise, return False.
    """
    if flow_to_install.get('cookie_mask') and 'cookie' in stored_flow_dict:
        cookie = flow_to_install['cookie_mask'] & flow_to_install['cookie']
        if cookie == stored_flow_dict['cookie']:
            return stored_flow_dict
        return False

    if 'match' not in flow_to_install:
        return False

    for key, value in flow_to_install.get('match').items():
        if 'match' not in stored_flow_dict:
            return False
        if key not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
            if value == stored_flow_dict['match'].get(key):
                return stored_flow_dict
        else:
            field = flow_to_install.get(key)
            packet_ip = int(ipaddress.ip_address(field))
            ip_addr = value
            if packet_ip & ip_addr.netmask == ip_addr.address:
                return stored_flow_dict

    return False


def match_flows(flow_to_install, version, stored_flow_dict):
    # pylint: disable=bad-staticmethod-argument
    """
    Match the packet in request against the flows installed in the switch.

    Try the match with each flow, in other. If many is True, tries the
    match with all flows, if False, tries until the first match.
    :param args: packet data
    :param many: Boolean, indicating whether to continue after matching the
            first flow or not
    :return: If many, the list of matched flows, or the matched flow
    """
    match = do_match(flow_to_install, version, stored_flow_dict)
    return match


1		"""Switch match."""
2
3	1	import ipaddress
4
5	1	from pyof.v0x01.common.flow_match import FlowWildCards
6
7	1	IPV4_ETH_TYPE = 2048
8
9
10	1	def format_request(request):
11		"""Format user request to match function format."""
12		args = {}
13		for key, value in request.items():
14		args[key] = value
15		return args
16
17
18	1	def do_match(flow_to_install, version, stored_flow_dict):
19		"""Match a packet against this flow."""
20		if version == 0x01:
21		return match10(stored_flow_dict, flow_to_install)
22		elif version == 0x04:
23		return match13_no_strict(flow_to_install, stored_flow_dict)
24		raise NotImplementedError(f'Unsupported OpenFlow version {version}')
25
26
27	1	def _get_match_fields(flow_dict):
28		"""Generate match fields."""
29		match_fields = {}
30		if 'match' in flow_dict:
31		for key, value in flow_dict['match'].items():
32		match_fields[key] = value
33		return match_fields
34
35
36		# pylint: disable=too-many-return-statements, too-many-statements, R0912
37	1	def _match_ipv4_10(match_fields, args, wildcards):
38		"""Match IPV4 fields against packet with Flow (OF1.0)."""
39		if not match_fields['eth_type'] == IPV4_ETH_TYPE:
40		return False
41		flow_ip_int = int(ipaddress.IPv4Address(match_fields.get('ipv4_src')))
42		if flow_ip_int != 0:
43		mask = (wildcards
44		& FlowWildCards.OFPFW_NW_SRC_MASK) >> \
45		FlowWildCards.OFPFW_NW_SRC_SHIFT
46		if mask > 32:
47		mask = 32
48		if mask != 32 and 'ipv4_src' not in args:
49		return False
50		mask = (0xffffffff << mask) & 0xffffffff
51		ip_int = int(ipaddress.IPv4Address(args.get('ipv4_src')))
52		if ip_int & mask != flow_ip_int & mask:
53		return False
54		flow_ip_int = int(ipaddress.IPv4Address(match_fields['ipv4_dst']))
55		if flow_ip_int != 0:
56		mask = (wildcards
57		& FlowWildCards.OFPFW_NW_DST_MASK) >> \
58		FlowWildCards.OFPFW_NW_DST_SHIFT
59		if mask > 32:
60		mask = 32
61		if mask != 32 and 'ipv4_dst' not in args:
62		return False
63		mask = (0xffffffff << mask) & 0xffffffff
64		ip_int = int(ipaddress.IPv4Address(args.get('ipv4_dst')))
65		if ip_int & mask != flow_ip_int & mask:
66		return False
67		if not wildcards & FlowWildCards.OFPFW_NW_TOS:
68		if 'ip_tos' not in args:
69		return False
70		if match_fields.get('ip_tos') != int(args.get('ip_tos')):
71		return False
72		if not wildcards & FlowWildCards.OFPFW_NW_PROTO:
73		if 'ip_proto' not in args:
74		return False
75		if match_fields.get('ip_proto') != int(args.get('ip_proto')):
76		return False
77		if not wildcards & FlowWildCards.OFPFW_TP_SRC:
78		if 'tp_src' not in args:
79		return False
80		if match_fields.get('tcp_src') != int(args.get('tp_src')):
81		return False
82		if not wildcards & FlowWildCards.OFPFW_TP_DST:
83		if 'tp_dst' not in args:
84		return False
85		if match_fields.get('tcp_dst') != int(args.get('tp_dst')):
86		return False
87		return True
88
89
90		# pylint: disable=too-many-return-statements, too-many-statements, R0912
91	1	def match10(flow_dict, args):
92		"""Match a packet against this flow (OF1.0)."""
93		match_fields = _get_match_fields(flow_dict)
94		wildcards = match_fields.get('wildcards')
95		if not wildcards & FlowWildCards.OFPFW_IN_PORT:
96		if 'in_port' not in args:
97		return False
98		if match_fields.get('in_port') != int(args.get('in_port')):
99		return False
100		if not wildcards & FlowWildCards.OFPFW_DL_VLAN_PCP:
101		if 'vlan_pcp' not in args:
102		return False
103		if match_fields.get('vlan_pcp') != int(args.get('vlan_pcp')):
104		return False
105		if not wildcards & FlowWildCards.OFPFW_DL_VLAN:
106		if 'vlan_vid' not in args:
107		return False
108		if match_fields.get('vlan_vid') != args.get('vlan_vid')[-1]:
109		return False
110		if not wildcards & FlowWildCards.OFPFW_DL_SRC:
111		if 'eth_src' not in args:
112		return False
113		if match_fields.get('eth_src') != args.get('eth_src'):
114		return False
115		if not wildcards & FlowWildCards.OFPFW_DL_DST:
116		if 'eth_dst' not in args:
117		return False
118		if match_fields.get('eth_dst') != args.get('eth_dst'):
119		return False
120		if not wildcards & FlowWildCards.OFPFW_DL_TYPE:
121		if 'eth_type' not in args:
122		return False
123		if match_fields.get('eth_type') != int(args.get('eth_type')):
124		return False
125		if not _match_ipv4_10(match_fields, args, wildcards):
126		return False
127		return flow_dict
128
129
130	1	def match13_no_strict(flow_to_install, stored_flow_dict):
131		"""Match a packet againts the stored flow (OF 1.3).
132
133		Return the flow if any fields match, otherwise, return False.
134		"""
135		if flow_to_install.get('cookie_mask') and 'cookie' in stored_flow_dict:
136		cookie = flow_to_install['cookie_mask'] & flow_to_install['cookie']
137		if cookie == stored_flow_dict['cookie']:
138		return stored_flow_dict
139		return False
140
141		if 'match' not in flow_to_install:
142		return False
143
144		for key, value in flow_to_install.get('match').items():
145		if 'match' not in stored_flow_dict:
146		return False
147		if key not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
148		if value == stored_flow_dict['match'].get(key):
149		return stored_flow_dict
150		else:
151		field = flow_to_install.get(key)
152		packet_ip = int(ipaddress.ip_address(field))
153		ip_addr = value
154		if packet_ip & ip_addr.netmask == ip_addr.address:
155		return stored_flow_dict
156
157		return False
158
159
160	1	def match_flows(flow_to_install, version, stored_flow_dict):
161		# pylint: disable=bad-staticmethod-argument
162		"""
163		Match the packet in request against the flows installed in the switch.
164
165		Try the match with each flow, in other. If many is True, tries the
166		match with all flows, if False, tries until the first match.
167		:param args: packet data
168		:param many: Boolean, indicating whether to continue after matching the
169		first flow or not
170		:return: If many, the list of matched flows, or the matched flow
171		"""
172		match = do_match(flow_to_install, version, stored_flow_dict)
173		return match
174

kytos / flow_manager

Pull Request — master (#117)

build.match._match_ipv4_10() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like