Implements quantum security groups support on OVS plugin

implements bp quantum-security-groups-iptables-ovs
- Adding [SECURITYGROUP] firewall_driver to the conf
- Adding NoopFirewallDriver
- Adding OVSHybridIptablesFirewallDriver
- Refactoring security group code to support OVS plugin

Change-Id: I7aabc296265afc47d938121543ace57cce6cc521

etc/quantum/plugins/openvswitch/ovs_quantum_plugin.ini

@@ -97,6 +97,10 @@ reconnect_interval = 2
 # Agent's polling interval in seconds
 polling_interval = 2
 
+[SECURITYGROUP]
+# Firewall driver for realizing quantum security group function
+# firewall_driver = quantum.agent.linux.iptables_firewall.OVSHybridIptablesFirewallDriver
+
 #-----------------------------------------------------------------------------
 # Sample Configurations.
 #-----------------------------------------------------------------------------

quantum/agent/firewall.py

@@ -103,3 +103,33 @@ class FirewallDriver(object):
             yield
         finally:
             self.filter_defer_apply_off()
+
+
+class NoopFirewallDriver(FirewallDriver):
+    """ Noop Firewall Driver.
+
+    Firewall driver which does nothing.
+    This driver is for disabling the firewall functionality.
+    """
+
+    def prepare_port_filter(self, port):
+        pass
+
+    def apply_port_filter(self, port):
+        pass
+
+    def update_port_filter(self, port):
+        pass
+
+    def remove_port_filter(self, port):
+        pass
+
+    def filter_defer_apply_on(self):
+        pass
+
+    def filter_defer_apply_off(self):
+        pass
+
+    @property
+    def ports(self):
+        return {}

quantum/agent/linux/iptables_firewall.py

@@ -18,7 +18,9 @@
 import netaddr
 
 from quantum.agent import firewall
+from quantum.agent.linux import iptables_manager
 from quantum.common import constants
+from quantum.openstack.common import cfg
 from quantum.openstack.common import log as logging
 
 
@@ -28,16 +30,18 @@ INGRESS_DIRECTION = 'ingress'
 EGRESS_DIRECTION = 'egress'
 CHAIN_NAME_PREFIX = {INGRESS_DIRECTION: 'i',
                      EGRESS_DIRECTION: 'o'}
-IPTABLES_DIRECTION = {INGRESS_DIRECTION: 'physdev-out',
+LINUX_DEV_LEN = 14
-                      EGRESS_DIRECTION: 'physdev-in'}
 
 
 class IptablesFirewallDriver(firewall.FirewallDriver):
     """Driver which enforces security groups through iptables rules."""
+    IPTABLES_DIRECTION = {INGRESS_DIRECTION: 'physdev-out',
+                          EGRESS_DIRECTION: 'physdev-in'}
 
-    def __init__(self, iptables_manager):
+    def __init__(self):
-        self.iptables = iptables_manager
+        self.iptables = iptables_manager.IptablesManager(
-
+            root_helper=cfg.CONF.AGENT.root_helper,
+            use_ipv6=True)
         # list of port which has security group
         self.filtered_ports = {}
         self._add_fallback_chain_v4v6()
@@ -121,6 +125,9 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
         for rule in ipv6_rules:
             self.iptables.ipv6['filter'].add_rule(chain_name, rule)
 
+    def _get_device_name(self, port):
+        return port['device']
+
     def _add_chain(self, port, direction):
         chain_name = self._port_chain_name(port, direction)
         self._add_chain_by_name_v4v6(chain_name)
@@ -131,16 +138,16 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
         # We accept the packet at the end of SG_CHAIN.
 
         # jump to the security group chain
