openstack
/
neutron
Code Issues Proposed changes

ARP spoofing patch: Low level ebtables integration 

ARP cache poisoning is not actually prevented by the firewall
driver 'iptables_firewall'. We are adding the use of the ebtables
command - with a corresponding ebtables-driver - in order to create
Ethernet frame filtering rules, which prevent the sending of ARP
cache poisoning frames.

The complete patch is broken into a set of smaller patches for easier review.

This patch here is th first of the series and includes the low-level ebtables
integration, unit and functional tests.

Note:
    This commit is based greatly on an original, now abandoned patch,
    presented for review here:

        https://review.openstack.org/#/c/70067/

    Full spec can be found here:

        https://review.openstack.org/#/c/129090/

SecurityImpact

Change-Id: I9ef57a86b1a1c1fa4ba1a034c920f23cb40072c0
Implements: blueprint arp-spoof-patch-ebtables
Related-Bug: 1274034
Co-Authored-By: jbrendel <jbrendel@cisco.com>
This commit is contained in:
Édouard Thuleau 2015-02-10 13:43:34 +13:00 committed by Juergen Brendel
parent 76d873a452
commit 2414834ffe
7 changed files with 653 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
etc/neutron/rootwrap.d/ebtables.filters Normal file
View File

@ -0,0 +1,13 @@
# neutron-rootwrap command filters for nodes on which neutron is
# expected to control network
#
# This file should be owned by (and only-writeable by) the root user
# format seems to be
# cmd-name: filter-name, raw-command, user, args
[Filters]
# neutron/agent/linux/ebtables_driver.py
ebtables: CommandFilter, ebtables, root
ebtablesEnv: EnvFilter, ebtables, root, EBTABLES_ATOMIC_FILE=

290
neutron/agent/linux/ebtables_driver.py Normal file
View File

