From c8a819aff4db62e58192af0a272a7f1ce7923146 Mon Sep 17 00:00:00 2001 From: Rodolfo Alonso Hernandez Date: Thu, 24 Sep 2020 09:44:47 +0000 Subject: [PATCH] Filter out port with invalid ofport in OVS firewall Since [1], "get_vif_port_by_id" is also returning ports with an invalid ofport. OVS firewall cannot set an OpenFlow rule for a port without a valid ofport. "get_ovs_port" should filter out those ports. Related-Bug: #1815989 Related-Bug: #1734320 [1]https://review.opendev.org/#/c/640258/ Change-Id: Id12486b3127ab4ac8ad9ef2b3641da1b79a25a50 --- neutron/agent/linux/openvswitch_firewall/firewall.py | 3 ++- .../agent/linux/openvswitch_firewall/test_firewall.py | 8 ++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/neutron/agent/linux/openvswitch_firewall/firewall.py b/neutron/agent/linux/openvswitch_firewall/firewall.py index 9e64250a384..8c7bfc4136e 100644 --- a/neutron/agent/linux/openvswitch_firewall/firewall.py +++ b/neutron/agent/linux/openvswitch_firewall/firewall.py @@ -585,7 +585,8 @@ class OVSFirewallDriver(firewall.FirewallDriver): def get_ovs_port(self, port_id): ovs_port = self.int_br.br.get_vif_port_by_id(port_id) - if not ovs_port: + if not ovs_port or ovs_port.ofport in (ovs_lib.UNASSIGNED_OFPORT, + ovs_lib.INVALID_OFPORT): raise exceptions.OVSFWPortNotFound(port_id=port_id) return ovs_port diff --git a/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py b/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py index 6d08f2064a5..3c744b03a95 100644 --- a/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py +++ b/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py @@ -933,6 +933,14 @@ class TestOVSFirewallDriver(base.BaseTestCase): with testtools.ExpectedException(exceptions.OVSFWPortNotFound): self.firewall.get_ovs_port('port_id') + def test_get_ovs_port_invalid(self): + vif_port = ovs_lib.VifPort('name', 'ofport', 'id', 'mac', 'switch') + self.mock_bridge.br.get_vif_port_by_id.return_value = vif_port + for ofport in (ovs_lib.UNASSIGNED_OFPORT, ovs_lib.INVALID_OFPORT): + vif_port.ofport = ofport + with testtools.ExpectedException(exceptions.OVSFWPortNotFound): + self.firewall.get_ovs_port('port_id') + def test__initialize_egress_no_port_security_sends_to_egress(self): self.mock_bridge.br.db_get_val.return_value = {'tag': TESTING_VLAN_TAG} self.firewall._initialize_egress_no_port_security('port_id')