-        device = port['device']
+        device = self._get_device_name(port)
         jump_rule = ['-m physdev --physdev-is-bridged --%s '
-                     '%s -j $%s' % (IPTABLES_DIRECTION[direction],
+                     '%s -j $%s' % (self.IPTABLES_DIRECTION[direction],
                                     device,
                                     SG_CHAIN)]
         self._add_rule_to_chain_v4v6('FORWARD', jump_rule, jump_rule)
 
         # jump to the chain based on the device
         jump_rule = ['-m physdev --physdev-is-bridged --%s '
-                     '%s -j $%s' % (IPTABLES_DIRECTION[direction],
+                     '%s -j $%s' % (self.IPTABLES_DIRECTION[direction],
                                     device,
                                     chain_name)]
         self._add_rule_to_chain_v4v6(SG_CHAIN, jump_rule, jump_rule)
@@ -278,3 +285,17 @@ class IptablesFirewallDriver(firewall.FirewallDriver):
 
     def filter_defer_apply_off(self):
         self.iptables.defer_apply_off()
+
+
+class OVSHybridIptablesFirewallDriver(IptablesFirewallDriver):
+    OVS_HYBRID_TAP_PREFIX = 'tap'
+
+    def _port_chain_name(self, port, direction):
+        #Note (nati) make chain name short less than 28 char
+        # with extra prefix
+        # ( see comment in iptables_manager )
+        return '%s%s' % (CHAIN_NAME_PREFIX[direction],
+                         port['device'][0:10])
+
+    def _get_device_name(self, port):
+        return (self.OVS_HYBRID_TAP_PREFIX + port['device'])[:LINUX_DEV_LEN]

quantum/agent/securitygroups_rpc.py

@@ -16,14 +16,21 @@
 #    under the License.
 #
 
-from quantum.agent.linux import iptables_firewall
-from quantum.agent.linux import iptables_manager
 from quantum.common import topics
+from quantum.openstack.common import cfg
+from quantum.openstack.common import importutils
 from quantum.openstack.common import log as logging
 
 LOG = logging.getLogger(__name__)
 SG_RPC_VERSION = "1.1"
 
+security_group_opts = [
+    cfg.StrOpt(
+        'firewall_driver',
+        default='quantum.agent.firewall.NoopFirewallDriver')
+]
+cfg.CONF.register_opts(security_group_opts, 'SECURITYGROUP')
+
 
 class SecurityGroupServerRpcApiMixin(object):
     """A mix-in that enable SecurityGroup support in plugin rpc
@@ -42,6 +49,8 @@ class SecurityGroupAgentRpcCallbackMixin(object):
     """A mix-in that enable SecurityGroup agent
     support in agent implementations.
     """
+    #mix-in object should be have sg_agent
+    sg_agent = None
 
     def security_groups_rule_updated(self, context, **kwargs):
         """ callback for security group rule update
@@ -51,7 +60,7 @@ class SecurityGroupAgentRpcCallbackMixin(object):
         security_groups = kwargs.get('security_groups', [])
         LOG.debug(
             _("Security group rule updated on remote: %s"), security_groups)
-        self.agent.security_groups_rule_updated(security_groups)
+        self.sg_agent.security_groups_rule_updated(security_groups)
 
     def security_groups_member_updated(self, context, **kwargs):
         """ callback for security group member update
@@ -61,14 +70,14 @@ class SecurityGroupAgentRpcCallbackMixin(object):
         security_groups = kwargs.get('security_groups', [])
         LOG.debug(
             _("Security group member updated on remote: %s"), security_groups)
-        self.agent.security_groups_member_updated(security_groups)
+        self.sg_agent.security_groups_member_updated(security_groups)
 
     def security_groups_provider_updated(self, context, **kwargs):
         """ callback for security group provider update
 
         """
         LOG.debug(_("Provider rule updated"))
-        self.agent.security_groups_provider_updated()
+        self.sg_agent.security_groups_provider_updated()
 
 
 class SecurityGroupAgentRpcMixin(object):
@@ -78,10 +87,8 @@ class SecurityGroupAgentRpcMixin(object):
 
     def init_firewall(self):
         LOG.debug(_("Init firewall settings"))
-        ip_manager = iptables_manager.IptablesManager(
+        self.firewall = importutils.import_object(
-            root_helper=self.root_helper,
+            cfg.CONF.SECURITYGROUP.firewall_driver)
-            use_ipv6=True)
-        self.firewall = iptables_firewall.IptablesFirewallDriver(ip_manager)
 
     def prepare_devices_filter(self, device_ids):
         if not device_ids:

quantum/db/migration/alembic_migrations/versions/3cb5d900c5de_security_groups.py

@@ -31,7 +31,8 @@ down_revision = '48b6f43f7471'
 
 migration_for_plugins = [
     'quantum.plugins.linuxbridge.lb_quantum_plugin.LinuxBridgePluginV2',
-    'quantum.plugins.nicira.nicira_nvp_plugin.QuantumPlugin.NvpPluginV2'
+    'quantum.plugins.nicira.nicira_nvp_plugin.QuantumPlugin.NvpPluginV2',
+    'quantum.plugins.openvswitch.ovs_quantum_plugin.OVSQuantumPluginV2',
 ]
 
 from alembic import op

quantum/db/securitygroups_rpc_base.py

@@ -18,8 +18,10 @@
 import netaddr
 
 from quantum.common import constants as q_const
+from quantum.common import utils
 from quantum.db import models_v2
 from quantum.db import securitygroups_db as sg_db
+from quantum.extensions import securitygroup as ext_sg
 from quantum.openstack.common import log as logging
 
 LOG = logging.getLogger(__name__)
@@ -59,6 +61,46 @@ class SecurityGroupServerRpcMixin(sg_db.SecurityGroupDbMixin):
         self.notifier.security_groups_rule_updated(context,
                                                    [rule['security_group_id']])
 
+    def update_security_group_on_port(self, context, id, port,
+                                      original_port, updated_port):
+        """ update security groups on port
+
+        This method returns a flag which indicates request notification
+        is required and does not perform notification itself.
+        It is because another changes for the port may require notification.
+        """
+        need_notify = False
+        if ext_sg.SECURITYGROUPS in port['port']:
+            # delete the port binding and read it with the new rules
+            port['port'][ext_sg.SECURITYGROUPS] = (
+                self._get_security_groups_on_port(context, port))
+            self._delete_port_security_group_bindings(context, id)
+            self._process_port_create_security_group(
+                context,
+                id,
+                port['port'][ext_sg.SECURITYGROUPS])
+            need_notify = True
+        self._extend_port_dict_security_group(context, updated_port)
+        return need_notify
+
+    def is_security_group_member_updated(self, context,
+                                         original_port, updated_port):
+        """ check security group member updated or not
+
+        This method returns a flag which indicates request notification
+        is required and does not perform notification itself.
+        It is because another changes for the port may require notification.
+        """
+        need_notify = False
+        if (original_port['fixed_ips'] != updated_port['fixed_ips'] or
+            not utils.compare_elements(
+                original_port.get(ext_sg.SECURITYGROUPS),
+                updated_port.get(ext_sg.SECURITYGROUPS))):
+            self.notifier.security_groups_member_updated(
+                context, updated_port.get(ext_sg.SECURITYGROUPS))
+            need_notify = True
+        return need_notify
+
 
 class SecurityGroupServerRpcCallbackMixin(object):
     """A mix-in that enable SecurityGroup agent

quantum/plugins/linuxbridge/agent/linuxbridge_quantum_agent.py

@@ -400,6 +400,7 @@ class LinuxBridgeRpcCallbacks(sg_rpc.SecurityGroupAgentRpcCallbackMixin):
     def __init__(self, context, agent):
         self.context = context
         self.agent = agent
+        self.sg_agent = agent
 
     def network_delete(self, context, **kwargs):
         LOG.debug(_("network_delete received"))
@@ -418,7 +419,7 @@ class LinuxBridgeRpcCallbacks(sg_rpc.SecurityGroupAgentRpcCallbackMixin):
             return
 
         if 'security_groups' in port:
-            self.agent.refresh_firewall()
+            self.sg_agent.refresh_firewall()
 
         if port['admin_state_up']:
             vlan_id = kwargs.get('vlan_id')

quantum/plugins/linuxbridge/db/l2network_db_v2.py

@@ -18,9 +18,9 @@ from sqlalchemy.orm import exc
 
 from quantum.common import exceptions as q_exc
 import quantum.db.api as db
-from quantum import manager
 from quantum.db import models_v2
 from quantum.db import securitygroups_db as sg_db
+from quantum import manager
 from quantum.openstack.common import log as logging
 # NOTE (e0ne): this import is needed for config init
 from quantum.plugins.linuxbridge.common import config

quantum/plugins/linuxbridge/lb_quantum_plugin.py

@@ -499,37 +499,23 @@ class LinuxBridgePluginV2(db_base_plugin_v2.QuantumDbPluginV2,
     def update_port(self, context, id, port):
         original_port = self.get_port(context, id)
         session = context.session
-        port_updated = False
+        need_port_update_notify = False
+
         with session.begin(subtransactions=True):
-            # delete the port binding and read it with the new rules
+            updated_port = super(LinuxBridgePluginV2, self).update_port(
-            if ext_sg.SECURITYGROUPS in port['port']:
-                port['port'][ext_sg.SECURITYGROUPS] = (
-                    self._get_security_groups_on_port(context, port))
-                self._delete_port_security_group_bindings(context, id)
-                self._process_port_create_security_group(
-                    context,
-                    id,
-                    port['port'][ext_sg.SECURITYGROUPS])
-                port_updated = True
-
-            port = super(LinuxBridgePluginV2, self).update_port(
                 context, id, port)
-            self._extend_port_dict_security_group(context, port)
+            need_port_update_notify = self.update_security_group_on_port(
+                context, id, port, original_port, updated_port)
 
-        if original_port['admin_state_up'] != port['admin_state_up']:
+        need_port_update_notify |= self.is_security_group_member_updated(
-            port_updated = True
+            context, original_port, updated_port)
 
-        if (original_port['fixed_ips'] != port['fixed_ips'] or
+        if original_port['admin_state_up'] != updated_port['admin_state_up']:
-            not utils.compare_elements(
+            need_port_update_notify = True
-                original_port.get(ext_sg.SECURITYGROUPS),
-                port.get(ext_sg.SECURITYGROUPS))):
-            self.notifier.security_groups_member_updated(
-                context, port.get(ext_sg.SECURITYGROUPS))
 
-        if port_updated:
+        if need_port_update_notify:
-            self._notify_port_updated(context, port)
+            self._notify_port_updated(context, updated_port)
-
+        return self._extend_port_dict_binding(context, updated_port)
-        return self._extend_port_dict_binding(context, port)
 
     def delete_port(self, context, id, l3_port_check=True):
 
@@ -544,6 +530,7 @@ class LinuxBridgePluginV2(db_base_plugin_v2.QuantumDbPluginV2,
             port = self.get_port(context, id)
             self._delete_port_security_group_bindings(context, id)
             super(LinuxBridgePluginV2, self).delete_port(context, id)
+
         self.notifier.security_groups_member_updated(
             context, port.get(ext_sg.SECURITYGROUPS))
 

quantum/plugins/openvswitch/agent/ovs_quantum_agent.py

@@ -29,6 +29,7 @@ from quantum.agent.linux import ip_lib
 from quantum.agent.linux import ovs_lib
 from quantum.agent.linux import utils
 from quantum.agent import rpc as agent_rpc
+from quantum.agent import securitygroups_rpc as sg_rpc
 from quantum.common import config as logging_config
 from quantum.common import topics
 from quantum.common import utils as q_utils
@@ -37,6 +38,7 @@ from quantum.openstack.common import cfg
 from quantum.openstack.common import log as logging
 from quantum.openstack.common.rpc import dispatcher
 from quantum.plugins.openvswitch.common import config
+from quantum.extensions import securitygroup as ext_sg
 from quantum.plugins.openvswitch.common import constants
 
 
@@ -95,7 +97,20 @@ class Port(object):
         return hash(self.id)
 
 
-class OVSQuantumAgent(object):
+class OVSPluginApi(agent_rpc.PluginApi,
+                   sg_rpc.SecurityGroupServerRpcApiMixin):
+    pass
+
+
+class OVSSecurityGroupAgent(sg_rpc.SecurityGroupAgentRpcMixin):
+    def __init__(self, context, plugin_rpc, root_helper):
+        self.context = context
+        self.plugin_rpc = plugin_rpc
+        self.root_helper = root_helper
+        self.init_firewall()
+
+
+class OVSQuantumAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin):
     '''Implements OVS-based tunneling, VLANs and flat networks.
 
     Two local bridges are created: an integration bridge (defaults to
@@ -128,8 +143,10 @@ class OVSQuantumAgent(object):
     # Upper bound on available vlans.
     MAX_VLAN_TAG = 4094
 
-    # Set RPC API version to 1.0 by default.
+    # history
-    RPC_API_VERSION = '1.0'
+    #   1.0 Initial version
+    #   1.1 Support Security Group RPC
+    RPC_API_VERSION = '1.1'
 
     def __init__(self, integ_br, tun_br, local_ip,
                  bridge_mappings, root_helper,
@@ -162,11 +179,16 @@ class OVSQuantumAgent(object):
 
         self.setup_rpc(integ_br)
 
+        # Security group agent supprot
+        self.sg_agent = OVSSecurityGroupAgent(self.context,
+                                              self.plugin_rpc,
+                                              root_helper)
+
     def setup_rpc(self, integ_br):
         mac = utils.get_interface_mac(integ_br)
         self.agent_id = '%s%s' % ('ovs', (mac.replace(":", "")))
         self.topic = topics.AGENT
-        self.plugin_rpc = agent_rpc.PluginApi(topics.PLUGIN)
+        self.plugin_rpc = OVSPluginApi(topics.PLUGIN)
 
         # RPC network init
         self.context = context.get_admin_context_without_session()
@@ -175,7 +197,8 @@ class OVSQuantumAgent(object):
         # Define the listening consumers for the agent
         consumers = [[topics.PORT, topics.UPDATE],
                      [topics.NETWORK, topics.DELETE],
-                     [constants.TUNNEL, topics.UPDATE]]
+                     [constants.TUNNEL, topics.UPDATE],
+                     [topics.SECURITY_GROUP, topics.UPDATE]]
         self.connection = agent_rpc.create_consumers(self.dispatcher,
                                                      self.topic,
                                                      consumers)
@@ -203,6 +226,9 @@ class OVSQuantumAgent(object):
         vif_port = self.int_br.get_vif_port_by_id(port['id'])
         if not vif_port:
             return
+
+        if ext_sg.SECURITYGROUPS in port:
+            self.sg_agent.refresh_firewall()
         network_type = kwargs.get('network_type')
         segmentation_id = kwargs.get('segmentation_id')
         physical_network = kwargs.get('physical_network')
@@ -549,6 +575,7 @@ class OVSQuantumAgent(object):
 
     def treat_devices_added(self, devices):
         resync = False
+        self.sg_agent.prepare_devices_filter(devices)
         for device in devices:
             LOG.info(_("Port %s added"), device)
             try:
@@ -578,6 +605,7 @@ class OVSQuantumAgent(object):
 
     def treat_devices_removed(self, devices):
         resync = False
+        self.sg_agent.remove_devices_filter(devices)
         for device in devices:
             LOG.info(_("Attachment %s removed"), device)
             try:

quantum/plugins/openvswitch/ovs_db_v2.py

@@ -21,6 +21,9 @@ from sqlalchemy.orm import exc
 from quantum.common import exceptions as q_exc
 import quantum.db.api as db
 from quantum.db import models_v2
+from quantum.db import securitygroups_db as sg_db
+from quantum.extensions import securitygroup as ext_sg
+from quantum import manager
 from quantum.openstack.common import cfg
 from quantum.openstack.common import log as logging
 from quantum.plugins.openvswitch.common import constants
@@ -300,6 +303,32 @@ def get_port(port_id):
     return port
 
 
+def get_port_from_device(port_id):
+    """Get port from database"""
+    LOG.debug(_("get_port_with_securitygroups() called:port_id=%s"), port_id)
+    session = db.get_session()
+    sg_binding_port = sg_db.SecurityGroupPortBinding.port_id
+
+    query = session.query(models_v2.Port,
+                          sg_db.SecurityGroupPortBinding.security_group_id)
+    query = query.outerjoin(sg_db.SecurityGroupPortBinding,
+                            models_v2.Port.id == sg_binding_port)
+    query = query.filter(models_v2.Port.id == port_id)
+    port_and_sgs = query.all()
+    if not port_and_sgs:
+        return None
+    port = port_and_sgs[0][0]
+    plugin = manager.QuantumManager.get_plugin()
+    port_dict = plugin._make_port_dict(port)
+    port_dict[ext_sg.SECURITYGROUPS] = [
+        sg_id for port, sg_id in port_and_sgs if sg_id]
+    port_dict['security_group_rules'] = []
+    port_dict['security_group_source_groups'] = []
+    port_dict['fixed_ips'] = [ip['ip_address']
+                              for ip in port['fixed_ips']]
+    return port_dict
+
+
 def set_port_status(port_id, status):
     session = db.get_session()
     try:

quantum/plugins/openvswitch/ovs_quantum_plugin.py

@@ -22,6 +22,7 @@
 
 import sys
 
+from quantum.agent import securitygroups_rpc as sg_rpc
 from quantum.api.v2 import attributes
 from quantum.common import constants as q_const
 from quantum.common import exceptions as q_exc
@@ -33,8 +34,10 @@ from quantum.db import l3_db
 from quantum.db import l3_rpc_base
 # NOTE: quota_db cannot be removed, it is for db model
 from quantum.db import quota_db
+from quantum.db import securitygroups_rpc_base as sg_db_rpc
 from quantum.extensions import portbindings
 from quantum.extensions import providernet as provider
+from quantum.extensions import securitygroup as ext_sg
 from quantum.openstack.common import cfg
 from quantum.openstack.common import log as logging
 from quantum.openstack.common import rpc
@@ -49,10 +52,14 @@ LOG = logging.getLogger(__name__)
 
 
 class OVSRpcCallbacks(dhcp_rpc_base.DhcpRpcCallbackMixin,
-                      l3_rpc_base.L3RpcCallbackMixin):
+                      l3_rpc_base.L3RpcCallbackMixin,
+                      sg_db_rpc.SecurityGroupServerRpcCallbackMixin):
 
-    # Set RPC API version to 1.0 by default.
+    # history
-    RPC_API_VERSION = '1.0'
+    #   1.0 Initial version
+    #   1.1 Support Security Group RPC
+
+    RPC_API_VERSION = '1.1'
 
     def __init__(self, notifier):
         self.notifier = notifier
@@ -65,6 +72,13 @@ class OVSRpcCallbacks(dhcp_rpc_base.DhcpRpcCallbackMixin,
         '''
         return q_rpc.PluginRpcDispatcher([self])
 
+    @classmethod
+    def get_port_from_device(cls, device):
+        port = ovs_db_v2.get_port_from_device(device)
+        if port:
+            port['device'] = device
+        return port
+
     def get_device_details(self, rpc_context, **kwargs):
         """Agent requests device details"""
         agent_id = kwargs.get('agent_id')
@@ -143,7 +157,8 @@ class OVSRpcCallbacks(dhcp_rpc_base.DhcpRpcCallbackMixin,
         return entry
 
 
-class AgentNotifierApi(proxy.RpcProxy):
+class AgentNotifierApi(proxy.RpcProxy,
+                       sg_rpc.SecurityGroupAgentRpcApiMixin):
     '''Agent side of the openvswitch rpc API.
 
     API version history:
@@ -191,7 +206,8 @@ class AgentNotifierApi(proxy.RpcProxy):
 
 
 class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
-                         l3_db.L3_NAT_db_mixin):
+                         l3_db.L3_NAT_db_mixin,
+                         sg_db_rpc.SecurityGroupServerRpcMixin):
     """Implement the Quantum abstractions using Open vSwitch.
 
     Depending on whether tunneling is enabled, either a GRE tunnel or
@@ -213,7 +229,8 @@ class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
     # bulk operations. Name mangling is used in order to ensure it
     # is qualified by class
     __native_bulk_support = True
-    supported_extension_aliases = ["provider", "router", "binding", "quotas"]
+    supported_extension_aliases = ["provider", "router",
+                                   "binding", "quotas", "security-group"]
 
     network_view = "extension:provider_network:view"
     network_set = "extension:provider_network:set"
@@ -425,6 +442,11 @@ class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
                                                           network['network'])
 
         session = context.session
+        #set up default security groups
+        tenant_id = self._get_tenant_id_for_create(
+            context, network['network'])
+        self._ensure_default_security_group(context, tenant_id)
+
         with session.begin(subtransactions=True):
             if not network_type:
                 # tenant network
@@ -521,32 +543,68 @@ class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
     def create_port(self, context, port):
         # Set port status as 'DOWN'. This will be updated by agent
         port['port']['status'] = q_const.PORT_STATUS_DOWN
+        session = context.session
+        with session.begin(subtransactions=True):
+            self._ensure_default_security_group_on_port(context, port)
+            sgids = self._get_security_groups_on_port(context, port)
             port = super(OVSQuantumPluginV2, self).create_port(context, port)
+            self._process_port_create_security_group(
+                context, port['id'], sgids)
+            self._extend_port_dict_security_group(context, port)
+        #Note(nati): In order to allow dhcp packets,
+        # changes for dhcp ip should be notifified
+        if port['device_owner'] == q_const.DEVICE_OWNER_DHCP:
+            self.notifier.security_groups_provider_updated(context)
+        else:
+            self.notifier.security_groups_member_updated(
+                context, port.get(ext_sg.SECURITYGROUPS))
         return self._extend_port_dict_binding(context, port)
 
     def get_port(self, context, id, fields=None):
-        port = super(OVSQuantumPluginV2, self).get_port(context, id, fields)
+        with context.session.begin(subtransactions=True):
-        return self._fields(self._extend_port_dict_binding(context, port),
+            port = super(OVSQuantumPluginV2, self).get_port(context,
-                            fields)
+                                                            id, fields)
+            self._extend_port_dict_security_group(context, port)
+            self._extend_port_dict_binding(context, port)
+        return self._fields(port, fields)
 
     def get_ports(self, context, filters=None, fields=None):
-        ports = super(OVSQuantumPluginV2, self).get_ports(context, filters,
+        with context.session.begin(subtransactions=True):
-                                                          fields)
+            ports = super(OVSQuantumPluginV2, self).get_ports(
-        return [self._fields(self._extend_port_dict_binding(context, port),
+                context, filters, fields)
-                             fields) for port in ports]
+            #TODO(nati) filter by security group
+            for port in ports:
+                self._extend_port_dict_security_group(context, port)
+                self._extend_port_dict_binding(context, port)
+        return [self._fields(port, fields) for port in ports]
 
     def update_port(self, context, id, port):
-        original_port = super(OVSQuantumPluginV2, self).get_port(context,
+        session = context.session
-                                                                 id)
+
-        port = super(OVSQuantumPluginV2, self).update_port(context, id, port)
+        need_port_update_notify = False
-        if original_port['admin_state_up'] != port['admin_state_up']:
+        with session.begin(subtransactions=True):
+            original_port = super(OVSQuantumPluginV2, self).get_port(
+                context, id)
+            updated_port = super(OVSQuantumPluginV2, self).update_port(
+                context, id, port)
+            need_port_update_notify = self.update_security_group_on_port(
+                context, id, port, original_port, updated_port)
+
+        need_port_update_notify |= self.is_security_group_member_updated(
+            context, original_port, updated_port)
+
+        if original_port['admin_state_up'] != updated_port['admin_state_up']:
+            need_port_update_notify = True
+
+        if need_port_update_notify:
             binding = ovs_db_v2.get_network_binding(None,
-                                                    port['network_id'])
+                                                    updated_port['network_id'])
-            self.notifier.port_update(context, port,
+            self.notifier.port_update(context, updated_port,
                                       binding.network_type,
                                       binding.segmentation_id,
                                       binding.physical_network)
-        return self._extend_port_dict_binding(context, port)
+
+        return self._extend_port_dict_binding(context, updated_port)
 
     def delete_port(self, context, id, l3_port_check=True):
 
@@ -554,5 +612,13 @@ class OVSQuantumPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
         # and l3-router.  If so, we should prevent deletion.
         if l3_port_check:
             self.prevent_l3_port_deletion(context, id)
+
+        session = context.session
+        with session.begin(subtransactions=True):
             self.disassociate_floatingips(context, id)
-        return super(OVSQuantumPluginV2, self).delete_port(context, id)
+            port = self.get_port(context, id)
+            self._delete_port_security_group_bindings(context, id)
+            super(OVSQuantumPluginV2, self).delete_port(context, id)
+
+        self.notifier.security_groups_member_updated(
+            context, port.get(ext_sg.SECURITYGROUPS))

quantum/tests/unit/linuxbridge/test_lb_security_group.py

@@ -22,6 +22,8 @@ from quantum.api.v2 import attributes
 from quantum.extensions import securitygroup as ext_sg
 from quantum.plugins.linuxbridge.db import l2network_db_v2 as lb_db
 from quantum.tests.unit import test_extension_security_group as test_sg
+from quantum.tests.unit import test_security_groups_rpc as test_sg_rpc
+
 
 PLUGIN_NAME = ('quantum.plugins.linuxbridge.'
                'lb_quantum_plugin.LinuxBridgePluginV2')
@@ -53,56 +55,9 @@ class LinuxBridgeSecurityGroupsTestCase(test_sg.SecurityGroupDBTestCase):
 
 
 class TestLinuxBridgeSecurityGroups(LinuxBridgeSecurityGroupsTestCase,
-                                    test_sg.TestSecurityGroups):
+                                    test_sg.TestSecurityGroups,
-
+                                    test_sg_rpc.SGNotificationTestMixin):
-    def test_security_group_rule_updated(self):
-        name = 'webservers'
-        description = 'my webservers'
-        with self.security_group(name, description) as sg:
-            with self.security_group(name, description) as sg2:
-                security_group_id = sg['security_group']['id']
-                direction = "ingress"
-                source_group_id = sg2['security_group']['id']
-                protocol = 'tcp'
-                port_range_min = 88
-                port_range_max = 88
-                with self.security_group_rule(security_group_id, direction,
-                                              protocol, port_range_min,
-                                              port_range_max,
-                                              source_group_id=source_group_id
-                                              ):
     pass
-            self.notifier.assert_has_calls(
-                [call.security_groups_rule_updated(mock.ANY,
-                                                   [security_group_id]),
-                 call.security_groups_rule_updated(mock.ANY,
-                                                   [security_group_id])])
-
-    def test_security_group_member_updated(self):
-        with self.network() as n:
-            with self.subnet(n):
-                with self.security_group() as sg:
-                    security_group_id = sg['security_group']['id']
-                    res = self._create_port(self.fmt, n['network']['id'])
-                    port = self.deserialize(self.fmt, res)
-
-                    data = {'port': {'fixed_ips': port['port']['fixed_ips'],
-                                     'name': port['port']['name'],
-                                     ext_sg.SECURITYGROUPS:
-                                     [security_group_id]}}
-
-                    req = self.new_update_request('ports', data,
-                                                  port['port']['id'])
-                    res = self.deserialize(self.fmt,
-                                           req.get_response(self.api))
-                    self.assertEquals(res['port'][ext_sg.SECURITYGROUPS][0],
-                                      security_group_id)
-                    self._delete('ports', port['port']['id'])
-                    self.notifier.assert_has_calls(
-                        [call.security_groups_member_updated(
-                            mock.ANY, [mock.ANY]),
-                         call.security_groups_member_updated(
-                             mock.ANY, [security_group_id])])
 
 
 class TestLinuxBridgeSecurityGroupsXML(TestLinuxBridgeSecurityGroups):

quantum/tests/unit/openvswitch/test_openvswitch_plugin.py

@@ -43,7 +43,7 @@ class TestOpenvswitchPortsV2(test_plugin.TestPortsV2,
                              test_bindings.PortBindingsTestCase):
 
     VIF_TYPE = portbindings.VIF_TYPE_OVS
-    HAS_PORT_FILTER = False
+    HAS_PORT_FILTER = True
 
     def test_update_port_status_build(self):
         with self.port() as port:

quantum/tests/unit/openvswitch/test_ovs_quantum_agent.py

@@ -21,6 +21,10 @@ from quantum.openstack.common import cfg
 from quantum.plugins.openvswitch.agent import ovs_quantum_agent
 
 
+NOTIFIER = ('quantum.plugins.openvswitch.'
+            'ovs_quantum_plugin.AgentNotifierApi')
+
+
 class CreateAgentConfigMap(unittest.TestCase):
 
     def test_create_agent_config_map_succeeds(self):
@@ -38,6 +42,11 @@ class TestOvsQuantumAgent(unittest.TestCase):
 
     def setUp(self):
         self.addCleanup(cfg.CONF.reset)
+        self.addCleanup(mock.patch.stopall)
+        notifier_p = mock.patch(NOTIFIER)
+        notifier_cls = notifier_p.start()
+        self.notifier = mock.Mock()
+        notifier_cls.return_value = self.notifier
         # Avoid rpc initialization for unit tests
         cfg.CONF.set_override('rpc_backend',
                               'quantum.openstack.common.rpc.impl_fake')
@@ -48,9 +57,7 @@ class TestOvsQuantumAgent(unittest.TestCase):
             with mock.patch('quantum.agent.linux.utils.get_interface_mac',
                             return_value='000000000001'):
                 self.agent = ovs_quantum_agent.OVSQuantumAgent(**kwargs)
-        self.agent.plugin_rpc = mock.Mock()
+        self.agent.sg_agent = mock.Mock()
-        self.agent.context = mock.Mock()
-        self.agent.agent_id = mock.Mock()
 
     def mock_port_bound(self, ofport=None):
         port = mock.Mock()

quantum/tests/unit/openvswitch/test_ovs_security_group.py

@@ -0,0 +1,93 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright 2013, Nachi Ueno, NTT MCL, 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.
+
+import mock
+
+from quantum.api.v2 import attributes
+from quantum.extensions import securitygroup as ext_sg
+from quantum import manager
+from quantum.tests.unit import test_extension_security_group as test_sg
+from quantum.tests.unit import test_security_groups_rpc as test_sg_rpc
+
+PLUGIN_NAME = ('quantum.plugins.openvswitch.'
+               'ovs_quantum_plugin.OVSQuantumPluginV2')
+AGENT_NAME = ('quantum.plugins.openvswitch.'
+              'agent.ovs_quantum_agent.OVSQuantumAgent')
+NOTIFIER = ('quantum.plugins.openvswitch.'
+            'ovs_quantum_plugin.AgentNotifierApi')
+
+
+class OpenvswitchSecurityGroupsTestCase(test_sg.SecurityGroupDBTestCase):
+    _plugin_name = PLUGIN_NAME
+
+    def setUp(self, plugin=None):
+        self.addCleanup(mock.patch.stopall)
+        notifier_p = mock.patch(NOTIFIER)
+        notifier_cls = notifier_p.start()
+        self.notifier = mock.Mock()
+        notifier_cls.return_value = self.notifier
+        self._attribute_map_bk_ = {}
+        for item in attributes.RESOURCE_ATTRIBUTE_MAP:
+            self._attribute_map_bk_[item] = (attributes.
+                                             RESOURCE_ATTRIBUTE_MAP[item].
+                                             copy())
+        super(OpenvswitchSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
+
+    def tearDown(self):
+        super(OpenvswitchSecurityGroupsTestCase, self).tearDown()
+        attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
+
+
+class TestOpenvswitchSecurityGroups(OpenvswitchSecurityGroupsTestCase,
+                                    test_sg.TestSecurityGroups,
+                                    test_sg_rpc.SGNotificationTestMixin):
+    def test_security_group_get_port_from_device(self):
+        with self.network() as n:
+            with self.subnet(n):
+                with self.security_group() as sg:
+                    security_group_id = sg['security_group']['id']
+                    res = self._create_port(self.fmt, n['network']['id'])
+                    port = self.deserialize(self.fmt, res)
+                    fixed_ips = port['port']['fixed_ips']
+                    data = {'port': {'fixed_ips': fixed_ips,
+                                     'name': port['port']['name'],
+                                     ext_sg.SECURITYGROUPS:
+                                     [security_group_id]}}
+
+                    req = self.new_update_request('ports', data,
+                                                  port['port']['id'])
+                    res = self.deserialize(self.fmt,
+                                           req.get_response(self.api))
+                    port_id = res['port']['id']
+                    plugin = manager.QuantumManager.get_plugin()
+                    port_dict = plugin.callbacks.get_port_from_device(port_id)
+                    self.assertEqual(port_id, port_dict['id'])
+                    self.assertEqual([security_group_id],
+                                     port_dict[ext_sg.SECURITYGROUPS])
+                    self.assertEqual([], port_dict['security_group_rules'])
+                    self.assertEqual([fixed_ips[0]['ip_address']],
+                                     port_dict['fixed_ips'])
+                    self._delete('ports', port_id)
+
+    def test_security_group_get_port_from_device_with_no_port(self):
+        plugin = manager.QuantumManager.get_plugin()
+        port_dict = plugin.callbacks.get_port_from_device('bad_device_id')
+        self.assertEqual(None, port_dict)
+
+
+class TestOpenvswitchSecurityGroupsXML(TestOpenvswitchSecurityGroups):
+    fmt = 'xml'

quantum/tests/unit/test_iptables_firewall.py

@@ -44,7 +44,8 @@ class IptablesFirewallTestCase(unittest.TestCase):
         self.iptables_inst.ipv6 = {'filter': self.v6filter_inst}
         iptables_cls.return_value = self.iptables_inst
 
-        self.firewall = IptablesFirewallDriver(self.iptables_inst)
+        self.firewall = IptablesFirewallDriver()
+        self.firewall.iptables = self.iptables_inst
 
     def tearDown(self):
         self.iptables_cls_p.stop()

quantum/tests/unit/test_security_groups_rpc.py

@@ -28,6 +28,8 @@ from quantum.agent import rpc as agent_rpc
 from quantum.agent import securitygroups_rpc as sg_rpc
 from quantum import context
 from quantum.db import securitygroups_rpc_base as sg_db_rpc
+from quantum.extensions import securitygroup as ext_sg
+from quantum.openstack.common import cfg
 from quantum.openstack.common.rpc import proxy
 from quantum.tests.unit import test_extension_security_group as test_sg
 from quantum.tests.unit import test_iptables_firewall as test_fw
@@ -371,23 +373,23 @@ class SGServerRpcCallBackMixinTestCaseXML(SGServerRpcCallBackMixinTestCase):
 class SGAgentRpcCallBackMixinTestCase(unittest.TestCase):
     def setUp(self):
         self.rpc = sg_rpc.SecurityGroupAgentRpcCallbackMixin()
-        self.rpc.agent = mock.Mock()
+        self.rpc.sg_agent = mock.Mock()
 
     def test_security_groups_rule_updated(self):
         self.rpc.security_groups_rule_updated(None,
                                               security_groups=['fake_sgid'])
-        self.rpc.agent.assert_has_calls(
+        self.rpc.sg_agent.assert_has_calls(
             [call.security_groups_rule_updated(['fake_sgid'])])
 
     def test_security_groups_member_updated(self):
         self.rpc.security_groups_member_updated(None,
                                                 security_groups=['fake_sgid'])
-        self.rpc.agent.assert_has_calls(
+        self.rpc.sg_agent.assert_has_calls(
             [call.security_groups_member_updated(['fake_sgid'])])
 
     def test_security_groups_provider_updated(self):
         self.rpc.security_groups_provider_updated(None)
-        self.rpc.agent.assert_has_calls(
+        self.rpc.sg_agent.assert_has_calls(
             [call.security_groups_provider_updated()])
 
 
@@ -582,16 +584,16 @@ IPTABLES_FILTER_1 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port1 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port1 ! -s 10.0.0.3 -j DROP
@@ -619,17 +621,17 @@ IPTABLES_FILTER_1_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port1 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port1 -j RETURN -s 10.0.0.4
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port1 ! -s 10.0.0.3 -j DROP
@@ -661,17 +663,17 @@ IPTABLES_FILTER_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port1 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port1 -j RETURN -s 10.0.0.4
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port1 ! -s 10.0.0.3 -j DROP
@@ -680,17 +682,17 @@ IPTABLES_FILTER_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-o_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-o_port1 -j RETURN
 -A %(bn)s-o_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port2 -j %(bn)s-i_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-i_port2
 -A %(bn)s-i_port2 -m state --state INVALID -j DROP
 -A %(bn)s-i_port2 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port2 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port2 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port2 -j RETURN -s 10.0.0.3
 -A %(bn)s-i_port2 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
 -A %(bn)s-o_port2 -m mac ! --mac-source 12:34:56:78:9a:bd -j DROP
 -A %(bn)s-o_port2 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port2 ! -s 10.0.0.4 -j DROP
@@ -720,16 +722,16 @@ IPTABLES_FILTER_2_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port1 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port1 ! -s 10.0.0.3 -j DROP
@@ -738,17 +740,17 @@ IPTABLES_FILTER_2_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-o_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-o_port1 -j RETURN
 -A %(bn)s-o_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port2 -j %(bn)s-i_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-i_port2
 -A %(bn)s-i_port2 -m state --state INVALID -j DROP
 -A %(bn)s-i_port2 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port2 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
 -A %(bn)s-i_port2 -j RETURN -p tcp --dport 22
 -A %(bn)s-i_port2 -j RETURN -s 10.0.0.3
 -A %(bn)s-i_port2 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
 -A %(bn)s-o_port2 -m mac ! --mac-source 12:34:56:78:9a:bd -j DROP
 -A %(bn)s-o_port2 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port2 ! -s 10.0.0.4 -j DROP
@@ -778,8 +780,8 @@ IPTABLES_FILTER_2_3 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
@@ -787,9 +789,9 @@ IPTABLES_FILTER_2_3 = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-i_port1 -j RETURN -s 10.0.0.4
 -A %(bn)s-i_port1 -j RETURN -p icmp
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port1 ! -s 10.0.0.3 -j DROP
@@ -798,8 +800,8 @@ IPTABLES_FILTER_2_3 = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-o_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-o_port1 -j RETURN
 -A %(bn)s-o_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port2 -j %(bn)s-i_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-i_port2
 -A %(bn)s-i_port2 -m state --state INVALID -j DROP
 -A %(bn)s-i_port2 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port2 -j RETURN -p udp --dport 68 --sport 67 -s 10.0.0.2
@@ -807,9 +809,9 @@ IPTABLES_FILTER_2_3 = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-i_port2 -j RETURN -s 10.0.0.3
 -A %(bn)s-i_port2 -j RETURN -p icmp
 -A %(bn)s-i_port2 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
 -A %(bn)s-o_port2 -m mac ! --mac-source 12:34:56:78:9a:bd -j DROP
 -A %(bn)s-o_port2 -p udp --sport 68 --dport 67 -j RETURN
 -A %(bn)s-o_port2 ! -s 10.0.0.4 -j DROP
@@ -855,14 +857,14 @@ IPTABLES_FILTER_V6_1 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p icmpv6 -j RETURN
 -A %(bn)s-o_port1 -m state --state INVALID -j DROP
@@ -890,27 +892,27 @@ IPTABLES_FILTER_V6_2 = """:%(bn)s-(%(chains)s) - [0:0]
 -A OUTPUT -j %(bn)s-OUTPUT
 -A FORWARD -j %(bn)s-FORWARD
 -A %(bn)s-sg-fallback -j DROP
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port1 -j %(bn)s-i_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port1 -j %(bn)s-i_port1
 -A %(bn)s-i_port1 -m state --state INVALID -j DROP
 -A %(bn)s-i_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port1 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port1 -j %(bn)s-o_port1
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port1 -j %(bn)s-o_port1
 -A %(bn)s-o_port1 -m mac ! --mac-source 12:34:56:78:9a:bc -j DROP
 -A %(bn)s-o_port1 -p icmpv6 -j RETURN
 -A %(bn)s-o_port1 -m state --state INVALID -j DROP
 -A %(bn)s-o_port1 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-o_port1 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-out tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-out tap_port2 -j %(bn)s-i_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-INGRESS tap_port2 -j %(bn)s-i_port2
 -A %(bn)s-i_port2 -m state --state INVALID -j DROP
 -A %(bn)s-i_port2 -m state --state ESTABLISHED,RELATED -j RETURN
 -A %(bn)s-i_port2 -j %(bn)s-sg-fallback
--A %(bn)s-FORWARD %(physdev)s --physdev-in tap_port2 -j %(bn)s-sg-chain
+-A %(bn)s-FORWARD %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-sg-chain
--A %(bn)s-sg-chain %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-sg-chain %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
--A %(bn)s-INPUT %(physdev)s --physdev-in tap_port2 -j %(bn)s-o_port2
+-A %(bn)s-INPUT %(physdev)s --physdev-EGRESS tap_port2 -j %(bn)s-o_port2
 -A %(bn)s-o_port2 -m mac ! --mac-source 12:34:56:78:9a:bd -j DROP
 -A %(bn)s-o_port2 -p icmpv6 -j RETURN
 -A %(bn)s-o_port2 -m state --state INVALID -j DROP
@@ -936,11 +938,25 @@ IPTABLES_FILTER_V6_EMPTY = """:%(bn)s-(%(chains)s) - [0:0]
 -A %(bn)s-sg-fallback -j DROP
 """ % IPTABLES_ARG
 
+FIREWALL_BASE_PACKAGE = 'quantum.agent.linux.iptables_firewall.'
+
 
 class TestSecurityGroupAgentWithIptables(unittest.TestCase):
+    FIREWALL_DRIVER = FIREWALL_BASE_PACKAGE + 'IptablesFirewallDriver'
+    PHYSDEV_INGRESS = 'physdev-out'
+    PHYSDEV_EGRESS = 'physdev-in'
+
     def setUp(self):
         self.mox = mox.Mox()
+        agent_opts = [
+            cfg.StrOpt('root_helper', default='sudo'),
+        ]
 
+        cfg.CONF.register_opts(agent_opts, "AGENT")
+        cfg.CONF.set_override(
+            'firewall_driver',
+            self.FIREWALL_DRIVER,
+            group='SECURITYGROUP')
         self.addCleanup(mock.patch.stopall)
         self.addCleanup(self.mox.UnsetStubs)
 
@@ -1018,6 +1034,8 @@ class TestSecurityGroupAgentWithIptables(unittest.TestCase):
                     'security_group1']}
 
     def _regex(self, value):
+        value = value.replace('physdev-INGRESS', self.PHYSDEV_INGRESS)
+        value = value.replace('physdev-EGRESS', self.PHYSDEV_EGRESS)
         value = value.replace('\n', '\\n')
         value = value.replace('[', '\[')
         value = value.replace(']', '\]')
@@ -1093,3 +1111,71 @@ class TestSecurityGroupAgentWithIptables(unittest.TestCase):
         self.agent.security_groups_rule_updated(['security_group1'])
 
         self.mox.VerifyAll()
+
+
+class SGNotificationTestMixin():
+    def test_security_group_rule_updated(self):
+        name = 'webservers'
+        description = 'my webservers'
+        with self.security_group(name, description) as sg:
+            with self.security_group(name, description) as sg2:
+                security_group_id = sg['security_group']['id']
+                direction = "ingress"
+                source_group_id = sg2['security_group']['id']
+                protocol = 'tcp'
+                port_range_min = 88
+                port_range_max = 88
+                with self.security_group_rule(security_group_id, direction,
+                                              protocol, port_range_min,
+                                              port_range_max,
+                                              source_group_id=source_group_id
+                                              ):
+                    pass
+            self.notifier.assert_has_calls(
+                [call.security_groups_rule_updated(mock.ANY,
+                                                   [security_group_id]),
+                 call.security_groups_rule_updated(mock.ANY,
+                                                   [security_group_id])])
+
+    def test_security_group_member_updated(self):
+        with self.network() as n:
+            with self.subnet(n):
+                with self.security_group() as sg:
+                    security_group_id = sg['security_group']['id']
+                    res = self._create_port(self.fmt, n['network']['id'])
+                    port = self.deserialize(self.fmt, res)
+
+                    data = {'port': {'fixed_ips': port['port']['fixed_ips'],
+                                     'name': port['port']['name'],
+                                     ext_sg.SECURITYGROUPS:
+                                     [security_group_id]}}
+
+                    req = self.new_update_request('ports', data,
+                                                  port['port']['id'])
+                    res = self.deserialize(self.fmt,
+                                           req.get_response(self.api))
+                    self.assertEquals(res['port'][ext_sg.SECURITYGROUPS][0],
+                                      security_group_id)
+                    self._delete('ports', port['port']['id'])
+                    self.notifier.assert_has_calls(
+                        [call.security_groups_member_updated(
+                            mock.ANY, [mock.ANY]),
+                         call.security_groups_member_updated(
+                             mock.ANY, [security_group_id])])
+
+
+class TestSecurityGroupAgentWithOVSIptables(
+        TestSecurityGroupAgentWithIptables):
+
+    FIREWALL_DRIVER = FIREWALL_BASE_PACKAGE + 'OVSHybridIptablesFirewallDriver'
+
+    def _regex(self, value):
+        #Note(nati): tap is prefixed on the device
+        # in the OVSHybridIptablesFirewallDriver
+
+        value = value.replace('tap_port', 'taptap_port')
+        value = value.replace('o_port', 'otap_port')
+        value = value.replace('i_port', 'itap_port')
+        return super(
+            TestSecurityGroupAgentWithOVSIptables,
+            self)._regex(value)