@ -0,0 +1,290 @@
# Copyright (c) 2015 OpenStack Foundation.
# 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.
#
"""Implement ebtables rules using linux utilities."""
import re
from retrying import retry
from oslo_config import cfg
from oslo_log import log as logging
from neutron.common import utils
ebtables_opts = [
cfg.StrOpt('ebtables_path',
default='$state_path/ebtables-',
help=_('Location of temporary ebtables table files.')),
]
CONF = cfg.CONF
CONF.register_opts(ebtables_opts)
LOG = logging.getLogger(__name__)
# Collection of regexes to parse ebtables output
_RE_FIND_BRIDGE_TABLE_NAME = re.compile(r'^Bridge table:[\s]*([a-z]+)$')
# get chain name, nunmber of entries and policy name.
_RE_FIND_BRIDGE_CHAIN_INFO = re.compile(
r'^Bridge chain:[\s]*(.*),[\s]*entries:[\s]*[0-9]+,[\s]*'
r'policy:[\s]*([A-Z]+)$')
_RE_FIND_BRIDGE_RULE_COUNTERS = re.compile(
r',[\s]*pcnt[\s]*=[\s]*([0-9]+)[\s]*--[\s]*bcnt[\s]*=[\s]*([0-9]+)$')
_RE_FIND_COMMIT_STATEMENT = re.compile(r'^COMMIT$')
_RE_FIND_COMMENTS_AND_BLANKS = re.compile(r'^#|^$')
_RE_FIND_APPEND_RULE = re.compile(r'-A (\S+) ')
# Regexes to parse ebtables rule file input
_RE_RULES_FIND_TABLE_NAME = re.compile(r'^\*([a-z]+)$')
_RE_RULES_FIND_CHAIN_NAME = re.compile(r'^:(.*)[\s]+([A-Z]+)$')
_RE_RULES_FIND_RULE_LINE = re.compile(r'^\[([0-9]+):([0-9]+)\]')
def _process_ebtables_output(lines):
"""Process raw output of ebtables rule listing file.
Empty lines and comments removed, ebtables listing output converted
into ebtables rules.
For example, if the raw ebtables list lines (input to this function) are:
Bridge table: filter
Bridge chain: INPUT, entries: 0, policy: ACCEPT
Bridge chain: FORWARD, entries: 0, policy: ACCEPT
Bridge chain: OUTPUT, entries: 0, policy: ACCEPT
The output then will be:
*filter
:INPUT ACCEPT
:FORWARD ACCEPT
:OUTPUT ACCEPT
COMMIT
Key point: ebtables rules listing output is not the same as the rules
format for setting new rules.
"""
table = None
chain = ''
chains = []
rules = []
for line in lines:
if _RE_FIND_COMMENTS_AND_BLANKS.search(line):
continue
match = _RE_FIND_BRIDGE_RULE_COUNTERS.search(line)
if table and match:
rules.append('[%s:%s] -A %s %s' % (match.group(1),
match.group(2),
chain,
line[:match.start()].strip()))
match = _RE_FIND_BRIDGE_CHAIN_INFO.search(line)
if match:
chains.append(':%s %s' % (match.group(1), match.group(2)))
chain = match.group(1)
continue
match = _RE_FIND_BRIDGE_TABLE_NAME.search(line)
if match:
table = '*%s' % match.group(1)
continue
return [table] + chains + rules + ['COMMIT']
def _match_rule_line(table, line):
match = _RE_RULES_FIND_RULE_LINE.search(line)
if table and match:
args = line[match.end():].split()
res = [(table, args)]
if int(match.group(1)) > 0 and int(match.group(2)) > 0:
p = _RE_FIND_APPEND_RULE
rule = p.sub(r'-C \1 %s %s ', line[match.end() + 1:])
args = (rule % (match.group(1), match.group(2))).split()
res.append((table, args))
return table, res
else:
return table, None
def _match_chain_name(table, tables, line):
match = _RE_RULES_FIND_CHAIN_NAME.search(line)
if table and match:
if match.group(1) not in tables[table]:
args = ['-N', match.group(1), '-P', match.group(2)]
else:
args = ['-P', match.group(1), match.group(2)]
return table, (table, args)
else:
return table, None
def _match_table_name(table, line):
match = _RE_RULES_FIND_TABLE_NAME.search(line)
if match:
# Initialize with current kernel table if we just start out
table = match.group(1)
return table, (table, ['--atomic-init'])
else:
return table, None
def _match_commit_statement(table, line):
match = _RE_FIND_COMMIT_STATEMENT.search(line)
if table and match:
# Conclude by issuing the commit command
return (table, ['--atomic-commit'])
else:
return None
def _process_ebtables_input(lines):
"""Import text ebtables rules. Similar to iptables-restore.
Was based on:
http://sourceforge.net/p/ebtables/code/ci/
3730ceb7c0a81781679321bfbf9eaa39cfcfb04e/tree/userspace/ebtables2/
ebtables-save?format=raw
The function prepares and returns a list of tuples, each tuple consisting
of a table name and ebtables arguments. The caller can then repeatedly call
ebtables on that table with those arguments to get the rules applied.
For example, this input:
*filter
:INPUT ACCEPT
:FORWARD ACCEPT
:OUTPUT ACCEPT
:neutron-nwfilter-spoofing-fallb ACCEPT
:neutron-nwfilter-OUTPUT ACCEPT
:neutron-nwfilter-INPUT ACCEPT
:neutron-nwfilter-FORWARD ACCEPT
[0:0] -A INPUT -j neutron-nwfilter-INPUT
[0:0] -A OUTPUT -j neutron-nwfilter-OUTPUT
[0:0] -A FORWARD -j neutron-nwfilter-FORWARD
[0:0] -A neutron-nwfilter-spoofing-fallb -j DROP
COMMIT
... produces this output:
('filter', ['--atomic-init'])
('filter', ['-P', 'INPUT', 'ACCEPT'])
('filter', ['-P', 'FORWARD', 'ACCEPT'])
('filter', ['-P', 'OUTPUT', 'ACCEPT'])
('filter', ['-N', 'neutron-nwfilter-spoofing-fallb', '-P', 'ACCEPT'])
('filter', ['-N', 'neutron-nwfilter-OUTPUT', '-P', 'ACCEPT'])
('filter', ['-N', 'neutron-nwfilter-INPUT', '-P', 'ACCEPT'])
('filter', ['-N', 'neutron-nwfilter-FORWARD', '-P', 'ACCEPT'])
('filter', ['-A', 'INPUT', '-j', 'neutron-nwfilter-INPUT'])
('filter', ['-A', 'OUTPUT', '-j', 'neutron-nwfilter-OUTPUT'])
('filter', ['-A', 'FORWARD', '-j', 'neutron-nwfilter-FORWARD'])
('filter', ['-A', 'neutron-nwfilter-spoofing-fallb', '-j', 'DROP'])
('filter', ['--atomic-commit'])
"""
tables = {'filter': ['INPUT', 'FORWARD', 'OUTPUT'],
'nat': ['PREROUTING', 'OUTPUT', 'POSTROUTING'],
'broute': ['BROUTING']}
table = None
ebtables_args = list()
for line in lines.splitlines():
if _RE_FIND_COMMENTS_AND_BLANKS.search(line):
continue
table, res = _match_rule_line(table, line)
if res:
ebtables_args.extend(res)
continue
table, res = _match_chain_name(table, tables, line)
if res:
ebtables_args.append(res)
continue
table, res = _match_table_name(table, line)
if res:
ebtables_args.append(res)
continue
res = _match_commit_statement(table, line)
if res:
ebtables_args.append(res)
continue
return ebtables_args
@retry(wait_exponential_multiplier=1000, wait_exponential_max=10000,
stop_max_delay=10000)
def _cmd_retry(func, *args, **kwargs):
return func(*args, **kwargs)
def run_ebtables(namespace, execute, table, args):
"""Run ebtables utility, with retry if necessary.
Provide table name and list of additional arguments to ebtables.
"""
cmd = ['ebtables', '-t', table]
if CONF.ebtables_path:
f = '%s%s' % (CONF.ebtables_path, table)
cmd += ['--atomic-file', f]
cmd += args
if namespace:
cmd = ['ip', 'netns', 'exec', namespace] + cmd
# TODO(jbrendel): The root helper is used for every ebtables command,
# but as we use an atomic file we only need root for
# init and commit commands.
# But the generated file by init ebtables command is
# only readable and writable by root.
#
# We retry the execution of ebtables in case of failure. Known issue:
# See bug: https://bugs.launchpad.net/nova/+bug/1316621
# See patch: https://review.openstack.org/#/c/140514/3
return _cmd_retry(execute, cmd, **{"run_as_root": True})
def run_ebtables_multiple(namespace, execute, arg_list):
"""Run ebtables utility multiple times.
Similar to run(), but runs ebtables for every element in arg_list.
Each arg_list element is a tuple containing the table name and a list
of ebtables arguments.
"""
for table, args in arg_list:
run_ebtables(namespace, execute, table, args)
@utils.synchronized('ebtables', external=True)
def ebtables_save(execute, tables_names, namespace=None):
"""Generate text output of the ebtables rules.
Based on:
http://sourceforge.net/p/ebtables/code/ci/master/tree/userspace/ebtables2/
ebtables-save?format=raw
"""
raw_outputs = (run_ebtables(namespace, execute,
t, ['-L', '--Lc']).splitlines() for t in tables_names)
parsed_outputs = (_process_ebtables_output(lines) for lines in raw_outputs)
return '\n'.join(l for lines in parsed_outputs for l in lines)
@utils.synchronized('ebtables', external=True)
def ebtables_restore(lines, execute, namespace=None):
"""Import text ebtables rules and apply."""
ebtables_args = _process_ebtables_input(lines)
run_ebtables_multiple(namespace, execute, ebtables_args)

