diff --git a/neutron/agent/linux/openvswitch_firewall/firewall.py b/neutron/agent/linux/openvswitch_firewall/firewall.py
index 8817751ba0a..413b7ae70a1 100644
--- a/neutron/agent/linux/openvswitch_firewall/firewall.py
+++ b/neutron/agent/linux/openvswitch_firewall/firewall.py
@@ -145,6 +145,8 @@ class SGPortMap(object):
     def __init__(self):
         self.ports = {}
         self.sec_groups = {}
+        # Maps port_id to ofport number
+        self.unfiltered = {}
 
     def get_sg(self, sg_id):
         return self.sec_groups.get(sg_id, None)
@@ -615,12 +617,19 @@ class OVSFirewallDriver(firewall.FirewallDriver):
             )
 
     def _initialize_egress_no_port_security(self, port_id):
-        ovs_port = self.get_ovs_port(port_id)
         try:
+            ovs_port = self.get_ovs_port(port_id)
             vlan_tag = self._get_port_vlan_tag(ovs_port.port_name)
         except exceptions.OVSFWTagNotFound:
             # It's a patch port, don't set anything
             return
+        except exceptions.OVSFWPortNotFound as not_found_e:
+            LOG.error("Initializing unfiltered port %(port_id)s that does not "
+                      "exist in ovsdb: %(err)s.",
+                      {'port_id': port_id,
+                       'err': not_found_e})
+            return
+        self.sg_port_map.unfiltered[port_id] = ovs_port.ofport
         self._add_flow(
             table=ovs_consts.TRANSIENT_TABLE,
             priority=100,
@@ -642,21 +651,20 @@ class OVSFirewallDriver(firewall.FirewallDriver):
         )
 
     def _remove_egress_no_port_security(self, port_id):
-        ovs_port = self.get_ovs_port(port_id)
         try:
-            # Test if it's a patch port
-            self._get_port_vlan_tag(ovs_port.port_name)
-        except exceptions.OVSFWTagNotFound:
-            # It's a patch port, don't do anything
+            ofport = self.sg_port_map.unfiltered[port_id]
+        except KeyError:
+            LOG.debug("Port %s is not handled by the firewall.", port_id)
             return
         self._delete_flows(
             table=ovs_consts.TRANSIENT_TABLE,
-            in_port=ovs_port.ofport
+            in_port=ofport
         )
         self._delete_flows(
             table=ovs_consts.ACCEPT_OR_INGRESS_TABLE,
-            reg_port=ovs_port.ofport
+            reg_port=ofport
         )
+        del self.sg_port_map.unfiltered[port_id]
 
     def _initialize_egress(self, port):
         """Identify egress traffic and send it to egress base"""
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 6c6cebc2eac..1c8d8ab4247 100644
--- a/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py
+++ b/neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py
@@ -639,6 +639,7 @@ class TestOVSFirewallDriver(base.BaseTestCase):
 
     def test__remove_egress_no_port_security_deletes_flow(self):
         self.mock_bridge.br.db_get_val.return_value = {'tag': TESTING_VLAN_TAG}
+        self.firewall.sg_port_map.unfiltered['port_id'] = 1
         self.firewall._remove_egress_no_port_security('port_id')
         expected_call = mock.call(
             table=ovs_consts.TRANSIENT_TABLE,
@@ -651,3 +652,23 @@ class TestOVSFirewallDriver(base.BaseTestCase):
         self.mock_bridge.br.db_get_val.return_value = {}
         self.firewall._remove_egress_no_port_security('port_id')
         self.assertFalse(self.mock_bridge.br.delete_flows.called)
+
+    def test_process_trusted_ports_caches_port_id(self):
+        self.firewall.process_trusted_ports(['port_id'])
+        self.assertIn('port_id', self.firewall.sg_port_map.unfiltered)
+
+    def test_process_trusted_ports_port_not_found(self):
+        """Check that exception is not propagated outside."""
+        self.mock_bridge.br.get_vif_port_by_id.return_value = None
+        self.firewall.process_trusted_ports(['port_id'])
+        # Processing should have failed so port is not cached
+        self.assertNotIn('port_id', self.firewall.sg_port_map.unfiltered)
+
+    def test_remove_trusted_ports_clears_cached_port_id(self):
+        self.firewall.sg_port_map.unfiltered['port_id'] = 1
+        self.firewall.remove_trusted_ports(['port_id'])
+        self.assertNotIn('port_id', self.firewall.sg_port_map.unfiltered)
+
+    def test_remove_trusted_ports_not_managed_port(self):
+        """Check that exception is not propagated outside."""
+        self.firewall.remove_trusted_ports(['port_id'])