11
neutron/cmd/sanity/checks.py
View File

@ -179,3 +179,14 @@ def ovsdb_native_supported():
LOG.exception(six.text_type(ex))
return False
def ebtables_supported():
try:
cmd = ['ebtables', '--version']
agent_utils.execute(cmd)
return True
except (OSError, RuntimeError, IndexError, ValueError) as e:
LOG.debug("Exception while checking for installed ebtables. "
"Exception: %s", e)
return False

12
neutron/cmd/sanity_check.py
View File

@ -146,7 +146,15 @@ def check_ovsdb_native():
return result
# Define CLI opts to test specific features, with a calback for the test
def check_ebtables():
result = checks.ebtables_supported()
if not result:
LOG.error(_LE('Cannot run ebtables. Please ensure that it '
'is installed.'))
return result
# Define CLI opts to test specific features, with a callback for the test
OPTS = [
BoolOptCallback('ovs_vxlan', check_ovs_vxlan, default=False,
help=_('Check for OVS vxlan support')),
@ -168,6 +176,8 @@ OPTS = [
help=_('Check minimal dnsmasq version')),
BoolOptCallback('ovsdb_native', check_ovsdb_native,
help=_('Check ovsdb native interface support')),
BoolOptCallback('ebtables_installed', check_ebtables,
help=_('Check ebtables installation')),
]

136
neutron/tests/functional/agent/linux/test_ebtables_driver.py Normal file
View File

@ -0,0 +1,136 @@
# Copyright (c) 2015 OpenStack Foundation.
# 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.
from neutron.agent.linux import ebtables_driver
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils as linux_utils
from neutron.tests.functional.agent.linux import base
from neutron.tests.functional.agent.linux import helpers
NO_FILTER_APPLY = (
"*filter\n"
":INPUT ACCEPT\n"
":FORWARD ACCEPT\n"
":OUTPUT ACCEPT\n"
":neutron-nwfilter-OUTPUT ACCEPT\n"
":neutron-nwfilter-INPUT ACCEPT\n"
":neutron-nwfilter-FORWARD ACCEPT\n"
":neutron-nwfilter-spoofing-fallb ACCEPT\n"
"[0:0] -A INPUT -j neutron-nwfilter-INPUT\n"
"[0:0] -A FORWARD -j neutron-nwfilter-FORWARD\n"
"[2:140] -A OUTPUT -j neutron-nwfilter-OUTPUT\n"
"[0:0] -A neutron-nwfilter-spoofing-fallb -j DROP\n"
"COMMIT")
FILTER_APPLY_TEMPLATE = (
"*filter\n"
":INPUT ACCEPT\n"
":FORWARD ACCEPT\n"
":OUTPUT ACCEPT\n"
":neutron-nwfilter-OUTPUT ACCEPT\n"
":neutron-nwfilter-isome-port-id ACCEPT\n"
":neutron-nwfilter-i-arp-some-por ACCEPT\n"
":neutron-nwfilter-i-ip-some-port ACCEPT\n"
":neutron-nwfilter-spoofing-fallb ACCEPT\n"
":neutron-nwfilter-INPUT ACCEPT\n"
":neutron-nwfilter-FORWARD ACCEPT\n"
"[0:0] -A neutron-nwfilter-OUTPUT -j neutron-nwfilter-isome-port-id\n"
"[0:0] -A INPUT -j neutron-nwfilter-INPUT\n"
"[2:140] -A OUTPUT -j neutron-nwfilter-OUTPUT\n"
"[0:0] -A FORWARD -j neutron-nwfilter-FORWARD\n"
"[0:0] -A neutron-nwfilter-spoofing-fallb -j DROP\n"
"[0:0] -A neutron-nwfilter-i-arp-some-por "
"-p arp --arp-opcode 2 --arp-mac-src %(mac_addr)s "
"--arp-ip-src %(ip_addr)s -j RETURN\n"
"[0:0] -A neutron-nwfilter-i-arp-some-por -p ARP --arp-op Request "
"-j ACCEPT\n"
"[0:0] -A neutron-nwfilter-i-arp-some-por "
"-j neutron-nwfilter-spoofing-fallb\n"
"[0:0] -A neutron-nwfilter-isome-port-id "
"-p arp -j neutron-nwfilter-i-arp-some-por\n"
"[0:0] -A neutron-nwfilter-i-ip-some-port "
"-s %(mac_addr)s -p IPv4 --ip-source %(ip_addr)s -j RETURN\n"
"[0:0] -A neutron-nwfilter-i-ip-some-port "
"-j neutron-nwfilter-spoofing-fallb\n"
"[0:0] -A neutron-nwfilter-isome-port-id "
"-p IPv4 -j neutron-nwfilter-i-ip-some-port\n"
"COMMIT")
class EbtablesLowLevelTestCase(base.BaseIPVethTestCase):
def setUp(self):
super(EbtablesLowLevelTestCase, self).setUp()
self.src_ns, self.dst_ns = self.prepare_veth_pairs()
devs = [d for d in self.src_ns.get_devices() if d.name != "lo"]
src_dev_name = devs[0].name
self.ns = self.src_ns.namespace
self.execute = linux_utils.execute
self.pinger = helpers.Pinger(self.src_ns)
# Extract MAC and IP address of one of my interfaces
self.mac = self.src_ns.device(src_dev_name).link.address
addr = [a for a in
self.src_ns.device(src_dev_name).addr.list()][0]['cidr']
self.addr = addr.split("/")[0]
# Pick one of the namespaces and setup a bridge for the local ethernet
# interface there, because ebtables only works on bridged interfaces.
self.src_ns.netns.execute("brctl addbr mybridge".split())
self.src_ns.netns.execute(("brctl addif mybridge %s" % src_dev_name).
split())
# Take the IP addrss off one of the interfaces and apply it to the
# bridge interface instead.
dev_source = ip_lib.IPDevice(src_dev_name, self.src_ns.namespace)
dev_mybridge = ip_lib.IPDevice("mybridge", self.src_ns.namespace)
dev_source.addr.delete(addr)
dev_mybridge.link.set_up()
dev_mybridge.addr.add(addr)
def _test_basic_port_filter_wrong_mac(self):
# Setup filter with wrong IP/MAC address pair. Basic filters only allow
# packets with specified address combinations, thus all packets will
# be dropped.
mac_ip_pair = dict(mac_addr="11:11:11:22:22:22", ip_addr=self.addr)
filter_apply = FILTER_APPLY_TEMPLATE % mac_ip_pair
ebtables_driver.ebtables_restore(filter_apply,
self.execute,
self.ns)
self.pinger.assert_no_ping(self.DST_ADDRESS)
# Assure that ping will work once we unfilter the instance
ebtables_driver.ebtables_restore(NO_FILTER_APPLY,
self.execute,
self.ns)
self.pinger.assert_ping(self.DST_ADDRESS)
def _test_basic_port_filter_correct_mac(self):
# Use the correct IP/MAC address pair for this one.
mac_ip_pair = dict(mac_addr=self.mac, ip_addr=self.addr)
filter_apply = FILTER_APPLY_TEMPLATE % mac_ip_pair
ebtables_driver.ebtables_restore(filter_apply,
self.execute,
self.ns)
self.pinger.assert_ping(self.DST_ADDRESS)
def test_ebtables_filtering(self):
# Cannot parallelize those tests. Therefore need to call them
# in order from a single function.
self._test_basic_port_filter_wrong_mac()
self._test_basic_port_filter_correct_mac()

191
neutron/tests/unit/agent/linux/test_ebtables_driver.py Normal file
View File

@ -0,0 +1,191 @@
# Copyright (c) 2015 OpenStack Foundation.
# 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.
#
import mock
from oslo_config import cfg
from neutron.agent.linux import ebtables_driver as eb
from neutron.cmd.sanity.checks import ebtables_supported
from neutron.tests import base
TABLES_NAMES = ['filter', 'nat', 'broute']
CONF = cfg.CONF
class EbtablesDriverLowLevelInputTestCase(base.BaseTestCase):
def test_match_rule_line(self):
self.assertEqual((None, None), eb._match_rule_line(None, "foo"))
rule_line = "[0:1] foobar blah bar"
self.assertEqual(('mytab', [('mytab', ['foobar', 'blah', 'bar'])]),
eb._match_rule_line("mytab", rule_line))
rule_line = "[2:3] foobar -A BAR -j BLAH"
self.assertEqual(
('mytab',
[('mytab', ['foobar', '-A', 'BAR', '-j', 'BLAH']),
('mytab', ['foobar', '-C', 'BAR', '2', '3', '-j', 'BLAH'])]),
eb._match_rule_line("mytab", rule_line))
def test_match_chain_name(self):
self.assertEqual((None, None), eb._match_chain_name(None, None, "foo"))
rule_line = ":neutron-nwfilter-OUTPUT ACCEPT"
tables = {"mytab": []}
self.assertEqual(
('mytab',
('mytab', ['-N', 'neutron-nwfilter-OUTPUT', '-P', 'ACCEPT'])),
eb._match_chain_name("mytab", tables, rule_line))
rule_line = ":neutron-nwfilter-OUTPUT ACCEPT"
tables = {"mytab": ['neutron-nwfilter-OUTPUT']}
self.assertEqual(
('mytab',
('mytab', ['-P', 'neutron-nwfilter-OUTPUT', 'ACCEPT'])),
eb._match_chain_name("mytab", tables, rule_line))
def test_match_table_name(self):
self.assertEqual((None, None), eb._match_table_name(None, "foo"))
rule_line = "*filter"
self.assertEqual(('filter', ('filter', ['--atomic-init'])),
eb._match_table_name("mytab", rule_line))
def test_commit_statement(self):
self.assertEqual(None, eb._match_commit_statement(None, "foo"))
rule_line = "COMMIT"
self.assertEqual(('mytab', ['--atomic-commit']),
eb._match_commit_statement("mytab", rule_line))
def test_ebtables_input_parse_comment(self):
# Comments and empty lines are stripped, nothing should be left.
test_input = ("# Here is a comment\n"
"\n"
"# We just had an empty line.\n")
res = eb._process_ebtables_input(test_input)
self.assertEqual(list(), res)
def test_ebtables_input_parse_start(self):
# Starting
test_input = "*filter"
res = eb._process_ebtables_input(test_input)
self.assertEqual([('filter', ['--atomic-init'])], res)
def test_ebtables_input_parse_commit(self):
# COMMIT without first starting a table should result in nothing,
test_input = "COMMIT"
res = eb._process_ebtables_input(test_input)
self.assertEqual(list(), res)
test_input = "*filter\nCOMMIT"
res = eb._process_ebtables_input(test_input)
self.assertEqual([('filter', ['--atomic-init']),
('filter', ['--atomic-commit'])],
res)
def test_ebtables_input_parse_rule(self):
test_input = "*filter\n[0:0] -A INPUT -j neutron-nwfilter-INPUT"
res = eb._process_ebtables_input(test_input)
self.assertEqual([('filter', ['--atomic-init']),
('filter',
['-A', 'INPUT', '-j', 'neutron-nwfilter-INPUT'])],
res)
def test_ebtables_input_parse_chain(self):
test_input = "*filter\n:foobar ACCEPT"
res = eb._process_ebtables_input(test_input)
self.assertEqual([('filter', ['--atomic-init']),
('filter', ['-N', 'foobar', '-P', 'ACCEPT'])],
res)
def test_ebtables_input_parse_all_together(self):
test_input = \
("*filter\n"
":INPUT ACCEPT\n"
":FORWARD ACCEPT\n"
":OUTPUT ACCEPT\n"
":neutron-nwfilter-spoofing-fallb ACCEPT\n"
":neutron-nwfilter-OUTPUT ACCEPT\n"
":neutron-nwfilter-INPUT ACCEPT\n"
":neutron-nwfilter-FORWARD ACCEPT\n"
"[0:0] -A INPUT -j neutron-nwfilter-INPUT\n"
"[0:0] -A OUTPUT -j neutron-nwfilter-OUTPUT\n"
"[0:0] -A FORWARD -j neutron-nwfilter-FORWARD\n"
"[0:0] -A neutron-nwfilter-spoofing-fallb -j DROP\n"
"COMMIT")
observed_res = eb._process_ebtables_input(test_input)
TNAME = 'filter'
expected_res = [
(TNAME, ['--atomic-init']),
(TNAME, ['-P', 'INPUT', 'ACCEPT']),
(TNAME, ['-P', 'FORWARD', 'ACCEPT']),
(TNAME, ['-P', 'OUTPUT', 'ACCEPT']),
(TNAME, ['-N', 'neutron-nwfilter-spoofing-fallb', '-P', 'ACCEPT']),
(TNAME, ['-N', 'neutron-nwfilter-OUTPUT', '-P', 'ACCEPT']),
(TNAME, ['-N', 'neutron-nwfilter-INPUT', '-P', 'ACCEPT']),
(TNAME, ['-N', 'neutron-nwfilter-FORWARD', '-P', 'ACCEPT']),
(TNAME, ['-A', 'INPUT', '-j', 'neutron-nwfilter-INPUT']),
(TNAME, ['-A', 'OUTPUT', '-j', 'neutron-nwfilter-OUTPUT']),
(TNAME, ['-A', 'FORWARD', '-j', 'neutron-nwfilter-FORWARD']),
(TNAME, ['-A', 'neutron-nwfilter-spoofing-fallb', '-j', 'DROP']),
(TNAME, ['--atomic-commit'])]
self.assertEqual(expected_res, observed_res)
class EbtablesDriverLowLevelOutputTestCase(base.BaseTestCase):
def test_ebtables_save_and_restore(self):
test_output = ('Bridge table: filter\n'
'Bridge chain: INPUT, entries: 1, policy: ACCEPT\n'
'-j CONTINUE , pcnt = 0 -- bcnt = 0\n'
'Bridge chain: FORWARD, entries: 1, policy: ACCEPT\n'
'-j CONTINUE , pcnt = 0 -- bcnt = 1\n'
'Bridge chain: OUTPUT, entries: 1, policy: ACCEPT\n'
'-j CONTINUE , pcnt = 1 -- bcnt = 1').split('\n')
observed_res = eb._process_ebtables_output(test_output)
expected_res = ['*filter',
':INPUT ACCEPT',
':FORWARD ACCEPT',
':OUTPUT ACCEPT',
'[0:0] -A INPUT -j CONTINUE',
'[0:1] -A FORWARD -j CONTINUE',
'[1:1] -A OUTPUT -j CONTINUE',
'COMMIT']
self.assertEqual(expected_res, observed_res)
class EbtablesDriverTestCase(base.BaseTestCase):
def setUp(self):
super(EbtablesDriverTestCase, self).setUp()
self.root_helper = 'sudo'
self.ebtables_path = CONF.ebtables_path
self.execute_p = mock.patch('neutron.agent.linux.utils.execute')
self.execute = self.execute_p.start()
def test_ebtables_sanity_check(self):
self.assertTrue(ebtables_supported())
self.execute.assert_has_calls([mock.call(['ebtables', '--version'])])
self.execute.side_effect = RuntimeError
self.assertFalse(ebtables_supported())

1
setup.cfg
View File

@ -34,6 +34,7 @@ data_files =
etc/neutron/rootwrap.d/debug.filters
etc/neutron/rootwrap.d/dhcp.filters
etc/neutron/rootwrap.d/iptables-firewall.filters
etc/neutron/rootwrap.d/ebtables.filters
etc/neutron/rootwrap.d/ipset-firewall.filters
etc/neutron/rootwrap.d/l3.filters
etc/neutron/rootwrap.d/linuxbridge-plugin.filters